烧瓶中的后台任务

Question

我正在编写一个 Web 应用程序，它会做一些繁重的工作。考虑到这一点，我考虑将任务作为后台任务（非阻塞），以便其他请求不会被先前的请求阻塞。

我继续妖魔化线程，以便在主线程（因为我正在使用threaded=True）完成后它不会退出，现在如果用户发送请求，我的代码将立即告诉他们他们的请求正在进行中（它将运行在后台）并且应用程序已准备好为其他请求提供服务。我当前的应用程序代码看起来像这样。

from flask import Flask
from flask import request
import threading

class threadClass:

    def __init__(self):
        thread = threading.Thread(target=self.run, args=())
        thread.daemon = True                       # Daemonize thread
        thread.start()                             # Start the execution

    def run(self):

         #
         # This might take several minutes to complete
         someHeavyFunction()

app = Flask(__name__)

@app.route('/start', methods=['POST'])
    try:
        begin = threadClass()
    except:
        abort(500)

    return "Task is in progress"

def main():
    """
    Main entry point into program execution

    PARAMETERS: none
    """
    app.run(host='0.0.0.0',threaded=True)

main()

我只是希望它能够处理一些并发请求（它不会在生产中使用）

我可以做得更好吗？我错过了什么吗？我正在浏览 python 的多线程包，发现了这个

multiprocessing 是一个包，它使用类似于线程模块的 API 支持生成进程。multiprocessing 包提供本地和远程并发，通过使用子进程而不是线程有效地绕过全局解释器锁。因此，多处理模块允许程序员充分利用给定机器上的多个处理器。它可以在 Unix 和 Windows 上运行。

我可以使用多处理来妖魔化一个进程吗？我怎样才能获得比线程模块更好的效果？

编辑

好的，所以我经历了 python 的多处理包，它类似于线程。

from flask import Flask
from flask import request
from multiprocessing import Process

class processClass:

    def __init__(self):
        p = Process(target=self.run, args=())
        p.daemon = True                       # Daemonize it
        p.start()                             # Start the execution

    def run(self):

         #
         # This might take several minutes to complete
         someHeavyFunction()

app = Flask(__name__)

@app.route('/start', methods=['POST'])
    try:
        begin = processClass()
    except:
        abort(500)

    return "Task is in progress"

def main():
    """
    Main entry point into program execution

    PARAMETERS: none
    """
    app.run(host='0.0.0.0',threaded=True)

main()

上面的方法看起来不错吗？

Answer 1

最佳实践

在 Flask 中实现后台任务的最佳方法是使用 Celery，如这篇 SO 帖子中所述。一个好的起点是官方Flask 文档和Celery 文档。

疯狂的方式：构建自己的装饰器

正如@MrLeeh 在评论中指出的那样，Miguel Grinberg 在他的Pycon 2016 演讲中通过实现装饰器提出了一个解决方案。我想强调的是，我对他的解决方案致以最崇高的敬意；他自己称其为“疯狂的解决方案”。下面的代码是他的解决方案的一个小改编。

警告！！！

不要在生产中使用它！主要原因是这个应用程序使用全局字典存在内存泄漏tasks。即使您修复了内存泄漏问题，维护此类代码也很困难。如果您只是想在私人项目中玩耍或使用它，请继续阅读。

最小的例子

假设您的/foo端点中有一个长时间运行的函数调用。我用一个 10 秒的sleep计时器来模拟这个。如果您调用端点 3 次，则需要 30 秒才能完成。

Miguel Grinbergs 装饰器解决方案在flask_async. 它在与当前 Flask 上下文相同的 Flask 上下文中运行一个新线程。每个线程都会发出一个新的task_id. 结果保存在全局字典中tasks[task_id]['result']。

装饰器就位后，您只需要装饰端点，@flask_async并且端点是异步的 - 就像那样！

import threading
import time
import uuid
from functools import wraps

from flask import Flask, current_app, request, abort
from werkzeug.exceptions import HTTPException, InternalServerError

app = Flask(__name__)
tasks = {}


def flask_async(f):
    """
    This decorator transforms a sync route to asynchronous by running it in a background thread.
    """
    @wraps(f)
    def wrapped(*args, **kwargs):
        def task(app, environ):
            # Create a request context similar to that of the original request
            with app.request_context(environ):
                try:
                    # Run the route function and record the response
                    tasks[task_id]['result'] = f(*args, **kwargs)
                except HTTPException as e:
                    tasks[task_id]['result'] = current_app.handle_http_exception(e)
                except Exception as e:
                    # The function raised an exception, so we set a 500 error
                    tasks[task_id]['result'] = InternalServerError()
                    if current_app.debug:
                        # We want to find out if something happened so reraise
                        raise

        # Assign an id to the asynchronous task
        task_id = uuid.uuid4().hex

        # Record the task, and then launch it
        tasks[task_id] = {'task': threading.Thread(
            target=task, args=(current_app._get_current_object(), request.environ))}
        tasks[task_id]['task'].start()

        # Return a 202 response, with an id that the client can use to obtain task status
        return {'TaskId': task_id}, 202

    return wrapped


@app.route('/foo')
@flask_async
def foo():
    time.sleep(10)
    return {'Result': True}


@app.route('/foo/<task_id>', methods=['GET'])
def foo_results(task_id):
    """
        Return results of asynchronous task.
        If this request returns a 202 status code, it means that task hasn't finished yet.
        """
    task = tasks.get(task_id)
    if task is None:
        abort(404)
    if 'result' not in task:
        return {'TaskID': task_id}, 202
    return task['result']


if __name__ == '__main__':
    app.run(debug=True)

但是，您需要一些技巧才能获得结果。端点/foo将只返回 HTTP 代码 202 和任务 id，而不返回结果。您需要另一个端点/foo/<task_id>来获得结果。这是本地主机的示例：

import time
import requests

task_ids = [requests.get('http://127.0.0.1:5000/foo').json().get('TaskId')
            for _ in range(2)]
time.sleep(11)
results = [requests.get(f'http://127.0.0.1:5000/foo/{task_id}').json()
           for task_id in task_ids]
# [{'Result': True}, {'Result': True}]