python – Celery – 链接远程回调

发布时间：2020-12-20 13:10:24 所属栏目：Python 来源：网络整理

导读：我有三个Celery任务分别在三个不同的服务器上运行. tasks.send_push_notification tasks.send_sms tasks.send_email 我想设置一个工作流程,如果发送推送通知失败,我应该尝试发送短信.如果发送短信失败,我应该发送电子邮件. 如果这3个任务及其代码库位于同一

我有三个Celery任务分别在三个不同的服务器上运行.

> tasks.send_push_notification
> tasks.send_sms
> tasks.send_email

我想设置一个工作流程,如果发送推送通知失败,我应该尝试发送短信.如果发送短信失败,我应该发送电子邮件.

如果这3个任务及其代码库位于同一台服务器上,我会按照the example on chained tasks并完成类似的操作

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',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',subtask('tasks.send_sms',subtask('tasks.send_email',payload)
)()
# fails because I am not adding the tasks on the broker

如何才能做到这一点？

更新：
我可以使用链接NOT链来做到这一点.

from celery import subtask
res = app.send_task(
    'tasks.send_push_notification',(payload,),link=subtask(
        'tasks.send_sms',link=subtask(
            'tasks.send_email',)
    )
)

有很多筑巢.而且因为我实际上需要创建一个数据库驱动的工作流,所以用这种方式创建它会很复杂.

解决方法

为什么不在你的任务中处理它,

def push_notification_task(payload):
    if not send_push_notification(payload):
        sms_notification_task.delay(payload)

def sms_notification_task(payload):
    if not send_sms_notification(payload):
        email_notification_task.delay(payload)

def email_notification_task(payload):
    send_email_notification(payload)

此外,chain将按给定顺序执行所有任务,而您希望下一个任务仅在首次失败时运行.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!