在 Flask 中启动和停止线程

Question

我想知道在 Flask 中结束线程作业需要做些什么特别的事情吗？下面是示例代码：

from flask import (
    Flask, jsonify, render_template, request
)
import threading
import time
import random

app = Flask(__name__)

def task(thread_return):
    random_seconds=random.random()
    print(f'Start task {threading.current_thread().name} ')
    time.sleep(random_seconds*15)
    print(f'completed {threading.current_thread().name} ')
    thread_return['seconds']= random_seconds*5

thread_return = {'seconds': 0}

        
@app.route("/", methods=["GET", "POST"])
def index():
    if request.method == "GET":
        return render_template("index-ajax.html")

    val = request.json.get("c_check")
    t =threading.Thread(target=task, args=(thread_return,), daemon=True)
    if val==1:
        print("val1 is ", val)
        t.start()
    elif val==0:
        try:
            print('trying to kill thread')
            exit_event= threading.Event()
            exit_event.set()
        except Exception as e:
            print(e)
        print("val0 is ", val)
    
    
    
    return jsonify({"data": {"val": val}})
    
if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000, debug=True)

我尝试合并我在网上看到的几个将任务函数包装在一个类中的示例，但它不起作用，即：Exit Thread Gracefully

我喜欢这个其他示例，但我也无法让它与 Flask 一起使用： How Do You Kill a Python Thread

感谢您的帮助。

Answer 1

我尝试了多次重写并注意到以下工作。显然，我需要

1. 在路由函数之外声明 threading.event()
2. 在任务函数内部设置exit_event.set()，以结束任务或阻止任务运行
3. 并修改任务函数以更好地模拟长时间或无限期运行的任务

也许其他人有更优雅的 Pythonic 解决方案。

---

from flask import (
    Flask, jsonify, render_template, request
)
import threading
import time
import random

app = Flask(__name__)
def task(thread_return):
    i=0
    while True:
        if exit_event.is_set(): 
            print(f'end {threading.current_thread().name} ')
            return
        i+=1
        random_seconds=random.random()
        print(f'Running task {threading.current_thread().name} {i} sec elapsed')
        time.sleep(1)

exit_event= threading.Event()
        
@app.route("/", methods=["GET", "POST"])
def index():
    if request.method == "GET":
        return render_template("index-ajax.html")

    val = request.json.get("c_check")
    exit_event.clear()
    t =threading.Thread(target=task, args=(thread_return,), daemon=True)
    if val==1:
        print("val1 is ", val)
        t.start()
    elif val==0:
        try:
            print('trying to end thread')
            exit_event.set()
        except Exception as e:
            print(e)
        print("val0 is ", val)
    
    return jsonify({"data": {"val": val}})
    
if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000, debug=True)