如何确定Celery任务的来源队列？及查看当前机器监听的队列

一、获取当前任务的来源队列 #

Celery任务实例自带的request属性包含任务投递信息，其中delivery_info字典的routing_key字段对应任务的来源队列（若未配置自定义路由规则，该值直接是队列名）。

任务内实现代码：

from celery import task

@task
def myfunc():
    # 提取来源队列名
    source_queue = myfunc.request.delivery_info.get('routing_key')
    print(f"当前任务来自队列: {source_queue}")

二、查看当前机器监听的队列 #

有两种实用方式：

1. 通过Celery控制接口查询 #

利用Celery应用的control.inspect()方法，可获取当前worker集群的队列监听状态，若仅需当前任务所在worker的队列，可按worker名称筛选：

from celery import current_app

@task
def myfunc():
    inspector = current_app.control.inspect()
    # 获取所有活跃worker的监听队列数据
    active_queues = inspector.active_queues()
    if active_queues:
        # 取当前worker的队列列表（示例取第一个worker，实际可根据worker名称精准筛选）
        worker_name = list(active_queues.keys())[0]
        queues = [q['name'] for q in active_queues[worker_name]]
        print(f"当前机器监听的队列: {queues}")

注意：使用该方法要求worker启动时默认开启的控制端口正常可用，且当前任务所在worker处于可被inspect访问的状态。

2. 通过环境变量传递启动参数 #

若worker启动时通过-Q指定了监听队列，可在启动时将队列名存入环境变量，任务内直接读取：
启动worker的命令：

export CELERY_LISTEN_QUEUES="queueA,queueB,queueC"
celery -A your_app worker -Q $CELERY_LISTEN_QUEUES

任务内读取代码：

import os

@task
def myfunc():
    listen_queues = os.environ.get('CELERY_LISTEN_QUEUES', '').split(',')
    print(f"当前机器监听的队列: {listen_queues}")

这种方式无需依赖Celery控制接口，实现更直接。

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