我目前正在使用 django 和 celery，一切正常。
但是，如果服务器过载，我希望能够通过检查当前计划了多少任务，让用户有机会取消任务。
我怎样才能做到这一点？
我使用 redis 作为代理。
我刚刚发现了这个： Retrieve list of tasks in a queue in Celery
这在某种程度上与我的问题有关，但我不需要列出任务，只需计算它们 :)
【问题讨论】：
标签： python django redis celery
这是使用与代理无关的 celery 获取队列中消息数量的方法。
通过使用connection_or_acquire，您可以利用 celery 的内部连接池来最大限度地减少与您的代理的打开连接数。
celery = Celery(app)
with celery.connection_or_acquire() as conn:
conn.default_channel.queue_declare(
queue='my-queue', passive=True).message_count
您还可以扩展 Celery 以提供此功能：
from celery import Celery as _Celery
class Celery(_Celery)
def get_message_count(self, queue):
'''
Raises: amqp.exceptions.NotFound: if queue does not exist
'''
with self.connection_or_acquire() as conn:
return conn.default_channel.queue_declare(
queue=queue, passive=True).message_count
celery = Celery(app)
num_messages = celery.get_message_count('my-queue')
【讨论】：
如果你的broker配置为redis://localhost:6379/1，你的任务提交到通用celery队列，那么你可以通过以下方式获取长度：
import redis
queue_name = "celery"
client = redis.Redis(host="localhost", port=6379, db=1)
length = client.llen(queue_name)
或者，从 shell 脚本（适用于监视器等）：
$ redis-cli -n 1 -h localhost -p 6379 llen celery
【讨论】：
如果你已经在你的应用中配置了redis，你可以试试这个：
from celery import Celery
QUEUE_NAME = 'celery'
celery = Celery(app)
client = celery.connection().channel().client
length = client.llen(QUEUE_NAME)
【讨论】：
client = app.broker_connection().channel().client
获取一个Celery使用的redis客户端实例，然后检查队列长度。每次使用时不要忘记释放连接（使用.acquire）：
# Get a configured instance of celery:
from project.celery import app as celery_app
def get_celery_queue_len(queue_name):
with celery_app.pool.acquire(block=True) as conn:
return conn.default_channel.client.llen(queue_name)
始终从池中获取连接，不要手动创建它。否则，您的 redis 服务器将用完连接槽，这将杀死您的其他客户端。
【讨论】：
我将围绕未找到错误扩展 @StephenFuhry 的答案，因为即使 Celery 建议 mess with brokers directly，检索队列长度的或多或少与代理无关的方式也是有益的。在 Celery 4（使用 Redis 代理）中，此错误如下所示：
ChannelError: Channel.queue_declare: (404) NOT_FOUND - no queue 'NAME' in vhost '/'
观察：
ChannelError 是一个 kombu 异常（事实上，它是 amqp 和 kombu “重新导出”它）。
在 Redis 代理 Celery/Kombu 上将队列表示为 Redis 列表
Redis 集合类型键为removed whenever the collection becomes empty
如果我们看看queue_declare 做了什么，它有these lines：
if passive and not self._has_queue(queue, **kwargs):
raise ChannelError(...)
Kombu Redis 虚拟传输的_has_queue 是this：
def _has_queue(self, queue, **kwargs):
with self.conn_or_acquire() as client:
with client.pipeline() as pipe:
for pri in self.priority_steps:
pipe = pipe.exists(self._q_for_pri(queue, pri))
return any(pipe.execute())
结论是，在从queue_declare 引发的 Redis 代理 ChannelError 上是可以的（当然对于现有队列），只是意味着队列是空的。
这是一个如何输出所有活动 Celery 队列长度的示例（通常应该为 0，除非您的工作人员无法处理任务）。
from kombu.exceptions import ChannelError
def get_queue_length(name):
with celery_app.connection_or_acquire() as conn:
try:
ok_nt = conn.default_channel.queue_declare(queue=name, passive=True)
except ChannelError:
return 0
else:
return ok_nt.message_count
for queue_info in celery_app.control.inspect().active_queues().values():
print(queue_info[0]['name'], get_queue_length(queue_info[0]['name']))
【讨论】：