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

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

要获得队列上的任务数，你可以使用flower库，下面是一个简化的例子:

from flower.utils.broker import Broker
from django.conf import settings

def get_queue_length(queue):
    broker = Broker(settings.CELERY_BROKER_URL)
    queues_result = broker.queues([queue])
    return queues_result.result()[0]['messages']

EDIT:查看获取队列中任务列表的其他答案。

你应该看这里: 芹菜指南-检查工人

基本上是这样的:

my_app = Celery(...)

# Inspect all nodes.
i = my_app.control.inspect()

# Show the items that have an ETA or are scheduled for later processing
i.scheduled()

# Show tasks that are currently active.
i.active()

# Show tasks that have been claimed by workers
i.reserved()

这取决于你想要什么

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

通过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存储结果，并检查哪些任务不在其中。

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