如何通过代码获取RabbitMQ队列生命周期内接收的总消息数？

嘿，这个问题抓得很准——你现在用的QueueDeclarePassive确实只能拿到队列当前剩余的未消费消息数，已经被消费者处理掉的消息根本不会被统计进去，要确认总投递量得换个思路。

RabbitMQ本身是记录了队列的总接收消息数的，但得通过以下几种方式获取：

1. 用RabbitMQ Management API（最靠谱的通用方案） #

RabbitMQ的管理插件会维护队列的全量统计数据，其中就包含投递到队列的总消息数。步骤如下：

• 先确保开启了Management插件：在RabbitMQ服务器上执行rabbitmq-plugins enable rabbitmq_management
• 然后通过HTTP请求调用队列的统计接口，接口格式是GET http://[RabbitMQ地址]:15672/api/queues/[虚拟主机名]/[队列名]
• 需要用管理员账号（或有monitor权限的账号）做Basic认证

Java代码示例 #

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.Base64;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

public class RabbitMQStats {
    public static void main(String[] args) throws Exception {
        String rabbitHost = "localhost";
        String vhost = "/"; // 默认虚拟主机是/，注意要转成URL编码的%2F
        String queueName = "your-queue-name";
        String username = "admin";
        String password = "admin";

        // 处理虚拟主机的URL编码
        String encodedVhost = vhost.equals("/") ? "%2F" : vhost;
        String url = String.format("http://%s:15672/api/queues/%s/%s", rabbitHost, encodedVhost, queueName);

        // 构建Basic认证头
        String auth = username + ":" + password;
        String encodedAuth = Base64.getEncoder().encodeToString(auth.getBytes());

        HttpClient client = HttpClient.newHttpClient();
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(url))
                .header("Authorization", "Basic " + encodedAuth)
                .build();

        HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
        ObjectMapper mapper = new ObjectMapper();
        JsonNode queueStats = mapper.readTree(response.body());

        // 提取总投递消息数：message_stats下的publish字段，没有的话默认0
        long totalReceived = queueStats.path("message_stats").path("publish").asLong(0);
        System.out.println("队列接收的总消息数：" + totalReceived);
    }
}

这个方案的好处是：统计数据由RabbitMQ服务器维护，不管生产者/消费者怎么重启，只要队列没被删除，数据就不会丢，支持多生产者场景。

2. 生产者端本地计数（适合可控的单生产者场景） #

如果你的消息都是由自己的生产者发送，且只有一个生产者（或者能统一计数），可以在生产者代码里维护一个原子计数器，每成功发送一条消息就递增：

import java.util.concurrent.atomic.AtomicLong;

public class MessageProducer {
    private static final AtomicLong totalSent = new AtomicLong(0);

    public static void sendMessage(com.rabbitmq.client.Channel channel, String exchange, String routingKey, String message) throws Exception {
        channel.basicPublish(exchange, routingKey, null, message.getBytes());
        totalSent.incrementAndGet(); // 发送成功后计数
    }

    // 随时获取总发送数
    public static long getTotalSent() {
        return totalSent.get();
    }
}

这个方案简单，但局限性大：如果有多个生产者、生产者崩溃重启，计数就会不准确或丢失，只适合小范围可控的场景。

3. 通过Prometheus监控（适合有监控体系的场景） #

如果你的RabbitMQ已经接入了Prometheus监控，可以启用rabbitmq_prometheus插件，然后通过指标rabbitmq_queue_messages_published_total来获取队列的总投递数。你可以在代码里调用Prometheus的API查询这个指标的值，适合需要长期监控或集成到现有监控系统的场景。

为什么原来的方法不行？ #

再补充下：QueueDeclarePassive返回的MessageCount是队列当前的未消费消息数——也就是已经投递到队列，但还没被消费者确认ack的消息数量。那些已经被处理掉的消息，RabbitMQ不会再把它们算到这个数值里，所以自然没法拿到总接收量。

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