Why does celery.control.inspect report fewer queued tasks than rabbitmqctl?

Question

rabbitmqctl correctly reports thousands of queued tasks:

$ sudo rabbitmqctl -q list_queues name messages messages_ready messages_unacknowledged
default 13142   13126   16

Yet celery reports:

>>> len(app.control.inspect().active()['celery@default'])
4
>>> len(app.control.inspect().scheduled()['celery@default'])
1
>>> len(app.control.inspect().reserved()['celery@default'])
16
>>> len(app.control.inspect().revoked()['celery@default'])
0

The correct number (thousands) of tasks seem to show up in app.control.inspect().stats()['celery@default']['total'], but I really want to know the correct number of outstanding queued tasks from within python, and active() et al seem to only ever report up to 16 or so -- perhaps there is a limit?

Short of using privileged subprocess calls to rabbitmqctl, how can I get the full queued task count from within python, preferably via celery (btw this server is using Celery 3.1.8 currently)

Answer 1

Celery's app.control.inspect will inspect tasks that are handled only by running workers.

Even though you have thousands of tasks in queue, your worker will execute only few specified tasks at any given point of time. These are active tasks.

In addition to that, a worker can prefetch some tasks which will be reserved for that worker. These will be shown in reserved tasks.

If you have set ETA for your tasks or if there are periodic tasks, they will come under scheduled tasks.

Looks like you have started a worker with concurrency of 4 (or worker with default settings on a 4 core machine). So active tasks are 4. Each worker process has prefetched 4 tasks, which resulted in 16 reserved tasks.

AFAIK, there is no way to get total count of tasks in queue with celery.

However there are several python solutions to get total count of messages in queue. You can check my other answer here for other ways to do this.

Update:

pika is a python client to interact with rabbitmq. You can use it to consume messages. Here is a simple example to consume each message. You can checkout more usage examples on pika docs.