Spring 6移除AMQP remoting后，Spring AMQP+RabbitMQ RPC调用替代方案是什么？

Spring 6移除了AMQP remoting的封装，但完全可以基于Spring AMQP核心API手动实现RPC逻辑，以下是几种可行的替代方案：

1. 手动实现Request/Reply模式（最灵活） #

通过手动管理correlationId、replyTo队列实现RPC调用，完全可控整个流程。

客户端实现 #

生成唯一correlationId，指定回复队列（支持临时队列），发送请求后监听回复队列获取响应：

@Service
public class RpcClient {
    private final RabbitTemplate rabbitTemplate;
    private final Map<String, CompletableFuture<Object>> responseFutures = new ConcurrentHashMap<>();

    public RpcClient(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        // 监听回复队列处理响应
        rabbitTemplate.setReplyListener(reply -> {
            String correlationId = reply.getMessageProperties().getCorrelationId();
            CompletableFuture<Object> future = responseFutures.remove(correlationId);
            if (future != null) {
                future.complete(reply.getBody());
            }
        });
    }

    public Object sendRpcRequest(Object request, String exchange, String routingKey) throws InterruptedException, ExecutionException {
        String correlationId = UUID.randomUUID().toString();
        MessageProperties props = new MessageProperties();
        props.setCorrelationId(correlationId);
        props.setReplyTo("rpc-reply-queue"); // 临时队列可替换为rabbitTemplate.getReplyAddress()

        Message message = rabbitTemplate.getMessageConverter().toMessage(request, props);
        CompletableFuture<Object> future = new CompletableFuture<>();
        responseFutures.put(correlationId, future);

        rabbitTemplate.send(exchange, routingKey, message);
        return future.get(5, TimeUnit.SECONDS); // 设置超时时间
    }
}

服务端实现 #

监听请求队列，处理请求后将响应发送到客户端指定的replyTo队列：

@Service
public class RpcServer {
    private final RabbitTemplate rabbitTemplate;

    public RpcServer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @RabbitListener(queues = "rpc-request-queue")
    public void handleRpcRequest(Message requestMessage) {
        // 反序列化请求
        Object request = rabbitTemplate.getMessageConverter().fromMessage(requestMessage);
        // 业务逻辑处理
        Object response = processRequest(request);

        // 发送响应到指定回复队列
        MessageProperties replyProps = new MessageProperties();
        replyProps.setCorrelationId(requestMessage.getMessageProperties().getCorrelationId());
        Message replyMessage = rabbitTemplate.getMessageConverter().toMessage(response, replyProps);

        rabbitTemplate.send(requestMessage.getMessageProperties().getReplyTo(), replyMessage);
    }

    private Object processRequest(Object request) {
        return "Processed: " + request.toString();
    }
}

2. 直接使用RabbitTemplate内置的RPC方法 #

Spring AMQP的RabbitTemplate自带同步/异步RPC调用能力，无需手动管理correlationId和回复队列，是最简便的方案：

客户端代码 #

@Service
public class SimpleRpcClient {
    private final RabbitTemplate rabbitTemplate;

    public SimpleRpcClient(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setReplyTimeout(5000); // 设置超时时间
    }

    public Object callRpcService(Object request) {
        // 同步调用
        return rabbitTemplate.convertSendAndReceive("rpc-exchange", "rpc-routing-key", request);
        // 异步调用可替换为：
        // return rabbitTemplate.convertSendAndReceiveAsync("rpc-exchange", "rpc-routing-key", request).get();
    }
}

服务端代码 #

只需用@RabbitListener处理请求并返回结果，Spring AMQP会自动处理响应发送：

@Service
public class SimpleRpcServer {
    @RabbitListener(queues = "rpc-request-queue")
    public Object handleRpcRequest(Object request) {
        return "Processed request: " + request.toString();
    }
}

3. 基于Reactor的异步RPC（适用于WebFlux场景） #

如果项目使用Spring WebFlux，可借助RabbitFlux实现非阻塞RPC调用：

客户端代码 #

@Service
public class ReactiveRpcClient {
    private final RabbitFlux rabbitFlux;

    public ReactiveRpcClient(RabbitFlux rabbitFlux) {
        this.rabbitFlux = rabbitFlux;
    }

    public Mono<Object> callReactiveRpc(Object request) {
        return rabbitFlux.sendAndReceive("rpc-exchange", "rpc-routing-key",
                message -> message
                        .body(request)
                        .properties(props -> props.correlationId(UUID.randomUUID().toString()))
        ).map(Message::getBody);
    }
}

服务端代码 #

@Service
public class ReactiveRpcServer {
    @RabbitListener(queues = "rpc-request-queue")
    public Mono<Object> handleReactiveRpcRequest(Object request) {
        return Mono.fromCallable(() -> "Processed reactive request: " + request.toString());
    }
}

关键注意事项 #

• 超时设置：所有方案都必须配置合理的超时时间，避免客户端无限等待
• 异常处理：覆盖超时、消息丢失、服务端异常等场景，返回明确的错误信息或默认值
• 回复队列优化：临时队列适合一次性RPC调用，可避免堆积；固定队列需配置合适的消费者数量
• 序列化一致性：客户端与服务端需使用相同的消息转换器（如Jackson2JsonMessageConverter），避免序列化错误

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