远程 celery worker 完成后通知或回调烧瓶答案

【问题标题】：Notify or callback flask after remote celery worker is completed远程 celery worker 完成后通知或回调烧瓶
【发布时间】：2020-08-14 12:17:05
【问题描述】：

我在两台不同的机器上运行 celery client(Flask) 和 worker，现在一旦 worker 完成任务，我需要在客户端回调一个函数。这可能吗？

Celery 客户端：-

celery_app=Celery('test_multihost', broker='amqp://test:test@<worker_ip>/test_host', backend='rpc')
result= testMethod1.apply_async((param1, param2,param3), link=testMethod2.s())

@celery_app.task
def testMethod2():
    #testMethod2 body.

芹菜工人：-

celery_app=Celery('test_multihost', broker='amqp://test:test@<worker_ip>/test_host', backend='rpc')
@celery_app.task
def testMethod1():
   #testMethod1 body

但问题是函数 testMethod2 在 celery worker 端执行，而不是在客户端。

无论如何我可以在客户端回调该方法吗？

【问题讨论】：

标签： python flask celery celery-task remote-client

【解决方案1】：

这样做的一种方法是让 Celery 将其结果写入数据库表中，并使用 Flask 通过重复查询数据库来轮询任务的结果。类似的结构可能是在 Redis 中保存已完成任务的寄存器，但要点是相同的。

您想向用户触发完成消息吗？如果您可以通过电子邮件/短信通知，当然可以让 Celery 处理。

如果您需要启动一些 Flask 进程 - 并且出于某种原因它确实需要在 Flask 的生态系统中 - 使用带有 requests 模块的工作线程来调用 Flask 正在侦听的端点。

【讨论】：

投票是我最后的选择。如果是电子邮件，我会通过 celery worker 发送，但我想要的是在 celery worker 完成任务后调用 Flask 应用程序中的一些数据处理方法。
但是为什么你需要在inside烧瓶中完成呢？请记住，它是一个 Web 应用程序，而您所描述的并不是它的构建目的。如果您确实需要这样做，请使用带有requests 的worker 来调用Flask 正在侦听的端点。这样，您就可以开始数据处理步骤了。
我是 Flask 和 Celery 的新手。谢谢你的建议。会试试这个。
我找到了使用 celery 信号的解决方案
不错！您可以在下面添加它以供将来参考吗？

【解决方案2】：

我使用 celery 信号中的 @after_task_publish 解决了这个问题。代码sn -p如下：-

@after_task_publish.connect(sender=<registered_celery_task>)
def testMethod2(sender=None, headers=None, body=None, **kwargs):
    #callback body

celery worker 在远程机器上完成后会调用 testMethod2。在这里，我可以使用 headers 参数访问 celery worker 的结果。

【讨论】：