本地主机上的 Django/Celery 多个队列 - 路由不起作用

我跟着芹菜docs http://celery.readthedocs.org/en/latest/userguide/routing.html#manual-routing在我的开发机器上定义 2 个队列。

我的芹菜设置：

CELERY_ALWAYS_EAGER = True
CELERY_TASK_RESULT_EXPIRES = 60  # 1 mins
CELERYD_CONCURRENCY = 2
CELERYD_MAX_TASKS_PER_CHILD = 4
CELERYD_PREFETCH_MULTIPLIER = 1
CELERY_CREATE_MISSING_QUEUES = True
CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('feeds', Exchange('feeds'), routing_key='arena.social.tasks.#'),
)
CELERY_ROUTES = {
    'arena.social.tasks.Update': {
        'queue': 'fs_feeds',
    },
}

我在项目的 virtualenv 中打开了两个终端窗口，并运行以下命令：

terminal_1$ celery -A arena worker -Q default -B -l debug --purge -n deafult_worker
terminal_2$ celery -A arena worker -Q feeds -B -l debug --purge -n feeds_worker

我得到的是所有任务都由两个队列处理。

我的目标是拥有一个队列来仅处理中定义的一项任务CELERY_ROUTES和默认队列来处理所有其他任务。

我也关注了这个所以问题 https://stackoverflow.com/questions/10079816/route-celery-task-to-specific-queue, rabbitmqctl list_queues回报celery 0，并运行rabbitmqctl list_bindings回报exchange celery queue celery []两次。重新启动兔子服务器没有改变任何东西。

好吧，所以我想通了。以下是我的整个设置、设置以及如何运行 celery，对于那些可能想知道与我的问题相同的事情的人。

Settings

CELERY_TIMEZONE = TIME_ZONE
CELERY_ACCEPT_CONTENT = ['json', 'pickle']
CELERYD_CONCURRENCY = 2
CELERYD_MAX_TASKS_PER_CHILD = 4
CELERYD_PREFETCH_MULTIPLIER = 1

# celery queues setup
CELERY_DEFAULT_QUEUE = 'default'
CELERY_DEFAULT_EXCHANGE_TYPE = 'topic'
CELERY_DEFAULT_ROUTING_KEY = 'default'
CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('feeds', Exchange('feeds'), routing_key='long_tasks'),
)
CELERY_ROUTES = {
    'arena.social.tasks.Update': {
        'queue': 'feeds',
        'routing_key': 'long_tasks',
    },
}

如何运行芹菜？

终端 - 选项卡 1：

celery -A proj worker -Q default -l debug -n default_worker

这将启动第一个工作进程来消耗默认队列中的任务。笔记！-n default_worker对于第一个工作人员来说不是必须的，但如果您有任何其他启动并运行的 celery 实例，则这是必须的。环境-n worker_name是相同的--hostname=default@%h.

终端 - 选项卡 2：

celery -A proj worker -Q feeds -l debug -n feeds_worker

这将启动第二个工作人员，从提要队列中消费任务。注意-n feeds_worker，如果你正在运行-l debug（日志级别=调试），您将看到两个工作人员正在它们之间同步。

终端 - 选项卡 3：

celery -A proj beat -l debug

这将开始节拍，根据您的计划执行任务CELERYBEAT_SCHEDULE。 我不必改变任务，或者CELERYBEAT_SCHEDULE.

例如，这就是我的样子CELERYBEAT_SCHEDULE对于应该进入 feeds 队列的任务：

CELERYBEAT_SCHEDULE = {
    ...
    'update_feeds': {
        'task': 'arena.social.tasks.Update',
        'schedule': crontab(minute='*/6'),
    },
    ...
}

如您所见，无需添加'options': {'routing_key': 'long_tasks'}或指定它应该进入哪个队列。另外，如果您想知道为什么Update是大写的，因为它是一个自定义任务，被定义为celery.Task.

更新芹菜5.0+

Celery 自版本 5 以来进行了一些更改，以下是任务路由的更新设置。

如何创建队列？

Celery 可以自动创建队列。它非常适合简单的情况，其中 celery 的默认路由值就可以了。

task_create_missing_queues=True或者，如果您使用 django 设置并且您正在为所有 celery 配置命名空间CELERY_ key, CELERY_TASK_CREATE_MISSING_QUEUES=True。请注意，它默认处于打开状态。

自动计划任务路由

配置 celery 应用程序后：

celery_app.conf.beat_schedule = {
  "some_scheduled_task": {
    "task": "module.path.some_task",
    "schedule": crontab(minute="*/10"),
    "options": {"queue": "queue1"}
  }
}

自动任务路由

Celery 应用程序仍然需要先配置，然后：

app.conf.task_routes = {
  "module.path.task2": {"queue": "queue2"},
}

手动分配任务

如果您想动态路由任务，则在发送任务时指定队列：

from module import task

def do_work():
  # do some work and launch the task
  task.apply_async(args=(arg1, arg2), queue="queue3")

重新路由的更多详细信息可以在这里找到：https://docs.celeryproject.org/en/stable/userguide/routing.html https://docs.celeryproject.org/en/stable/userguide/routing.html

关于此处的调用任务：https://docs.celeryproject.org/en/stable/userguide/calling.html https://docs.celeryproject.org/en/stable/userguide/calling.html