Spring Boot使用AsyncRabbitTemplate::sendAndReceive出现Reply timeout问题

AsyncRabbitTemplate.sendAndReceive() 是RPC请求-响应模式的API，和你之前使用的普通单向send方法逻辑完全不同：调用该接口后框架会一直等待消费者端返回对应应答消息，超时未收到就会抛出你遇到的AmqpReplyTimeoutException，你调整超时参数只是改变等待时长，本质没解决无应答返回的问题。

• 先确认业务场景是否真的需要RPC模式：如果不需要消费者返回处理结果，直接换回AsyncRabbitTemplate.send()方法即可，不需要等待应答自然不会触发超时。
• 如果确实需要RPC请求-响应能力，按以下顺序排查：
  • 检查消费者逻辑是否正确处理应答返回：RabbitMQ RPC模式要求消费者收到消息后，必须把应答消息发送到请求消息Properties里的replyTo队列，同时要带上原请求的correlationId做匹配，否则发送端识别不到对应应答。
  • 检查路由绑定是否正确：确认你的primary交换机和1.1.1.csv-routing-key绑定的队列确实有消费者在监听消费，无消息堆积。
  • 检查消费者处理耗时：如果消费者逻辑处理时间超过你设置的10s超时，也会触发该异常，可适当调大replyTimeout参数同时优化消费者处理速度。
  • 检查RabbitMQ版本兼容性：Direct Reply To功能默认需要RabbitMQ 3.4以上版本支持，若你使用的是更低版本，需要手动开启该功能，或者切换为自定义固定回复队列模式测试。

场景1：仅需要异步发送消息，不需要接收消费者响应 #

直接把sendAndReceive替换为send方法即可，修改调用逻辑：

// 替换原sendAndReceive调用
asyncRabbitTemplate.send(
    "1.1.1.csv-routing-key",
    new Message(
        "a,test,csv".getBytes(StandardCharsets.UTF_8),
        props.build()
    )
);
// 如果需要确认消息是否成功到达Broker，开启publisher confirm机制即可，不需要等待消费者应答

场景2：确实需要RPC响应模式 #

消费者侧补充应答返回逻辑即可，Spring Boot集成场景下框架会自动处理correlationId匹配和replyTo队列发送：

// 消费者侧示例
@RabbitListener(queues = "你的业务队列名称")
public Message consume(Message requestMsg) {
    // 执行业务处理逻辑
    String handleResult = "处理完成";
    // 构造返回的应答消息
    return MessageBuilder.withBody(handleResult.getBytes(StandardCharsets.UTF_8))
            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
            .build();
}

如果是原生手动编写消费者代码，需要显式获取请求的replyTo和correlationId参数：

// 原生手动返回应答示例
String replyTo = requestMsg.getMessageProperties().getReplyTo();
String correlationId = requestMsg.getMessageProperties().getCorrelationId();
// 构造应答消息
Message replyMsg = MessageBuilder.withBody("处理成功".getBytes(StandardCharsets.UTF_8))
        .setCorrelationId(correlationId)
        .build();
// 发送应答到指定replyTo队列
rabbitTemplate.send("", replyTo, replyMsg);

内容的提问来源于stack exchange，提问作者Thomas Turner