spring amqp-outbound gateway 生成来自不同 thead 的回复（如 jms-outbound gateway）

问题陈述：

Spring amqp-outbound gateway 从不同线程生成回复（像 jms-outbound gateway 一样，具有不同的队列，使用相关键关联请求/响应）。

无法将该消息与此示例关联起来。

弹簧集成

    <int:gateway id="outboundGateway" service-interface="com.amqp.outbound.gateway.OutboundGateway"     
                        default-reply-channel="defaultReplyChannel" >
        <int:method name="process"   request-channel="inboundRequestChannel"/>
    </int:gateway>

    <int:channel id="defaultReplyChannel"/>
    <int:channel id="inboundRequestChannel"/>
    <int:channel id="enrichedInboundRequestChannel"/>
    <int:channel id="processAuthRequestChannel"/>
    <int:channel id="postProcessorChannel"/>

    <int:chain input-channel="inboundRequestChannel" output-channel="enrichedInboundRequestChannel">
        <int:service-activator id="serviceActivator"
                       ref="ouboundService"  method="createRequest"/>
    </int:chain>

    <int-amqp:outbound-gateway id="outboundGtwyId" header-mapper="headerMapper"
                        request-channel="enrichedInboundRequestChannel"
                        reply-channel="defaultReplyChannel"
                        amqp-template="template" 
                        reply-timeout="30000" 
                        exchange-name="request_exchange" 
                        routing-key="request_exchange_queue"/>

    <int-amqp:inbound-channel-adapter id="amqpMessageDriven"  queue-names="request_queue" 
                                 connection-factory="rabbitConnectionFactory"  channel="processAuthRequestChannel"/>

    <int:service-activator id="serviceActivator"
                       ref="ouboundService" input-channel="processAuthRequestChannel" output-channel="postProcessorChannel"
                       method="processRequest"/>

    <int-amqp:outbound-channel-adapter amqp-template="template" channel="postProcessorChannel" 
            header-mapper="headerMapper" exchange-name="reply_exchange" routing-key="reply_exchange_queue"/>

    <bean id="headerMapper" class="org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper"/>

Config

@Bean
public RabbitTemplate template(ConnectionFactory rabbitConnectionFactory){
    final RabbitTemplate template = new RabbitTemplate(rabbitConnectionFactory);
    template.setQueue("reply_queue");
    return template;
}



@Bean
public Binding binding(){
    return BindingBuilder.bind(this.queue()).to(this.exchange()).with("request_exchange_queue");
}

@Bean
public DirectExchange exchange(){
    return new DirectExchange("request_exchange");
}

@Bean
public Queue queue(){
    return new Queue("request_queue", true, false, true);
}

@Bean
public Binding bindingReply(){
    return BindingBuilder.bind(this.queue()).to(this.exchange()).with("reply_exchange_queue");
}

@Bean
public DirectExchange exchangeReply(){
    return new DirectExchange("reply_exchange");
}


@Bean
public Queue replyQueue(){
    return new Queue("reply_queue", true, false, true);
}

Service

@Service
public final class OuboundService {


    public Message createRequest(String message){
        System.out.println("Inside createRequest : "+ message);
        final String transactionId = UUID.randomUUID().toString();
        final Message builtMessage = MessageBuilder.withBody(message.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                .setHeader(AmqpHeaders.CORRELATION_ID, transactionId)
                .build();
        return builtMessage;
    }


    public Message processRequest(Message message){
        System.out.println("Inside process Request : "+ new String(message.getBody()));
        System.out.println("Header values : "+message.getMessageProperties().getHeaders());
        final Message result = MessageBuilder.withBody("Successful".getBytes()).copyProperties(message.getMessageProperties())
                                .copyHeaders(message.getMessageProperties().getHeaders()).build();
        return result;
    }

}

Error:

org.springframework.integration.handler.ReplyRequiredException：处理程序“outboundGtwyId”没有生成回复，并且其“requiresReply”属性设置为true。

GitHub 源代码（已解决的解决方案）

https://github.com/kingkongprab/spring-amqp-outbound-gateway https://github.com/kingkongprab/spring-amqp-outbound-gateway

关联也在 Spring AMQP 中完成。见其RabbitTemplate#sendAndRecevie()了解更多信息。另外，在参考手册 https://docs.spring.io/spring-amqp/docs/2.0.0.RELEASE/reference/html/_reference.html#request-reply.

Spring 集成及其AbstractAmqpOutboundEndpoint and AmqpInboundGateway实现提供了开箱即用的请求-答复关联解决方案。如果您无法使用AmqpInboundGateway在服务器端，您应该确保correlationId从收到的请求转移到要发回的回复。是的，您可以使用专用交换来回复，这也是RabbitTemplate#setQueue()等待客户端、出站端的回复。但如果没有适当的方法，这仍然行不通correlation转移。另请参阅https://docs.spring.io/spring-integration/docs/4.3.12.RELEASE/reference/html/amqp.html#amqp-message-headers https://docs.spring.io/spring-integration/docs/4.3.12.RELEASE/reference/html/amqp.html#amqp-message-headers有关如何标头的信息（包括correlationId）在 Spring Integration 中映射。

UPDATE

感谢您分享您的申请。

好吧，现在我看到了几个问题：

1. 你肯定错过了replyQueue捆绑：

   @Bean
   public Binding bindingReply(){
       return BindingBuilder.bind(this.replyQueue()).to(this.exchangeReply()).with("reply_exchange_queue");
   }
   

2. RabbitTemplate必须使用setReplyAddress()。你必须配置MessageListenerContainer为了reply_queue并有RabbitTemplate作为听众：

   @Bean
   public RabbitTemplate template(ConnectionFactory rabbitConnectionFactory){
       final RabbitTemplate template = new RabbitTemplate(rabbitConnectionFactory);
       template.setReplyAddress(replyQueue().getName());
       return template;
   }
   
   @Bean
   public MessageListenerContainer replyContainer(RabbitTemplate template) {
       SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
       container.setConnectionFactory(template.getConnectionFactory());
       container.setQueues(replyQueue());
       container.setMessageListener(template);
       return container;
   }
   

3. Your OuboundService with org.springframework.amqp.core.Message操纵是没有用的。通道适配器不知道这种类型payload和你的定制Message只是成为序列化的body另一个的org.springframework.amqp.core.Message。我已将其更改为这样，一切正常：

   public String createRequest(String message){
       System.out.println("Inside createRequest : "+ message);
       return message;
   }
   
   
   public Message processRequest(Message message){
       System.out.println("Inside process Request : " + message);
       return message;
   }
   

无论如何，我建议你重新考虑你的设计并回到AmqpInboundGateway.

顺便说一句，在最终的解决方案中你不需要关心任何correlation。框架会自动为您完成此操作。