Django/Celery在本地主机上有多个队列-路由不工作

2024-03-29 01:33:07 发布

您现在位置:Python中文网/ 问答频道 /正文

我跟着芹菜docs在我的开发机器上定义了两个队列。

我的芹菜设置:

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中定义的一个任务,并让默认队列处理所有其他任务。

我也遵循这个SO questionrabbitmqctl list_queues返回celery 0,运行rabbitmqctl list_bindings返回exchange celery queue celery []两次。重新启动rabbit服务器没有任何改变。


Tags: truedefault定义exchange队列queueroutingcelery
1条回答
网友
1楼 · 发布于 2024-03-29 01:33:07

好吧,我想出来了。以下是我的整个设置,设置和如何运行芹菜,为那些谁可能会想同我的问题做了同样的事情。

设置

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不是第一个worker必须的,但如果您有任何其他芹菜实例正在运行,则必须。设置-n worker_name--hostname=default@%h相同。

终端-表2:

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

这将启动消费者从feeds队列执行任务的第二个worker。注意-n feeds_worker,如果使用-l debug(log level=debug)运行,您将看到两个工作进程正在它们之间同步。

终端-表3:

celery -A proj beat -l debug

这将启动节拍,根据CELERYBEAT_SCHEDULE中的计划执行任务。 我不需要改变任务,也不需要改变CELERYBEAT_SCHEDULE

例如,对于应该转到feeds队列的任务,这就是我的CELERYBEAT_SCHEDULE的外观:

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

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

相关问题 更多 >