spring 6 移除了已废弃的 amqp remoting 模块，推荐使用 rabbittemplate 的 sendandreceive 方法配合 @rabbitlistener 实现轻量、可控的 rpc 风格通信。
spring 6 移除了已废弃的 amqp remoting 模块，推荐使用 rabbittemplate 的 sendandreceive 方法配合 @rabbitlistener 实现轻量、可控的 rpc 风格通信。
在 Spring Framework 6 及 Spring AMQP 3.x 中，原有的基于 AmqpInvokerServiceExporter 和 AmqpInvokerProxyFactoryBean 的 AMQP Remoting 功能已被彻底移除——官方明确指出该模块长期处于维护停滞状态，且与现代消息语义（如显式路由、错误隔离、超时控制）存在设计冲突。取而代之的是更灵活、更符合 AMQP 原生模型的显式 RPC 模式：客户端主动发送请求并等待响应，服务端通过监听器处理请求并返回结果。
✅ 推荐实践：基于 RabbitTemplate + @RabbitListener 的 RPC 流程
1. 客户端（调用方）：使用 RabbitTemplate.sendAndReceive()
@Service
public class OrderServiceClient {
private final RabbitTemplate rabbitTemplate;
public OrderServiceClient(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
// 确保配置 reply-to 和默认超时（单位：毫秒）
rabbitTemplate.setReplyTimeout(5000);
rabbitTemplate.setReplyAddress("amq.rabbitmq.reply-to"); // 使用 broker 内置临时队列
}
public OrderResult placeOrder(OrderRequest request) {
Message message = MessageBuilder.withBody(JsonUtils.toJson(request).getBytes(StandardCharsets.UTF_8))
.setContentType(MessageProperties.CONTENT_TYPE_JSON)
.build();
Message response = rabbitTemplate.sendAndReceive("order.exchange", "order.routing.key", message);
if (response == null) {
throw new RuntimeException("RPC timeout or no response received");
}
return JsonUtils.fromJson(new String(response.getBody(), StandardCharsets.UTF_8), OrderResult.class);
}
}
⚠️ 注意：sendAndReceive() 默认依赖 RabbitMQ 的 reply-to 机制（无需手动声明 reply queue），要求服务端正确设置 MessageProperties.setReplyTo() 或直接使用 amq.rabbitmq.reply-to。
2. 服务端（被调用方）：使用 @RabbitListener 处理请求并返回响应
@Component
public class OrderRpcEndpoint {
@RabbitListener(queues = "order.rpc.queue")
public Message handleOrderRequest(Message requestMsg) {
try {
String body = new String(requestMsg.getBody(), StandardCharsets.UTF_8);
OrderRequest request = JsonUtils.fromJson(body, OrderRequest.class);
OrderResult result = processOrder(request); // 业务逻辑
byte[] responseBytes = JsonUtils.toJson(result).getBytes(StandardCharsets.UTF_8);
return MessageBuilder.withBody(responseBytes)
.setContentType(MessageProperties.CONTENT_TYPE_JSON)
.setCorrelationId(requestMsg.getMessageProperties().getCorrelationId())
.build();
} catch (Exception e) {
// 必须返回错误响应，避免客户端无限等待
return MessageBuilder.withBody("{\"error\":\"Internal server error\"}".getBytes())
.setContentType(MessageProperties.CONTENT_TYPE_JSON)
.setCorrelationId(requestMsg.getMessageProperties().getCorrelationId())
.build();
}
}
private OrderResult processOrder(OrderRequest req) {
return new OrderResult("ORDER_" + System.currentTimeMillis(), "ACCEPTED");
}
}
✅ 关键点：
- 服务端必须将 correlationId 从请求复制到响应，确保客户端能正确匹配响应；
- 建议启用 spring.rabbitmq.listener.simple.acknowledge-mode=manual 并在处理成功后手动 ack，提升可靠性；
- 生产环境应配置死信队列（DLX）捕获异常请求，便于监控与重试。
? 补充建议与最佳实践
- 不推荐复用旧 Remoting 抽象层：试图迁移 AmqpInvokerProxyFactoryBean 将导致兼容性问题和不可维护代码；
- 考虑异步优先：若非强实时要求，建议改用事件驱动架构（如发布 OrderPlacedEvent），解耦更彻底；
- 增强可观测性：为每个 RPC 调用添加 MDC 日志（含 correlationId）、Micrometer 计时器及 Spring Sleuth 链路追踪支持；
- 超时与重试策略：RabbitTemplate 支持 setReplyTimeout()，但重试应由业务层控制（例如结合 RetryTemplate），避免消息重复消费引发幂等问题。
综上，Spring 6 下 AMQP RPC 的演进不是功能退化，而是回归消息中间件本质——以显式、可调试、可监控的方式组织请求/响应交互。掌握 sendAndReceive 与 @RabbitListener 的协同模式，是构建健壮分布式通信能力的关键一步。
