使用 asyncio 并行化 Flask 视图中的工作

Question

我正在开发一个 Flask 应用程序，其中对客户端的响应取决于我从几个外部 API 获得的回复。对这些 API 的请求在逻辑上是相互独立的，因此可以通过并行发送这些请求来实现速度增益（在下面的示例中，响应时间将几乎减少一半）。

在我看来，实现这一目标的最简单和最现代的方法是使用 asyncio 并在一个单独的 async 函数中处理所有工作，该函数是使用 asyncio.run() 从烧瓶视图函数调用的。我在下面提供了一个简短的工作示例。

使用 celery 或任何其他类型的队列与单独的工作进程在这里实际上没有意义，因为响应必须等待 API 结果才能发送回复。据我所知this 是这个想法的一个变体，其中处理循环是通过 asyncio 访问的。当然有这方面的应用，但我认为如果我们真的只是想在响应请求之前并行化 IO，这会不必要地复杂。

但是，我知道在 Flask 中使用各种多线程可能存在一些缺陷。因此我的问题是：

1. 在生产环境中使用时，以下实施是否被认为是安全的？这如何取决于我们运行 Flask 的服务器类型？特别是内置的开发服务器或典型的多工人 gunicorn 设置，例如 https://flask.palletsprojects.com/en/1.1.x/deploying/wsgi-standalone/#gunicorn? 上建议的？
2. 对于 Flask 的应用程序和异步函数中的请求上下文是否有任何考虑，或者我可以像在任何其他函数中一样简单地使用它们吗？ IE。我可以简单地导入 current_app 来访问我的应用程序配置或使用 g 和 session 对象吗？在写信给他们时，显然必须考虑可能的竞争条件，但是还有其他问题吗？在我的基本测试中（不在示例中），一切似乎都正常。
3. 是否有其他解决方案可以对此进行改进？

这是我的示例应用程序。由于 ascynio 接口随着时间的推移发生了一些变化，因此可能值得注意的是，我在 Python 3.7 和 3.8 上对此进行了测试，并且我已尽力避免 asyncio 的已弃用部分。

import asyncio
import random
import time
from flask import Flask

app = Flask(__name__)

async def contact_api_a():
    print(f'{time.perf_counter()}: Start request 1')
    # This sleep simulates querying and having to wait for an external API
    await asyncio.sleep(2)

    # Here is our simulated API reply
    result = random.random()

    print(f'{time.perf_counter()}: Finish request 1')

    return result


async def contact_api_b():
    print(f'{time.perf_counter()}: Start request 2')
    await asyncio.sleep(1)

    result = random.random()

    print(f'{time.perf_counter()}: Finish request 2')

    return result


async def contact_apis():
    # Create the two tasks
    task_a = asyncio.create_task(contact_api_a())
    task_b = asyncio.create_task(contact_api_b())

    # Wait for both API requests to finish
    result_a, result_b = await asyncio.gather(task_a, task_b)

    print(f'{time.perf_counter()}: Finish both requests')

    return result_a, result_b


@app.route('/')
def hello_world():
    start_time = time.perf_counter()

    # All async processes are organized in a separate function
    result_a, result_b = asyncio.run(contact_apis())

    # We implement some final business logic before finishing the request
    final_result = result_a + result_b

    processing_time = time.perf_counter() - start_time

    return f'Result: {final_result:.2f}; Processing time: {processing_time:.2f}'

Answer 1

1. 这在生产环境中运行是安全的，但 asyncio 无法与 Gunicorn 异步工作器（例如 gevent 或 eventlet）一起有效地工作。这是因为result_a, result_b = asyncio.run(contact_apis()) 将阻塞 gevent/eventlet 事件循环直到它完成，而使用 gevent/eventlet spawn 等效项不会。 Flask 服务器不应该在生产中使用。 Gunicorn 线程工作者（或多个 Gunicorn 进程）会很好，因为 asyncio 会阻塞线程/进程。

2. 全局变量可以正常工作，因为它们被绑定到线程（线程工作者）或绿色线程（gevent/eventlet）而不是异步任务。

3. 我会说Quart 是一种改进（我是 Quart 作者）。 Quart 是使用 asyncio 重新实现的 Flask API。对于 Quart，上面的 sn-p 是，

import asyncio
import random
import time
from quart import Quart
    
app = Quart(__name__)
    
async def contact_api_a():
    print(f'{time.perf_counter()}: Start request 1')
    # This sleep simulates querying and having to wait for an external API
    await asyncio.sleep(2)

    # Here is our simulated API reply
    result = random.random()

    print(f'{time.perf_counter()}: Finish request 1')

    return result
    
async def contact_api_b():
    print(f'{time.perf_counter()}: Start request 2')
    await asyncio.sleep(1)

    result = random.random()

    print(f'{time.perf_counter()}: Finish request 2')

    return result
    

async def contact_apis():
    # Create the two tasks
    task_a = asyncio.create_task(contact_api_a())
    task_b = asyncio.create_task(contact_api_b())

    # Wait for both API requests to finish
    result_a, result_b = await asyncio.gather(task_a, task_b)

    print(f'{time.perf_counter()}: Finish both requests')

    return result_a, result_b
    
@app.route('/')
async def hello_world():
    start_time = time.perf_counter()

    # All async processes are organized in a separate function
    result_a, result_b = await contact_apis()

    # We implement some final business logic before finishing the request
    final_result = result_a + result_b

    processing_time = time.perf_counter() - start_time

    return f'Result: {final_result:.2f}; Processing time: {processing_time:.2f}'

我还建议使用基于 asyncio 的请求库，例如 httpx