用 Celery 同时执行两个任务

Question

我正在本地环境中测试 celery。我的 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.单个 Worker - 单个队列。

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

这将启动一个同时执行 2 个任务的工作器。

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