Celery：路由任务问题 - 只有一名工作人员使用所有队列中的所有任务

Question

我有一些任务具有手动配置的路线和 3 个工作人员，这些工作人员配置为使用特定队列中的任务。但是只有一名工作人员完成了所有任务，我不知道如何解决这个问题。

我的celeryconfig.py

    class CeleryConfig:
    enable_utc = True
    timezone = 'UTC'

    imports = ('events.tasks')
    broker_url = Config.BROKER_URL
    broker_transport_options = {'visibility_timeout': 10800}  # 3H

    worker_hijack_root_logger = False

    task_protocol = 2
    task_ignore_result = True
    task_publish_retry_policy = {'max_retries': 3, 'interval_start': 0, 'interval_step': 0.2, 'interval_max': 0.2}
    task_time_limit = 30  # sec
    task_soft_time_limit = 15  # sec

    task_default_queue = 'low'
    task_default_exchange = 'low'
    task_default_routing_key = 'low'

    task_queues = (
        Queue('daily', Exchange('daily'), routing_key='daily'),
        Queue('high', Exchange('high'), routing_key='high'),
        Queue('normal', Exchange('normal'), routing_key='normal'),
        Queue('low', Exchange('low'), routing_key='low'),
        Queue('service', Exchange('service'), routing_key='service'),
        Queue('award', Exchange('award'), routing_key='award'),
    )

    task_route = {
        # -- SCHEDULE QUEUE --
        base_path.format(task='refresh_rank'): {'queue': 'daily'}
        # -- HIGH QUEUE --
        base_path.format(task='execute_order'): {'queue': 'high'},
        # -- NORMAL QUEUE --
        base_path.format(task='calculate_cost'): {'queue': 'normal'},
        # -- SERVICE QUEUE --
        base_path.format(task='send_pin'): {'queue': 'service'},
        # -- LOW QUEUE
        base_path.format(task='invite_to_tournament'): {'queue': 'low'},
        # -- AWARD QUEUE
        base_path.format(task='get_lesson_award'): {'queue': 'award'},
        # -- TEST TASK

    worker_concurrency = multiprocessing.cpu_count() * 2 + 1
    worker_prefetch_multiplier = 1  #
    worker_max_tasks_per_child = 1
    worker_max_memory_per_child = 90000  # 90MB

    beat_max_loop_interval = 60 * 5  # 5 min

我在 docker 中运行工人，这是我的 stack.yml 的一部分

    version: "3.7"

    services:

      worker_high:
        command: celery worker -l debug -A runcelery.celery -Q high -n worker.high@%h

      worker_normal:
        command: celery worker -l debug -A runcelery.celery -Q normal,award,service,low -n worker.normal@%h

      worker_schedule:
        command: celery worker -l debug -A runcelery.celery -Q daily -n worker.schedule@%h

      beat:
        command: celery beat -l debug -A runcelery.celery

      flower:
        command: flower -l debug -A runcelery.celery --port=5555 

      broker:
        image: redis:5.0-alpine

我认为我的配置是正确的并且运行命令也是正确的，但是 docker logs 和flower 显示只有 worker.normal 消耗所有任务。

我

更新

这是task.py的一部分：

def refresh_rank_in_tournaments():
    logger.debug(f'Start task refresh_rank_in_tournaments')
    return AnalyticBackgroundManager.refresh_tournaments_rank()

base_path 是完整任务路径的快捷方式： base_path = 'events.tasks.{task}'

execute_order任务代码：

    @celery.task(bind=True, default_retry_delay=5)
    def execute_order(self, private_id, **kwargs):
        try:
            return OrderBackgroundManager.execute_order(private_id, **kwargs)
        except IEXException as exc:
            raise self.retry(exc=exc)

此任务将在视图中调用tasks.execute_order.delay(id)

Answer 1

您的 worker.normal 订阅了 normal,award,service,low 队列。此外，low 队列是默认队列，因此每个没有明确设置队列的任务都将在 worker.normal 上执行。