本文详解如何重构 Kafka 流式处理 + Playwright 调用场景，通过统一事件循环、合理使用 asyncio.run_in_executor 和协程调度，实现高并发、非阻塞、线程安全的异步 API 服务。
本文详解如何重构 kafka 流式处理 + playwright 调用场景，通过统一事件循环、合理使用 `asyncio.run_in_executor` 和协程调度，实现高并发、非阻塞、线程安全的异步 api 服务。
传统多线程混合 asyncio 的写法（如在同步函数中反复调用 asyncio.run()）不仅性能低下，还会引发事件循环冲突、资源泄漏和线程不安全问题——这正是原始代码中 method_A 被阻塞的根本原因：每次 asyncio.run() 都新建独立事件循环，无法共享 executor，也无法并发等待多个 run_in_executor 任务。
✅ 正确解法是：所有异步逻辑统一运行于单个顶层事件循环中，Kafka 消费、HTTP 响应、Playwright 调用全部纳入协程协作调度，仅将真正阻塞的 I/O 或 CPU 密集型操作（如 Playwright 同步 API）卸载至线程池。
✅ 关键重构原则
- 移除嵌套 asyncio.run()：method_A 不再是同步函数，而应为 async def；禁止在协程内调用 asyncio.run()。
- method_C 应为同步函数：Playwright 的 page.content() 等操作本质是阻塞调用，定义为 def method_C(val): ... 更合理（无需 async），由 run_in_executor 托管执行。
- 共用全局 ThreadPoolExecutor：在应用启动时创建一次，复用于所有 run_in_executor 调用，避免频繁创建销毁线程开销。
- Kafka 消费需异步化或桥接：原 for msg in kafka_topic 是同步阻塞迭代，应改用 aiokafka（推荐）或通过后台线程+asyncio.Queue 安全桥接。
? 示例：端到端重构代码结构
# ProcessingScript.py
from concurrent.futures import ThreadPoolExecutor
import asyncio
# ✅ 全局复用线程池（max_workers 根据 Playwright 实例数与内存权衡）
executor = ThreadPoolExecutor(max_workers=5)
# ✅ 同步函数：Playwright 实际执行逻辑（不可 async）
def method_C(val: str) -> str:
# 注意：此处需实际初始化 Playwright Browser 实例（建议复用）
# 示例伪代码：
# with sync_playwright() as p:
# browser = p.chromium.launch()
# page = browser.new_page()
# page.goto(f"https://example.com?q={val}")
# result = page.content()
# browser.close()
return f"processed:{val}"
# ✅ 协程函数：封装 executor 调用，支持 await
async def method_B(val: str) -> str:
return await asyncio.get_event_loop().run_in_executor(
executor, method_C, val
)
# ✅ 协程函数：对外暴露的主入口（不再用 asyncio.run）
async def method_A(val: str) -> str:
return await method_B(val)
# api_server.py（替代原 Flask 同步路由）
from flask import Flask, request
import asyncio
import threading
from ProcessingScript import method_A
app = Flask(__name__)
# ✅ 将 Flask 与 asyncio 集成：使用单独线程运行事件循环
loop = None
loop_thread = None
def start_event_loop():
global loop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_forever()
@app.route('/v1/generate', methods=['GET'])
def generate():
data = request.args.get('data', type=str)
if not data:
return {"error": "missing 'data' param"}, 400
# ✅ 在事件循环中调度协程并同步获取结果（非阻塞整个 Flask 线程）
future = asyncio.run_coroutine_threadsafe(method_A(data), loop)
try:
result = future.result(timeout=30) # 可设超时防 hang
return {"result": result}
except asyncio.TimeoutError:
return {"error": "timeout"}, 504
except Exception as e:
return {"error": str(e)}, 500
if __name__ == '__main__':
# ? 启动独立事件循环线程
loop_thread = threading.Thread(target=start_event_loop, daemon=True)
loop_thread.start()
app.run(host='0.0.0.0', port=5000)
⚠️ 重要注意事项
- Playwright 实例复用：method_C 中每次 launch() 开销极大。务必通过单例模式或连接池管理浏览器实例（如 playwright.sync_api.sync_playwright().start() 后复用 browser），否则线程池吞吐量将严重受限。
- Flask 并非原生异步框架：若追求极致性能，建议迁移到 FastAPI + uvicorn（原生支持 async def 路由），可直接 return await method_A(data)，无需线程桥接。
- Kafka 消费推荐 aiokafka：完全异步驱动，与事件循环无缝集成，避免线程队列桥接复杂度。
- 资源清理：ThreadPoolExecutor 应在应用退出时显式 shutdown(wait=True)；Playwright 浏览器需正确 close()，防止僵尸进程。
✅ 总结
真正的“非阻塞并行”不在于开多少线程，而在于让事件循环成为唯一调度中心：I/O 协作让出控制权，CPU/阻塞操作交由线程池并行执行，结果通过 await 自然汇入协程流。摒弃 asyncio.run() 的滥用、统一事件循环、精准区分同步/异步边界——这是构建可靠高并发 asyncio 服务的基石。
