企业官网建设流程全解析

Flask + Gevent 实战指南：greenlet 并发、asyncio 协同与 libuv 事件循环

【免费下载链接】flaskThe Python micro framework for building web applications.项目地址: https://gitcode.com/gh_mirrors/fl/flask

Gevent 通过修补（patch）Python 标准库，让 Flask 应用在不写任何async def/await的情况下获得处理大量长连接并发的能力。本文以 docs/gevent.rst 为骨架，结合本仓库 Flask 源码 中的async_to_sync、stream_with_context、copy_current_request_context实现与 Gunicorn / uWSGI 部署文档，完整讲解 gevent 的启用方式、视图内并发任务写法、与 asyncio 混合使用的边界，以及 libuv 事件循环的切换方法。读完本文你将掌握一套可直接落地的 "Flask + gevent" 高并发方案及其底层原理。

Gevent 是什么：绿色线程与标准库补丁

Gevent 将 Python 标准库修补为运行在名为greenlet（绿色线程）的特殊异步协程之上。它诞生于 Python 原生 asyncio 出现之前，而 Flask 从诞生之初就与它兼容（见 docs/gevent.rst 开篇说明）。

它的核心价值在于：以极低的改造代价实现与 ASGI / asyncio 相近的并发能力。你不需要在代码里写async def或await，一切依赖 gevent 与 greenlet 对 Python 解释器的底层操纵（monkey patch）来达成。这使得 gevent 成为处理大量、长生命周期的并发连接（例如长轮询、WebSocket 代理、流式响应场景）的可靠手段。

不过选择 gevent、Quart 还是其他方案，最终取决于你对项目具体需求的理解——gevent 不是async/await，也不是 ASGI 服务器规范，它是一条独立的技术路线。

启用 Gevent

尽早执行 monkey patch

启用 gevent 的第一步，是在你的代码尽可能早的位置应用 gevent 的补丁。这能让 gevent 底层的事件循环接管许多 Python 内部实现。在项目模块的顶部或顶层__init__.py中加入：

import gevent.monkey gevent.monkey.patch_all()

patch_all()会替换 socket、ssl、threading 等标准库模块，使阻塞调用自动让出控制权给事件循环。"尽可能早"是关键约束——如果补丁在模块导入之后才执行，某些已被导入的模块将无法被正确替换，并发能力会大打折扣。

生产部署：Gunicorn 的 gevent worker

当部署到生产环境时，官方文档（docs/deploying/gunicorn.rst）推荐使用 Gunicorn 并指定 gevent worker。Gunicorn 是纯 Python 实现的 WSGI 服务器，配置简单、无需额外编译依赖，且内置基于 gevent 的异步 worker。

安装方式：

$ cd hello-app $ python -m venv .venv $ . .venv/bin/activate $ pip install . # install your application $ pip install gunicorn

运行 Flask 应用（{module_import}:{app_variable}语法，也支持create_app()工厂函数调用）：

$ gunicorn -k gevent 'hello:create_app()' Starting gunicorn 20.1.0 Listening at: http://127.0.0.1:8000 (x) Using worker: gevent Booting worker with pid: x

关键参数：

• -k gevent：指定异步 worker 类型为 gevent（默认是sync，适合大多数常规场景）；
• -w 4：指定进程数，起步值可设为CPU * 2；
• --access-logfile=-：默认不打印每个请求的访问日志，此选项可输出到 stdout。

文档同时给出版本约束：使用 gevent 时要求greenlet>=1.0；使用 PyPy 时要求PyPy>=7.3.7。该约束在 docs/gevent.rst 的注释与 docs/deploying/gunicorn.rst 的 "Async with gevent" 一节中重复出现，是部署时务必检查的硬性前提。

生产部署：uWSGI 的 gevent 支持

uWSGI（docs/deploying/uwsgi.rst）同样提供基于 gevent 的异步 worker，通过--gevent参数指定并发连接数：

