Celery：如何可靠且可测试地获取队列任务数量？

我明白你现在的痛点：想实现一个能在测试环境（内存后端）和生产环境（Redis后端）都能用的get_queue_size方法，之前的尝试要么在测试环境返回0，要么在Redis环境失效。结合你的代码和需求，我来梳理下问题根源和解决方案。

为什么之前的尝试失效？ #

先拆解下你三个尝试的核心问题：

1. 尝试1（queue_declare被动模式）：内存后端的Channel实现并没有正确返回队列中的任务数，所以始终返回0；但这个方法在RabbitMQ等AMQP后端是有效的。
2. 尝试2（control.inspect）：inspect.active()/scheduled()/reserved()都是查询worker正在处理/待调度/已保留的任务，不是队列中等待被worker拾取的任务，自然拿不到队列的待执行任务数。
3. 尝试3（channel.queues）：这是内存后端特有的实现，Redis等其他后端的Channel对象根本没有queues属性，所以生产环境直接失效。
4. 额外坑点：你的测试用例中，启动worker后任务会被立即执行，队列很快被清空，这也是导致你拿到0的重要原因！

兼容多后端的解决方案 #

我们可以根据当前使用的broker类型，分别实现对应的队列数量获取逻辑，同时调整测试用例确保任务在查询时仍处于队列中。

1. 实现EnhancedCelery的get_queue_size方法 #

from celery import Celery
from celery.exceptions import ChannelError, NotFound
import redis
from urllib.parse import urlparse
from typing import Optional

class EnhancedCelery(Celery):
    def get_queue_size(self, queue_name: str) -> Optional[int]:
        broker_url = self.conf.broker_url
        parsed_url = urlparse(broker_url)

        # 处理内存后端（测试环境）
        if parsed_url.scheme == 'memory':
            with self.connection_or_acquire() as connection:
                channel = connection.default_channel
                if hasattr(channel, 'queues'):
                    queue = channel.queues.get(queue_name)
                    return queue.unfinished_tasks if queue else None

        # 处理Redis后端（生产环境）
        elif parsed_url.scheme in ('redis', 'rediss'):
            # 构建Redis客户端连接
            redis_kwargs = {
                'host': parsed_url.hostname,
                'port': parsed_url.port or 6379,
                'password': parsed_url.password,
                'db': int(parsed_url.path.lstrip('/')) if parsed_url.path else 0,
                'ssl': parsed_url.scheme == 'rediss'
            }
            redis_client = redis.Redis(**redis_kwargs)
            
            # 处理Redis key前缀（如果配置了的话）
            queue_key = queue_name
            key_prefix = self.conf.get('redis_key_prefix', '')
            if key_prefix:
                queue_key = f"{key_prefix}:{queue_name}"
            
            return redis_client.llen(queue_key)

        # 处理RabbitMQ等AMQP后端
        elif parsed_url.scheme in ('amqp', 'amqps'):
            with self.connection_or_acquire() as connection:
                channel = connection.default_channel
                try:
                    _, jobs, _ = channel.queue_declare(queue=queue_name, passive=True)
                    return jobs
                except (ChannelError, NotFound):
                    return None

        # 其他后端可自行扩展
        else:
            raise NotImplementedError(f"暂不支持该broker类型: {parsed_url.scheme}")

2. 调整测试用例，确保任务留在队列中 #

你的测试用例中，worker会立即执行任务，导致队列快速清空。我们可以给任务添加延迟执行，确保查询时任务还在队列里：

class EnhancedCeleryTest(TestCase):
    def test_get_queue_size_returns_expected_value(self):
        def add_task(task):
            # 添加10秒延迟，避免worker立即执行任务
            task.apply_async(countdown=10)
        
        with start_worker(celery_test_app):
            for _ in range(7):
                add_task(sample_task_in_queue_1)
            for _ in range(4):
                add_task(sample_task_in_queue_2)
            for _ in range(2):
                add_task(sample_task_in_queue_3)
            
            self.assertEqual(celery_test_app.get_queue_size('queue_1'), 7)
            self.assertEqual(celery_test_app.get_queue_size('queue_2'), 4)
            self.assertEqual(celery_test_app.get_queue_size('queue_3'), 2)

关键说明 #

• 内存后端：利用测试环境内存broker特有的channel.queues属性，直接获取unfinished_tasks（未完成的任务数，也就是队列中待执行的数量）。
• Redis后端：Celery在Redis中用列表存储队列任务，所以用llen命令获取列表长度即可；注意如果配置了redis_key_prefix，要给队列名加上前缀。
• AMQP后端：使用标准的queue_declare被动模式，获取队列中的任务数，这是RabbitMQ等AMQP broker的标准做法。
• 测试延迟：通过countdown=10让任务延迟执行，确保在查询队列大小时，worker还没开始处理这些任务，这样就能拿到正确的数量。

内容的提问来源于stack exchange，提问作者daveoncode