Spring AMQP中sendAndReceive多RPC请求回复队列冲突问题

这个问题我之前也遇到过，核心原因是默认情况下RabbitTemplate的sendAndReceive会共享一个临时回复队列（默认名为.replies），当多个RPC请求复用这个队列时，很容易出现回复与请求不匹配的情况。你完全不需要为每个操作创建独立交换机，现有单交换机+多路由键的配置是合理的，下面给你两种可行的解决方案：

方案1：为每个RPC操作配置独立的RabbitTemplate（推荐） #

给products.get和products.stock.update分别创建专用的RabbitTemplate实例，每个模板绑定自己的专属回复队列，从根源上隔离两个操作的回复流，彻底避免混淆。

生产者配置修改：

@Bean
public DirectExchange productsExchange() {
    return new DirectExchange("products");
}

// 为"获取商品"操作配置专用RabbitTemplate
@Bean
public RabbitTemplate getProductRabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMessageConverter(producerJackson2MessageConverter());
    // 设置专属回复队列
    Queue getReplyQueue = new Queue("products.get.replies");
    template.setReplyQueue(getReplyQueue);
    // 设置回复超时时间（按需调整）
    template.setReplyTimeout(10000);
    return template;
}

// 为"库存更新"操作配置专用RabbitTemplate
@Bean
public RabbitTemplate updateStockRabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMessageConverter(producerJackson2MessageConverter());
    Queue updateReplyQueue = new Queue("products.stock.update.replies");
    template.setReplyQueue(updateReplyQueue);
    template.setReplyTimeout(10000);
    return template;
}

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

发送逻辑修改：

// 使用"获取商品"专用模板发送请求
Message getResponse = getProductRabbitTemplate.sendAndReceive(productsExchange.getName(), "products.get", getRequestMessage);

// 使用"库存更新"专用模板发送请求
Message updateResponse = updateStockRabbitTemplate.sendAndReceive(productsExchange.getName(), "products.stock.update", updateRequestMessage);

消费者端无需额外修改，Spring AMQP会自动将回复发送到请求消息中replyTo字段指定的专属队列，两个操作的回复完全隔离。

方案2：确保回复消息携带正确的CorrelationId #

如果不想创建多个RabbitTemplate，可以通过维护请求与回复的correlationId关联，让RabbitTemplate准确匹配对应的回复。

消费者监听方法修改：

@RabbitListener(queues = "getProductBySku")
public Message getProduct(GetProductResource request, Message requestMessage) {
    // 处理业务逻辑，生成响应数据
    GetProductResponse response = ...;

    // 构建回复消息，复制请求的correlationId到回复
    MessageProperties replyProps = new MessageProperties();
    replyProps.setCorrelationId(requestMessage.getMessageProperties().getCorrelationId());
    replyProps.setContentType(MessageProperties.CONTENT_TYPE_JSON);
    
    return new Message(objectMapper.writeValueAsBytes(response), replyProps);
}

@RabbitListener(queues = "updateProductStock")
public Message updateStock(UpdateStockResource request, Message requestMessage) {
    // 处理业务逻辑，生成响应数据
    UpdateStockResponse response = ...;

    MessageProperties replyProps = new MessageProperties();
    replyProps.setCorrelationId(requestMessage.getMessageProperties().getCorrelationId());
    replyProps.setContentType(MessageProperties.CONTENT_TYPE_JSON);
    
    return new Message(objectMapper.writeValueAsBytes(response), replyProps);
}

这种方式下，即使共享默认的.replies队列，RabbitTemplate也会根据correlationId只接收属于当前请求的回复。不过高并发场景下，共享队列可能成为性能瓶颈，还是推荐方案1。

内容的提问来源于stack exchange，提问作者Piotr Piotrowski