celery配置多队列实践笔记

shidifen

0.上下文交代我有一个爬虫任务，5分钟跑一次
我今天又加了一个计算任务，一天只跑一次但是要做大量的运算
我用一个队列（default），结果计算任务堵死了队列，导致所有任务都卡住

查了一下决定计算和爬虫走不同的队列，
两个任务分别是：

umonitor.tasks.multiple_thread_metric_collector
compute.tasks.multiple_thread_metric_aggregate

1
2

1.配置：# 配置队列（settings.py）
CELERY_QUEUES = (
Queue('default', Exchange('default'), routing_key='default'),
Queue('for_task_collect', Exchange('for_task_collect'), routing_key='for_task_collect'),
Queue('for_task_compute', Exchange('for_task_compute'), routing_key='for_task_compute'),
)

1
2
3
4
5
6
7

# 路由（哪个任务放入哪个队列）
CELERY_ROUTES = {
'umonitor.tasks.multiple_thread_metric_collector': {'queue': 'for_task_collect', 'routing_key': 'for_task_collect'},
'compute.tasks.multiple_thread_metric_aggregate': {'queue': 'for_task_compute', 'routing_key': 'for_task_compute'},
'compute.tasks.test': {'queue': 'for_task_compute', 'routing_key': 'for_task_compute'},
}
1
2
3
4
5
6

2.启动不同worker处理不同队列的任务如果没有启动默认队列，没在上文路由内注册的任务就没地方执行了
# 指定worker_compute去处理队列for_task_compute的任务
python manage.py celery worker -E -l INFO -n worker_compute -Q for_task_compute
# 指定worker_collect去处理队列for_task_collect的任务
python manage.py celery worker -E -l INFO -n worker_collect -Q for_task_collect

1
2
3
4

3.结论接收的任务会经过路由放入指定队列，相对应的不同worker从不同的队列中取任务来执行，这样我的计算任务出错就不会影响爬虫任务了