$ uwsgi --http 127.0.0.1:8000 --master --gevent 100 -w wsgi:app *** Starting uWSGI 2.0.20 (64bit) on [x] *** *** Operational MODE: async *** mounting hello:app on / spawned uWSGI master process (pid: x) spawned uWSGI worker 1 (pid: x, cores: 100) spawned uWSGI http 1 (pid: x) *** running gevent loop engine [addr:x] ***

注意输出中的 "Operational MODE: async" 与 "running gevent loop engine"——这说明 uWSGI 已切换为 gevent 事件循环引擎。若使用 app factory 模式，需要先写一个wsgi.py创建应用再指向它：

from hello import create_app app = create_app()

与 Gunicorn 一节相同，uWSGI 的 gevent 模式同样要求greenlet>=1.0（PyPy 下要求PyPy>=7.3.7）。

在视图内并发执行任务：gevent.spawn()

启用 gevent 后，你可以在自己的代码（例如视图函数）中通过gevent.spawn()并发执行任务。文档给出的典型场景是"发邮件"——把耗时的发送动作丢到后台 greenlet，立即返回响应：

@app.post("/send") def send_email(): gevent.spawn(email.send, to="example@example.example", text="example") return "Email is being sent."

后台任务如何访问 Flask 上下文

gevent.spawn()出的函数运行在新的 greenlet 中，默认访问不到request等 Flask 上下文全局变量。如果你确实需要在被 spawn 的函数里访问上下文，官方给出两个装饰器（见 docs/gevent.rst）：

• stream_with_context：位于 src/flask/helpers.py，通过 ContextVar_cv_app捕获当前应用上下文并重新进入（其实现会先执行一次next(wrapped_g)提前捕获上下文，再在每次迭代时with ctx:保持上下文存活），适用于响应生成器场景；
• copy_current_request_context：位于 src/flask/ctx.py，在装饰应用时保存当前激活的上下文，供后台任务重入。

copy_current_request_context的 docstring 里就自带一个与 gevent 联用的官方示例（src/flask/ctx.py）：

import gevent from flask import copy_current_request_context @app.route('/') def index(): @copy_current_request_context def do_some_work(): # do some work here, it can access flask.request or # flask.session like you would otherwise in the view function. ... gevent.spawn(do_some_work) return 'Regular response'

不过，文档与源码都给出了同样的建议：与其依赖上下文装饰器，不如在spawn时把真正需要的数据作为参数显式传递。这样更简单、也更安全（避免后台任务在请求结束后仍持有上下文带来的隐性依赖）。源码注释还特别指出：如果任务会访问session，父代码里也应先访问一次，以保证Vary: cookie头被正确设置；尽量避免在后台任务中修改session，因为它可能在响应 Cookie 写回之后才执行。

与 async/await 协同：覆盖 async_to_sync

Gevent 的 monkey patch 与 Flask 内建的 asyncio 支持不能很好地共存。如果你坚持要在同一应用里同时使用 gevent 和 asyncio，官方给出的方案是：覆盖Flask.async_to_sync方法，让 async 函数运行在 gevent 内部的 asyncio 事件循环上。

Flask 的 async 调度机制

先理解底层机制。Flask 2.0 起通过ensure_sync/async_to_sync支持 async 视图（src/flask/app.py）：

• ensure_sync检测函数是否为协程函数（iscoroutinefunction），是则交给async_to_sync包装，否则原样返回；
• async_to_sync默认实现依赖asgiref.sync.async_to_sync（需安装 Flask 的asyncextra，否则抛出RuntimeError）。

默认情况下，async 视图是在asgiref的事件循环策略中运行的。要把它换成 gevent 管理的 asyncio 事件循环，只需重写async_to_sync：

import gevent.monkey gevent.monkey.patch_all() import asyncio from flask import Flask, request loop = asyncio.EventLoop() gevent.spawn(loop.run_forever) class GeventFlask(Flask): def async_to_sync(self, func): def run(*args, **kwargs): coro = func(*args, **kwargs) future = asyncio.run_coroutine_threadsafe(coro, loop) return future.result() return run app = GeventFlask(__name__) @app.get("/") async def greet(): await asyncio.sleep(1) return f"Hello, {request.args.get("name", "World")}!"

