如何从Flask向Celery工作人员发送中断？

Question

题

我已经阅读了一些关于从Flask应用程序中获取Celery工作者状态的信息，就像在这个tutorial中一样，但是你能走另一条路吗？在Celery工作人员启动后发送中断或内省？

我已经阅读了一些关于signals的内容，但要么不理解它们，要么它不是我想要的。可能两者都有。

背景

我正在使用Celery启动一个订阅MQTT主题的长期循环，我希望能够从我的Flask应用程序中的另一个端点关闭该进程/订阅。最好的方法是什么？还是一种方式？

示例代码

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工作人员启动后发送中断或内省？如何阻止从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对于你的用例可能有点过分，但我不完全确定你的最终目标是什么，所以我无法提供任何替代方案。