axon 3 通过 rabbitmq 分离命令和查询服务：json vs. xml

Question

我正在研究一个场景，其中命令和查询服务是独立的节点（弹簧启动）。 我将 RabbitTemplate 配置为使用 Jackson 进行 JSON 序列化：

@Bean
RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory, Jackson2JsonMessageConverter converter) {
  final RabbitTemplate rabbitTemplate = new  RabbitTemplate(connectionFactory);
  rabbitTemplate.setMessageConverter(converter);
  return rabbitTemplate;
}

@Bean
Jackson2JsonMessageConverter producerJackson2MessageConverter() {
  return new Jackson2JsonMessageConverter();
}

这行得通，当我将命令发送到命令服务时，我可以在 RabbitMQ 队列中检查正确的 JSON 格式。

My Aggregate 接收命令并触发事件，这些已正确存储在 mysql-eventstore 中。这里，事件是 xml 格式的。

现在是棘手的部分。我使用 axon-springboot 通过属性再次将事件发送到兔子队列：

axon.amqp.exchange: my-event-exchange

再次，这有效，我可以检查队列并查看事件（以 xml 格式）。

在队列服务上，我想接收事件并在内存中生成 h2 表示...我使用：

@Bean
public SpringAMQPMessageSource myMessageSource(Serializer serializer, EventHandlingConfiguration ehConfig) {
   return new SpringAMQPMessageSource(serializer) {
      @RabbitListener(queues = "my-event-queue")
      @Override
      public void onMessage(Message message, Channel channel) throws Exception {
           log.info("receiving event: {}#{}", message, channel);
           super.onMessage(message, channel);
      }
    };
   }

   @Autowired
   public void configure(EventHandlingConfiguration ehConfig, SpringAMQPMessageSource myMessageSource) {
     ehConfig.registerSubscribingEventProcessor(
          "taskQueryObjectUpdater", c -> myMessageSource);
   }

如文档中所述。

但是，当我看到“接收事件”日志时，我得到了一个

2017-06-16 15:17:38.168  WARN 69212 --- [cTaskExecutor-1] o.s.a.s.c.Jackson2JsonMessageConverter   : Could not convert incoming message with content-type [null]

警告，我的 h2-repository-service 永远不会被调用。

我担心我为 rabbit 配置的 json 转换器和 axon 中的 xml 事件处理会干扰......还是我走错了路？如何通过rabbit正确分离命令和查询服务，eventProcessor名称有什么用？我使用了处理事件流的spring bean的bean名称......

Answer 1

您的配置中的两种方法似乎相互冲突。自动配置使用缺少特定类型的 bean 来定义默认值。但是，如果您定义了预期类型的​​ bean（即使是用于不同目的），则使用该 bean。

Axon 使用 Spring 的 AMQP AutoConfiguration 来读取消息。如果您指定要以特定方式转换消息，那么 Axon 也会间接使用它。

解决方法可能是不将 Jackson2JsonMessageConverter 定义为 bean。我怀疑 MessageListenerContainer（Spring 用来调用 @RabbitListener 方法）是否使用 RabbitTemplate，但我几乎可以肯定它会使用您的应用程序上下文中定义的任何 MessageConverter。

或者，您可以定义从 AMQP 显式读取消息所需的组件，而不是依赖自动配置。在这种情况下，您可以配置一个（额外的）SimpleMessageListenerContainerFactory，并将该容器分配给您的 @RabbitListener(containerFactory="") 注释。

希望这会有所帮助。