Spring 集成和 RabbitMQ：接收消息时没有 output-channel 或 replyChannel header 可用

Question

我写了一个带有请求和回复的简单消息流。我必须使用两个独立的 queues 所以我声明 AmqpOutboundAdapter 发送消息和 AmqpInboundAdapter 接收回复。

@Bean
@FindADUsers
public AmqpOutboundEndpoint newFindADUsersOutboundAdapter() {
    return Amqp.outboundAdapter(amqpTemplate())
            .routingKeyExpression("headers[" + ADUsersFindConfig.ROUTING_KEY_HEADER + "]")
            .exchangeName(getExchange())
            .headerMapper(amqpHeaderMapper())
            .get();
}

@Bean
public AmqpInboundChannelAdapter newFindADUsersResponseInboundChannelAdapter(
        ADUsersFindResponseConfig config) {
    return Amqp.inboundAdapter(rabbitConnectionFactory(), findADUsersResponseQueue)
            .headerMapper(amqpHeaderMapper())
            .outputChannel(config.newADUsersFindResponseOutputChannel())
            .get();
}

它应该与@MessagingGateway 一起工作：

@MessagingGateway
public interface ADUsersFindService {

     String FIND_AD_USERS_CHANNEL = "adUsersFindChannel";

     String FIND_AD_USERS_REPLY_OUTPUT_CHANNEL = "adUsersFindReplyOutputChannel";

     String FIND_AD_USERS_REPLY_CHANNEL = "adUsersFindReplyChannel";

     String CORRELATION_ID_REQUEST_HEADER = "correlation_id";

     String ROUTING_KEY_HEADER = "replyRoutingKey";

     String OBJECT_TYPE_HEADER = "object.type";

     @Gateway(requestChannel = FIND_AD_USERS_CHANNEL, replyChannel = FIND_AD_USERS_REPLY_CHANNEL)
ADResponse find(ADRequest adRequest, @Header(ROUTING_KEY_HEADER) String routingKey, @Header(OBJECT_TYPE_HEADER) String objectType);
}

ADUsersFindResponseConfig class 看起来像：

 @Configuration
 @Import(JsonConfig.class)
 public class ADUsersFindResponseConfig {

     @Autowired
     public NullChannel nullChannel;

     @Autowired
     private JsonObjectMapper<?, ?> mapper;

     /**
      * @return The output channel for the flow
      */
     @Bean(name = ADUsersFindService.FIND_AD_USERS_REPLY_OUTPUT_CHANNEL)
     public MessageChannel newADUsersFindResponseOutputChannel() {
         return MessageChannels.direct().get();
     }

     /**
      * @return The output channel for gateway
      */
     @Bean(name = ADUsersFindService.FIND_AD_USERS_REPLY_CHANNEL)
     public MessageChannel newADUsersFindResponseChannel() {
         return MessageChannels.direct().get();
     }

     @Bean
     public IntegrationFlow findADUsersResponseFlow() {
         return IntegrationFlows
                 .from(newADUsersFindResponseOutputChannel())
                 .transform(new JsonToObjectTransformer(ADResponse.class, mapper))
                 .channel(newADUsersFindResponseChannel())
                 .get();
     }
 }

发送消息正常，但接收消息有问题。我希望收到的消息将传递到名为 FIND_AD_USERS_REPLY_OUTPUT_CHANNEL 的通道，然后使用 findADUsersResponseFlow 将消息反序列化为 ADResponse object，然后将下一个 ADResponse object 传递到网关 replyChannel - FIND_AD_USERS_REPLY_CHANNEL。最后，'find'方法return这个object。不幸的是，当 org.springframework.integration.handler.BridgeHandler 收到一条消息时，我遇到了异常：

 org.springframework.messaging.MessagingException: ; nested exception is org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available

消息日志如下所示：

 11:51:35.697 [SimpleAsyncTaskExecutor-1] INFO  New message - GenericMessage [payload={...somepayload...}, headers={correlation_id=7cbd958e-4b09-4e4c-ba8e-5ba574f3309a, replyRoutingKey=findADUsersResponse.ad, amqp_consumerQueue=findADUsersResponseQueue, history=newFindADUsersResponseInboundChannelAdapter,adUsersFindReplyOutputChannel,adUsersFindReplyChannel,infoLog,infoLoggerChain.channel#0,infoLoggerChain.channel#1, id=37a4735d-6983-d1ad-e0a1-b37dc17e48ef, amqp_consumerTag=amq.ctag-8Qs5YEun1jXYRf85Hu1URA, object.type=USER, timestamp=1469094695697}]

所以我很确定消息已传递给 adUsersFindReplyChannel。此外（如果重要的话）请求消息和回复消息都将 'replyTo' header 设置为空。我做错了什么？

Answer 1

replyChannel header 是实时的 object，不能通过 AMQP 序列化。

您可以使用出站网关而不是一对适配器，框架将处理 headers。

如果出于某种原因必须使用适配器，则需要做两件事：

使用 header channel registry 将通道 object 转换为在注册表中注册的字符串。
确保 header 映射器配置为 send/receive replyChannel header 并且您的接收系统 returns header在回复中。

Spring 集成和 RabbitMQ：接收消息时没有 output-channel 或 replyChannel header 可用

Spring Integration and RabbitMQ: no output-channel or replyChannel header available while receiving message

java

spring

amqp

spring-integration

rabbitmq