[python] Celery에서 대기열에있는 작업 목록 검색

우선 순위가 지정된 작업을 사용하지 않으면 Redis를 사용하는 경우 실제로 매우 간단 합니다. 작업 수를 얻으려면 :

redis-cli -h HOST -p PORT -n DATABASE_NUMBER llen QUEUE_NAME

그러나 우선 순위가 지정된 작업 은 redis에서 다른 키를 사용 하므로 전체 그림이 약간 더 복잡합니다. 전체 그림은 모든 작업 우선 순위에 대해 redis를 쿼리해야한다는 것입니다. 파이썬과 꽃 프로젝트에서 다음과 같이 보입니다.

PRIORITY_SEP = '\x06\x16'
DEFAULT_PRIORITY_STEPS = [0, 3, 6, 9]


def make_queue_name_for_pri(queue, pri):
    """Make a queue name for redis

    Celery uses PRIORITY_SEP to separate different priorities of tasks into
    different queues in Redis. Each queue-priority combination becomes a key in
    redis with names like:

     - batch1\x06\x163 <-- P3 queue named batch1

    There's more information about this in Github, but it doesn't look like it
    will change any time soon:

      - https://github.com/celery/kombu/issues/422

    In that ticket the code below, from the Flower project, is referenced:

      - https://github.com/mher/flower/blob/master/flower/utils/broker.py#L135

    :param queue: The name of the queue to make a name for.
    :param pri: The priority to make a name with.
    :return: A name for the queue-priority pair.
    """
    if pri not in DEFAULT_PRIORITY_STEPS:
        raise ValueError('Priority not in priority steps')
    return '{0}{1}{2}'.format(*((queue, PRIORITY_SEP, pri) if pri else
                                (queue, '', '')))


def get_queue_length(queue_name='celery'):
    """Get the number of tasks in a celery queue.

    :param queue_name: The name of the queue you want to inspect.
    :return: the number of items in the queue.
    """
    priority_names = [make_queue_name_for_pri(queue_name, pri) for pri in
                      DEFAULT_PRIORITY_STEPS]
    r = redis.StrictRedis(
        host=settings.REDIS_HOST,
        port=settings.REDIS_PORT,
        db=settings.REDIS_DATABASES['CELERY'],
    )
    return sum([r.llen(x) for x in priority_names])

실제 작업을 원하면 다음과 같이 사용할 수 있습니다.

redis-cli -h HOST -p PORT -n DATABASE_NUMBER lrange QUEUE_NAME 0 -1

거기에서 반환 된 목록을 직렬화 해제해야합니다. 제 경우에는 다음과 같은 방법으로 이것을 달성 할 수있었습니다.

r = redis.StrictRedis(
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    db=settings.REDIS_DATABASES['CELERY'],
)
l = r.lrange('celery', 0, -1)
pickle.loads(base64.decodestring(json.loads(l[0])['body']))

역 직렬화에는 다소 시간이 걸릴 수 있으므로 위의 명령을 조정하여 다양한 우선 순위로 작업해야합니다.

답변

from amqplib import client_0_8 as amqp
conn = amqp.Connection(host="localhost:5672 ", userid="guest",
                       password="guest", virtual_host="/", insist=False)
chan = conn.channel()
name, jobs, consumers = chan.queue_declare(queue="queue_name", passive=True)

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

$ sudo rabbitmqctl list_queues

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