spring - 使用 Spring amqp 库的请求-响应模式

Question

每个人。我有一个用于在 RabbitMQ 代理中发布消息的 HTTP API，我需要实现请求-响应模式才能接收来自服务器的响应。所以我就像是客户端和服务器之间的桥梁。我将消息推送到具有特定路由 key 的代理，并且该消息有一个消费者，它将消息作为响应发布回，我的 API 必须为每个请求使用响应。所以图表是这样的:



所以我要做的是以下 - 对于每个 HTTP session ，我创建一个临时 responseQueue(绑定(bind)到默认交换，路由键是该队列的名称)，之后我将消息的 replyTo header 设置为名称响应队列(我将在其中等待响应)并将模板replyQueue设置为该队列。这是我的代码:

public void sendMessage(AbstractEvent objectToSend, final String routingKey) {
    final Queue responseQueue = rabbitAdmin.declareQueue();
    byte[] messageAsBytes = null;
    try {
        messageAsBytes = new ObjectMapper().writeValueAsBytes(objectToSend);
    } catch (JsonProcessingException e) {
        e.printStackTrace();
    }
    MessageProperties properties = new MessageProperties();
    properties.setHeader("ContentType", MessageBodyFormat.JSON);
    properties.setReplyTo(responseQueue.getName());
    requestTemplate.setReplyQueue(responseQueue);

    Message message = new Message(messageAsBytes, properties);
    Message receivedMessage = (Message)requestTemplate.convertSendAndReceive(routingKey, message);
}

那么问题出在哪里:消息被发送，之后被消费者消费，其响应被正确发送到正确的队列，但由于某种原因，它没有在 convertSendAndReceived 方法中取回，并且在设置超时后我的 receivedMessage一片空白。所以我尝试做几件事——我开始检查 spring 代码(顺便说一句，这样做是一场真正的噩梦)并看到我没有声明它为我创建一个临时的响应队列，以及 replyTo header 设置为队列的名称(与我所做的相同)。结果是一样的——receivedMessage 仍然为空。之后我决定使用另一个使用默认交换的模板，因为 responseQueue 绑定(bind)到该交换:

requestTemplate.send(routingKey, message);
Message receivedMessage = receivingTemplate.receive(responseQueue.getName());

结果是一样的——responseMessage 仍然为空。
amqp 和 rabbit 的版本分别是 1.2.1 和 1.2.0。所以我确定我错过了一些东西，但我不知道它是什么，所以如果有人可以帮助我，我将非常感激。

Answer 1

1> 奇怪的是RabbitTemplate使用 doSendAndReceiveWithFixed如果您提供 requestTemplate.setReplyQueue(responseQueue) .看起来你的解释是错误的。

2> 使其与固定 ReplyQueue 一起使用你应该配置一个reply ListenerContainer :

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory);
container.setQueues(responseQueue);
container.setMessageListener(requestTemplate);

3> 但这里最重要的部分是 correlation . RabbitTemplate.sendAndReceive填充 correlationId message 属性，但消费者也必须处理它:仅仅发送回复到 responseQueue 是不够的，回复消息应该有相同的 correlationId属性(property)。见这里: how to send response from consumer to producer to the particular request using Spring AMQP?

顺便说一句，没有理由填充 Message手动:您可以简单地支持 Jackson2JsonMessageConverter到 RabbitTemplate它会转换你的 objectToSend使用适当的 header 自动发送到 JSON 字节。