这段代码的工作方式（docs/gevent.rst "Combining with async/await" 一节）：

1. 在 gevent worker 中启动一个 asyncio 事件循环（gevent.spawn(loop.run_forever)，由 gevent 负责调度）；
2. 重写async_to_sync，把协程通过asyncio.run_coroutine_threadsafe投递到该事件循环，并同步等待结果；
3. 于是async def视图中的await asyncio.sleep(1)可以在 gevent 进程内正确执行。

async_to_sync的 docstring 明确写着"Override this method to change how the app converts async code to be synchronously callable"（src/flask/app.py），这正是官方预留的扩展点，本示例即是对它的直接利用。文档同时提醒：该方案仍可能有局限，当使用其他 asyncio 实现时可能需要进一步修改。另外需注意原文档示例中的loop = asyncio.EventLoop()写法，实际运行时通常应使用asyncio.new_event_loop()创建事件循环，再调用loop.run_forever()。

一个容易踩的坑

由于 gevent 的 patch 与 Flask 内建 asyncio 支持的兼容性问题，如果你没有重写async_to_sync就同时使用两者，异步任务可能被调度到错误的事件循环或直接阻塞 greenlet。建议：要么纯 gevent（同步视图 +gevent.spawn），要么按上述方式显式接管async_to_sync，不要依赖默认行为。

切换 libuv 事件循环

gevent 默认使用基于 select/poll 的 libev 事件循环，但它也支持 libuv（一个跨平台的高性能事件库）。官方在 docs/gevent.rst 的 "libuv" 一节给出启用方法：必须在gevent.monkey.patch_all()之前、代码的最顶端设置配置：

import gevent gevent.config.loop = "libuv" import gevent.monkey gevent.monkey.patch_all()

需要注意的几点：

• 存在一个叫uvloop的项目，它把 libuv 引入 asyncio；但这里要使用 libuv 的话，应当用 gevent 自身的 libuv 支持，而不是 uvloop——uvloop 面向 asyncio，与 gevent 是两套体系；
• 理论上可以把上一节的async_to_sync重写代码进一步改造以配合 uvloop 工作，但目前并不确定可行，文档对此持保留态度；
• 顺序敏感：gevent.config.loop必须在patch_all()之前设置，否则事件循环已经初始化，切换不会生效。

小结：何时选择 Gevent

综合 docs/gevent.rst 与相关部署文档，给出如下决策要点：

• 需要大量、长生命周期的并发连接（长轮询、流式响应、后台任务密集），且不愿引入async def/await的代码改造时，gevent 是最低侵入的并发方案；
• 生产环境通过gunicorn -k gevent或uwsgi --gevent N启动，并确认满足greenlet>=1.0（PyPy 为>=7.3.7）的版本前提；
• 视图内的并发任务用gevent.spawn()，优先传参而非依赖上下文装饰器；确需上下文时用stream_with_context或copy_current_request_context；
• 需要 asyncio 时，重写Flask.async_to_sync让协程跑在 gevent 管理的事件循环上；需要 libuv 时在patch_all()前设置gevent.config.loop = "libuv"；
• 若项目重度依赖原生async/await生态，则应评估 Quart 或纯 ASGI 路线，而不是强行在 gevent 上叠加 asyncio。

进一步的参考材料：完整的生产部署配置见 docs/deploying/gunicorn.rst 与 docs/deploying/uwsgi.rst；Flask 原生 asyncio 支持的完整说明见 docs/async-await.rst；上下文复制与流式上下文的源码实现在 src/flask/ctx.py 与 src/flask/helpers.py。

【免费下载链接】flaskThe Python micro framework for building web applications.项目地址: https://gitcode.com/gh_mirrors/fl/flask