使用Celery同时执行两个任务

Question

我正在本地环境中测试芹菜。我的Python文件有以下两行代码：

celery_app.send_task('tasks.test1', args=[self.id], kwargs={})
celery_app.send_task('tasks.test2', args=[self.id], kwargs={})

从控制台输出来看，它们似乎依次执行。但test2只在test1完成后运行。至少这是读取控制台输出的方式。

这些任务彼此之间没有依赖性，所以我不希望一个任务等待另一个任务完成后再转到下一行。

如何同时执行两项任务？

---- **** -----
--- * ***  * -- Darwin-14.0.0-x86_64-i386-64bit
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x104cd8c10
- ** ---------- .> transport:   sqs://123
- ** ---------- .> results:     disabled
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ----
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery

Answer 1

有多种方法可以实现这一点。

1.单个工作线程-单个队列

$ celery -A my_app worker -l info  -c 2 -n my_worker

这将启动一个同时执行两个任务的工作程序。

2.多个工人-单个队列

$ celery -A my_app worker -l info  -c 1 -n my_worker1
$ celery -A my_app worker -l info  -c 1 -n my_worker2

这将启动两个工人，一次执行一个任务。请注意，两个任务都在同一队列中。

3.多个工作者-多个队列

$ celery -A my_app worker -l info  -c 1 -n my_worker1 -Q queue1
$ celery -A my_app worker -l info  -c 1 -n my_worker2 -Q queue2

这将启动两个工人，一次执行一个任务。但在这里，您可以相应地安排任务。

celery_app.send_task('tasks.test1', args=[self.id], kwargs={}, queue='queue1')
celery_app.send_task('tasks.test2', args=[self.id], kwargs={}, queue='queue2')

4.单个工作程序-所有队列

$ celery -A my_app worker -l info -n my_worker1 

如果您没有提到任何队列，默认情况下它将从所有队列中消耗。

Answer 2

使用--autoscale选项调用工作人员，该选项将根据需要放大和缩小流程。

--autoscale AUTOSCALE
                       Enable autoscaling by providing max_concurrency,
                       min_concurrency. Example:: --autoscale=10,3 (always
                       keep 3 processes, but grow to 10 if necessary)

例如。

celery -A sandbox worker --autoscale=10,0 --loglevel=info