【问题标题】：Long async calculation with callback in Python Flask WebservicePython Flask Webservice中带有回调的长异步计算
【发布时间】：2019-08-18 11:16:02
【问题描述】：

我想在 python 中实现一个 web 服务，其行为如下：

request to service: mysvc.com/doSomethingLong?callbackurl=http://callbackurl
response (immediate): 200 OK

（时间久了，python计算完毕）

service makes an http request to the received callback url: http://callbackurl

最好的方法是什么？

我看到的大多数异步计算示例，不立即返回“200 OK”，而是等待响应，同时让出控制权以允许其他代码并行工作。

【问题讨论】：

使用celery任务
感谢@RomanPerekhrest。如果“长计算”是 20 秒，你还会建议吗？我希望避免一个完整的任务队列服务。有没有其他选择？
你也可以考虑使用在主脚本结束后继续运行的子进程
这是一个网络服务（flask app），你说的“主脚本结束”是什么意思？简单地启动一个线程有什么缺点？（例如：threading.Thread(target=long_process, args=(params, callback_url))

标签： python flask python-asyncio

【解决方案1】：

ThreadPoolExecutor 是一种在 Flask 应用程序中运行异步任务并能够立即返回响应的解决方案。

class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())

Executor 子类，它使用最多 max_workers 个线程池来异步执行调用。

initializer 是一个可选的可调用对象，在每个工作线程开始时被调用； initargs 是传递给初始化器的参数元组。如果initializer 引发异常，所有当前待处理的作业都会引发BrokenThreadPool，以及任何向池提交更多作业的尝试。

在 3.5 版更改：如果 max_workers 为 None 或未给出，则默认为机器上的处理器数量，乘以 5，假设 ThreadPoolExecutor 经常用于重叠 I/O 而不是 CPU 工作，并且 worker 的数量应该高于 ProcessPoolExecutor 的 worker 数量。

3.6 版中的新功能：添加了 thread_name_prefix 参数以允许用户控制池创建的工作线程的 threading.Thread 名称，以便于调试。 p>

3.7 版更改：添加了 initializer 和 initargs 参数。

烧瓶示例

from concurrent.futures import ThreadPoolExecutor
from flask import Blueprint, request
from werkzeug.wrappers import BaseResponse as Response

client_session = Blueprint('client_session', __name__)


@client_session.route('/session-login', methods=['POST', 'PUT'])
def session_login():

    ...

    executor = ThreadPoolExecutor(5)
    executor.submit(my_long_running_task, my_task_param=42)

    # Return response immediately.
    return Response(
        response='{"status_text": "OK"}',
        status=200,
        mimetype='application/json; charset=UTF-8')

有关在 Flask 中使用 ThreadPoolExecutor 的预配置设计模式，请参阅 Flask-Executor。

Example from Docs

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
        'http://www.cnn.com/',
        'http://europe.wsj.com/',
        'http://www.bbc.co.uk/',
        'http://some-made-up-domain.com/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

【讨论】：

为什么每次访问路由时都要创建一个新的 ThreadPoolExecutor() 实例？我原以为你应该在 Flask 启动时创建一个 ThreadPoolExecutor() 实例，然后在处理路由时在该实例上调用 submit() ，不是吗？
@IanAsh 是的，ThreadPoolExecutor 的实例可以被缓存和重用。我通过引用包Flask-Executor 更新了答案，它提供了使用ThreadPoolExecutor 的设计模式。
在考虑了替代方案之后，我最初使用：x = threading.Thread(target, args)、x.start() 实现了一个基于线程的基本解决方案。此解决方案似乎是一种更安全的资源管理方式。公认。谢谢。