RabbitMQ RPC模式下发布者未等待订阅者响应问题求助

咱们先明确核心问题：你要实现的是RPC（远程过程调用）风格的RabbitMQ通信，这要求发布者发送请求后，必须阻塞等待订阅者处理完成并返回响应——但目前你的发布端应该是缺少了「监听响应队列+同步等待」的关键逻辑，哪怕订阅者处理耗时久，发布者也没停下来等结果。

下面给你梳理具体的修复方案，结合代码示例来对齐你的场景：

一、发布者端的核心调整：加上等待响应的逻辑 #

发布者必须做这几件事才能实现等待：

1. 创建一个临时自动删除队列，专门用来接收当前请求的响应
2. 给每个请求生成唯一的correlationId，用来关联请求和对应的响应
3. 发送请求时指定replyTo字段为这个临时队列的名称
4. 监听临时队列，阻塞等待带有对应correlationId的响应消息

补全后的发布者代码示例（和你给出的代码风格对齐）：

public String publishToDirectExchangeRPCStyle(String msg) throws InterruptedException, TimeoutException {
    String requestRoutingKey = Configuratins.requestRoutingKey;
    // 1. 创建临时自动删除队列，用于接收订阅者的响应
    String replyQueueName = channel.queueDeclare().getQueue();
    // 2. 生成唯一的correlationId，确保请求和响应一一对应
    String correlationId = java.util.UUID.randomUUID().toString();

    // 3. 构建请求消息，设置replyTo（响应队列）和correlationId（关联标识）
    AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .replyTo(replyQueueName)
            .correlationId(correlationId)
            .build();

    // 发送请求到指定交换机和路由键
    channel.basicPublish(Configuratins.directExchangeName, requestRoutingKey, props, msg.getBytes(StandardCharsets.UTF_8));

    // 4. 用CountDownLatch实现同步等待响应
    final CountDownLatch latch = new CountDownLatch(1);
    final String[] responseHolder = new String[1];

    // 监听临时队列，只处理对应correlationId的响应
    String consumerTag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
        if (delivery.getProperties().getCorrelationId().equals(correlationId)) {
            responseHolder[0] = new String(delivery.getBody(), StandardCharsets.UTF_8);
            latch.countDown(); // 收到匹配的响应，释放等待锁
        }
    }, consumerTag -> {});

    // 设置超时时间（比如30秒，根据你的业务耗时调整），避免无限阻塞
    if (latch.await(30, TimeUnit.SECONDS)) {
        channel.basicCancel(consumerTag); // 取消监听，清理资源
        return responseHolder[0];
    } else {
        channel.basicCancel(consumerTag);
        throw new RuntimeException("RPC请求超时，未收到订阅者的响应");
    }
}

二、订阅者端的配合调整：正确返回响应 #

订阅者处理完消息后，必须从请求的replyTo字段拿到发布者的响应队列，并且把correlationId原封不动带回，这样发布者才能匹配到对应的请求：

public void consumeRequest() throws IOException, InterruptedException {
    channel.basicConsume(Configuratins.requestQueueName, false, (consumerTag, delivery) -> {
        AMQP.BasicProperties requestProps = delivery.getProperties();
        String replyToQueue = requestProps.getReplyTo();
        String correlationId = requestProps.getCorrelationId();

        // 模拟你的耗时处理逻辑
        String requestMsg = new String(delivery.getBody(), StandardCharsets.UTF_8);
        String responseMsg = "处理完成：" + requestMsg;
        Thread.sleep(5000); // 模拟5秒的耗时操作

        // 构建响应消息，必须带上原correlationId
        AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                .correlationId(correlationId)
                .build();

        // 把响应发送到发布者指定的临时队列
        channel.basicPublish("", replyToQueue, replyProps, responseMsg.getBytes(StandardCharsets.UTF_8));

        // 手动确认消息已处理完成，避免RabbitMQ重复投递
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }, consumerTag -> {});
}

三、几个关键注意点 #

• 超时机制必须加：一定要给发布者的等待逻辑设置超时时间，避免因为订阅者异常导致发布者无限阻塞
• correlationId要唯一：用UUID生成即可，确保每个请求的标识不重复，避免响应匹配错误
• 临时队列自动清理：用queueDeclare()默认创建的队列是临时、自动删除的，发布者取消监听后会自动销毁，不用手动清理
• 消息手动确认：订阅者用basicAck手动确认消息，避免处理中的消息因为服务重启等原因丢失

这样调整后，发布者就会乖乖等待订阅者处理完成并返回响应，哪怕订阅者处理耗时较长（只要在超时时间内），也能正确获取到结果。

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