如果你正在使用芹菜+Django，最简单的方法是在虚拟环境中直接从终端使用命令检查任务，或者使用芹菜的完整路径:

道格:http://docs.celeryproject.org/en/latest/userguide/workers.html?highlight=revoke inspecting-workers

$ celery inspect reserved
$ celery inspect active
$ celery inspect registered
$ celery inspect scheduled

另外，如果你正在使用芹菜+RabbitMQ，你可以使用下面的命令检查队列列表:

更多信息:https://linux.die.net/man/1/rabbitmqctl

$ sudo rabbitmqctl list_queues

这在我的申请中很奏效:

def get_celery_queue_active_jobs(queue_name):
    connection = <CELERY_APP_INSTANCE>.connection()

    try:
        channel = connection.channel()
        name, jobs, consumers = channel.queue_declare(queue=queue_name, passive=True)
        active_jobs = []

        def dump_message(message):
            active_jobs.append(message.properties['application_headers']['task'])

        channel.basic_consume(queue=queue_name, callback=dump_message)

        for job in range(jobs):
            connection.drain_events()

        return active_jobs
    finally:
        connection.close()

Active_jobs将是一个字符串列表，对应于队列中的任务。

不要忘记将CELERY_APP_INSTANCE替换为您自己的。

感谢@ashish在这里为我指出了正确的方向:https://stackoverflow.com/a/19465670/9843399

我认为获取正在等待的任务的唯一方法是保留一个已启动任务的列表，并让任务在启动时将自己从列表中删除。

通过rabbitmqctl和list_queues，你可以了解有多少任务正在等待，而不是任务本身:http://www.rabbitmq.com/man/rabbitmqctl.1.man.html

如果你想要的包括正在处理的任务，但还没有完成，你可以保留一个任务列表，并检查它们的状态:

from tasks import add
result = add.delay(4, 4)

result.ready() # True if finished

或者让芹菜使用CELERY_RESULT_BACKEND存储结果，并检查哪些任务不在其中。

Redis json序列化的复制粘贴解决方案:

def get_celery_queue_items(queue_name):
    import base64
    import json  

    # Get a configured instance of a celery app:
    from yourproject.celery import app as celery_app

    with celery_app.pool.acquire(block=True) as conn:
        tasks = conn.default_channel.client.lrange(queue_name, 0, -1)
        decoded_tasks = []

    for task in tasks:
        j = json.loads(task)
        body = json.loads(base64.b64decode(j['body']))
        decoded_tasks.append(body)

    return decoded_tasks

它与Django一起工作。只是别忘了改变你的项目。

from celery.task.control import inspect
def key_in_list(k, l):
    return bool([True for i in l if k in i.values()])

def check_task(task_id):
    task_value_dict = inspect().active().values()
    for task_list in task_value_dict:
        if self.key_in_list(task_id, task_list):
             return True
    return False

subprocess.run:

import subprocess
import re
active_process_txt = subprocess.run(['celery', '-A', 'my_proj', 'inspect', 'active'],
                                        stdout=subprocess.PIPE).stdout.decode('utf-8')
return len(re.findall(r'worker_pid', active_process_txt))

注意使用your_proj更改my_proj