我有三个 Celery 任务,它们在三个不同的服务器上分别运行。
如果这三个任务及其代码基础位于同一服务器上,我将遵循链接任务示例并执行类似操作。请参考:链接任务示例。
- tasks.send_push_notification
- tasks.send_sms
- tasks.send_email
如果这三个任务及其代码基础位于同一服务器上,我将遵循链接任务示例并执行类似操作。请参考:链接任务示例。
from celery import chain
from tasks import send_push_notification, send_sms, send_email
import json
# some paylaod
payload = json.dumps({})
res = chain(
send_push_notification.subtask(payload),
send_sms.subtask(payload),
send_email.subtask(payload)
)()
但是任务分布在3台不同的服务器上!
我已经尝试过
# 1
from celery import chain
from my_celery_app import app
res = chain(
app.send_task('tasks.send_push_notification', payload),
app.send_task('tasks.send_sms', payload),
app.send_task('tasks.send_email', payload)
)()
# Which fails because I am chaining tasks not subtasks
并且
# 2
from celery import chain, subtask
res = chain(
subtask('tasks.send_push_notification', payload),
subtask('tasks.send_sms', payload),
subtask('tasks.send_email', payload)
)()
# fails because I am not adding the tasks on the broker
这个怎么做呢?
更新:
我可以使用 link
而不是 chain
来实现。
from celery import subtask
res = app.send_task(
'tasks.send_push_notification', (payload, ),
link=subtask(
'tasks.send_sms', (payload, ),
link=subtask(
'tasks.send_email', (payload, ),
)
)
)
这里涉及到很多嵌套。而且因为我实际上需要创建一个基于数据库的工作流程,所以用这种方式创建会变得很复杂。
chain
是因为我想遵循“责任链”设计模式。你的解决方案行不通,因为这三个任务/它们的代码库不在一个地方。它们位于远程位置。 - Hussain