Spring集成- AMQP支持消息通道和消息转换

Question

我正在我的Spring应用程序中尝试使用AMQP支持的消息通道，但我认为我从根本上误解了一些东西，特别是围绕Message<?>接口以及GenericMessage<?>实例是如何写入和读取RabbitMQ队列的。

假设我有一个包含以下域模型对象的应用程序：

@Immutable
class Foo {
  String name
  long quantity
}

我声明一个名为fooChannel的AMQP支持消息通道如下：

@Bean
public AmqpChannelFactoryBean deliveryPlacementChannel(CachingConnectionFactory connectionFactory) {
  AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true)
  factoryBean.setConnectionFactory(connectionFactory)
  factoryBean.setQueueName("foo")
  factoryBean.beanName = 'fooChannel'
  factoryBean.setPubSub(false)
  factoryBean
}

当我最初试图向我的fooChannel发送消息时，我收到了一个java.io.NotSerializableException。我意识到这是因为我的AMQP支持的fooChannel所使用的fooChannel使用的org.springframework.amqp.support.converter.SimpleMessageConverter只能使用字符串、可序列化实例或字节数组，而fooChannel的Foo模型并不是这些东西中的任何一个。

因此，我认为应该使用org.springframework.amqp.support.converter.Jackson2JsonMessageConverter来确保正确地将Foo模型转换为/from和AMQP消息。但是，添加到支持MyfooChannel的RabbitMQ队列中的消息的类型似乎是org.springframework.messaging.support.GenericMessage类型。这意味着当我的AMQP支持的fooChannel尝试使用来自RabbitMQ队列的消息时，它会收到以下异常：

Caused by: com.fasterxml.jackson.databind.JsonMappingException: No suitable constructor found for type [simple type, class org.springframework.messaging.support.GenericMessage]: can not instantiate from JSON object (missing default constructor or creator, or perhaps need to add/enable type information?)

通过查看GenericMessage类，我发现它设计为不可变，这清楚地解释了为什么Jackson2JsonMessageConverter不能从JSON转换为GenericMessage类型。但是，我不确定我应该做什么才能让我的fooChannel得到Foo的支持，并使包含我的Foo模型的Spring消息的转换工作正常？

在我的应用程序流程中，我有以下Transformer组件，它使用来自(非AMQP支持的) barChannel的Bar模型，并将Foo模型放置在fooChannel上，如下所示：

@Transformer(inputChannel = 'barChannel', outputChannel = 'fooChannel')
public Foo transform(Bar bar) {
  //transform logic removed for brevity
  new Foo(name: 'Foo1', quantity: 1)
}

然后，我有一个ServiceActivator组件，我希望从fooChannel中使用它，如下所示：

@ServiceActivator(inputChannel = 'fooChannel')
void consumeFoos(Foo foo){
  // Do something with foo
} 

我正在使用spring-integration-core:4.2.5.RELEASE和spring-integration-amqp:4.2.5.RELEASE。

有人能告诉我Spring应用程序的配置哪里出了问题吗？

如果需要进一步的资料，以便更好地澄清我的问题或问题，请告诉我。谢谢

Answer 1

是的-amqp支持的通道目前仅限于Java可序列化对象.

我们应该提供一个选项来将Message<?>映射到Spring Message (就像通道适配器那样)，而不是.

this.amqpTemplate.convertAndSend(this.getExchangeName(), this.getRoutingKey(), message);

...which转换整个消息。

您可以使用一对通道适配器(出站/入站)来代替通道。

由于您使用的是Java，所以可以将适配器对封装在一个新的MessageChannel实现中。

我打开了一个JIRA问题。