python使用flask+Thread实现异步任务

Posted ZH奶酪(张贺)

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python使用flask+Thread实现异步任务相关的知识,希望对你有一定的参考价值。

# 定义一个WorkerController,用于执行业务代码
class WorkerController(object):

    def __init__(self):
        pass

    def do_something(self, params):
        print("do something")
# 定义一个AsyncWorker,继承Thread,重写run方法,run方法中调用WorkerController的do_something方法

from threading import Thread
from worker_controller import WorkerController

WORKER_CONTROLLER = WorkerController()


class AsyncWorker(Thread):
    ASYNC_WORKER_INFO = dict()

    def __init__(self, params, ticket_id):
        Thread.__init__(self)
        self.params = params
        self.ticket_id = ticket_id
        self.daemon = True
        self.start()

    def run(self):
        AsyncWorker.ASYNC_WORKER_INFO[self.ticket_id] = {
            # 保存一些业务信息,之后轮询的时候,可以作为输出返回
            "some": "information",
            "status": "running"
        }
        WORKER_CONTROLLER.do_something(self.params)
# 开始任务
@app.route(\'/do-something\', methods=[\'GET\'])
def do_something():
    params = request.args.get(\'params\', \'\').strip()
    # 生成一个任务id,用于轮询,例如ticket id
    ticket_id = genearteMD5(str(int(round(time.time() * 1000))))
    # 实例化一个AsyncWorker
    AsyncWorker(params=params, ticket_id=ticket_id)
    # 将ticket_id返回
    return jsonify({"ticket_id": ticket_id})

# 轮询任务
@app.route("/check-status", methods=[\'GET\'])
def check_status():
    ticket_id = request.args.get("ticket_id")
    logs = AsyncWorker.ASYNC_WORKER_INFO[str(ticket_id)]
    return jsonify(logs)

如何在前端进行异步轮询呢?以angularjs的interval方法为例:https://www.cnblogs.com/CheeseZH/p/12444034.html

以上是关于python使用flask+Thread实现异步任务的主要内容,如果未能解决你的问题,请参考以下文章

Python Flask后端异步处理

python 使用gevent在Flask中进行异步请求

Flask入门邮件同步与异步发送

Flask 实现异步服务

python flask同步/异步性能对比

uWSGI, Thread, time.sleep 使用问题