如何在代码中查询Celery队列当前的消费者数量

总预留任务数的计算逻辑为：总预留任务数 = 消费者总数量 * worker_prefetch_multiplier 配置值，要在代码中获取消费者总数可按以下方案实现：

前置准备 #

首先需要启用RabbitMQ管理插件，在节点执行命令：
rabbitmq-plugins enable rabbitmq_management
启用后默认会在15672端口开放HTTP API接口，使用RabbitMQ的账号密码通过Basic Auth即可调用。

通用API调用方法 #

普通AMQP协议没有提供查询消费者数量的标准接口，基于管理插件API的方案是最通用的实现方式：

• 查询单队列消费者数：发送GET请求到 http://<你的RabbitMQ主机地址>:15672/api/queues/<vhost编码>/<队列名>，返回JSON中的consumers字段即为该队列当前的消费者数量

  注意：默认vhost为/，需要转义为%2F传入请求地址

• 查询所有队列总消费者数：发送GET请求到 http://<你的RabbitMQ主机地址>:15672/api/queues，遍历返回的所有队列数据，累加每个队列的consumers字段值即可得到全局消费者总数

常用语言示例 #

Python 示例 #

import requests
from requests.auth import HTTPBasicAuth

# 配置参数
RABBITMQ_HOST = "你的主机地址"
RABBITMQ_PORT = 15672
USERNAME = "你的账号"
PASSWORD = "你的密码"
VHOST = "%2F" # 若使用自定义vhost替换为对应编码后的值
QUEUE_NAME = "你的队列名"

# 单队列消费者查询
response = requests.get(
    f"http://{RABBITMQ_HOST}:{RABBITMQ_PORT}/api/queues/{VHOST}/{QUEUE_NAME}",
    auth=HTTPBasicAuth(USERNAME, PASSWORD)
)
queue_info = response.json()
single_queue_consumers = queue_info["consumers"]

# 总消费者查询
all_queues = requests.get(
    f"http://{RABBITMQ_HOST}:{RABBITMQ_PORT}/api/queues/{VHOST}",
    auth=HTTPBasicAuth(USERNAME, PASSWORD)
).json()
total_consumers = sum(queue["consumers"] for queue in all_queues)

Java 示例（基于OkHttp） #

import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import java.util.Base64;

public class RabbitmqConsumerCount {
    public static void main(String[] args) throws Exception {
        String host = "你的主机地址";
        int port = 15672;
        String username = "你的账号";
        String password = "你的密码";
        String vhost = "%2F"; // 若使用自定义vhost替换为对应编码后的值
        
        OkHttpClient client = new OkHttpClient();
        String auth = Base64.getEncoder().encodeToString((username + ":" + password).getBytes());
        
        // 查询总消费者数
        Request request = new Request.Builder()
                .url("http://" + host + ":" + port + "/api/queues/" + vhost)
                .addHeader("Authorization", "Basic " + auth)
                .build();
        
        try (Response response = client.newCall(request).execute()) {
            JSONArray queues = JSONArray.parseArray(response.body().string());
            int totalConsumers = 0;
            for (int i = 0; i < queues.size(); i++) {
                JSONObject queue = queues.getJSONObject(i);
                totalConsumers += queue.getIntValue("consumers");
            }
            System.out.println("总消费者数：" + totalConsumers);
        }
    }
}

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