如何从 Flask 向 Celery worker 发送中断？

Question

问题

我已经阅读了一些关于从 Flask 应用程序中访问 Celery worker 的状态的信息，就像这个 tutorial，但是你能走另一条路吗？在 Celery worker 启动后发送中断或对其进行内省？

我读过一些关于 signals 的内容，但要么还不理解它们，要么这不是我要找的东西。可能两者都有。

背景

我正在使用 Celery 启动一个订阅 MQTT 主题的长运行循环，我还希望能够从另一个端点关闭 process/subscription我的烧瓶应用程序。最好的方法是什么？或者有什么办法？

示例代码

from flask import Flask
from celery import Celery
import time

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@celery.task(bind=True)
def test_loop(self):
    i=0
    running = True
    while running:
        i = i+1
        print "loop running %d" % i
        time.sleep(1)

@app.route('/')
def index():
    return 'index page'

@app.route('/start')
def start():
    global task
    task = test_loop.delay()
    return "started loop"

@app.route('/stop')
def stop():
    global task             ### What I'm having trouble with
    task.running = False    ### How can I interrupt/introspect into the task?
    return "stopped loop"

TL/DR

有没有办法在 Celery worker 启动后发送中断或对其进行内省？如何停止在 Flask 的 Celery Worker 中启动的长运行循环？

Answer 1

我个人的想法是永远远离运行的任务。

如果您绝对必须中止任务，则可以使用撤销。 http://docs.celeryproject.org/en/latest/userguide/workers.html#revoke-revoking-tasks

@app.route('/stop')
def stop():
    global task
    task.revoke(terminate=True, signal='SIGKILL')
    return "stopped loop"

Celery 对于您的用例来说可能有些过分，但我不完全确定您的最终目标是什么，所以我无法提供任何替代方案。

如何从 Flask 向 Celery worker 发送中断？

How to send interrupt to Celery worker from Flask?

python

signals

interrupt

celery

flask