Python Asyncio无阻塞处理并行请求的线程安全方案问询

你的代码存在几个关键问题，导致异步逻辑无法正确并行执行：

1. 异步函数误用线程池：你把异步的method_C放到线程池里用asyncio.run执行，这是冗余操作，反而会额外创建事件循环，降低效率。
2. 未正确等待异步结果：method_B中没有await run_in_executor的返回值，导致method_A拿到的是Future对象而非实际结果。
3. 对await的误解：await只会阻塞当前协程，不会阻塞整个线程或其他请求——因为你的API框架（比如Flask）默认是多线程模式，每个请求会在独立线程中处理，线程间的事件循环互不干扰。

修正后的代码示例 #

ProcessingScript.py

from concurrent.futures import ThreadPoolExecutor
import asyncio

# 全局线程池，线程安全，控制并行阻塞任务数量
executor = ThreadPoolExecutor(max_workers=5)

def method_C(val):
    # 这里写阻塞的playwright同步调用（比如page.content()）
    return "processed:" + val

async def method_B(val):
    loop = asyncio.get_running_loop()
    # 将阻塞任务提交到线程池，await获取结果
    response = await loop.run_in_executor(executor, method_C, val)
    return response

def method_A(val):
    # 每个请求独立启动事件循环执行异步逻辑
    return asyncio.run(method_B(val))

API代码（Flask）

import ProcessingScript
from flask import Flask, request

app = Flask(__name__)

@app.route('/v1/generate', methods=['GET'])
def generate():
    data = request.args.get('data', type=str)
    # 直接调用修正后的method_A，返回处理后的字符串
    response = ProcessingScript.method_A(data)
    return response

进一步优化：使用异步API框架（如FastAPI） #

如果改用FastAPI这类异步框架，可以避免在同步函数中调用asyncio.run，进一步提升并行效率：

优化后的API代码

import ProcessingScript
from fastapi import FastAPI

app = FastAPI()

@app.get('/v1/generate')
async def generate(data: str):
    # 直接异步调用method_B，无需同步包装的method_A
    return await ProcessingScript.method_B(data)

对应简化的ProcessingScript.py

from concurrent.futures import ThreadPoolExecutor
import asyncio

executor = ThreadPoolExecutor(max_workers=5)

def method_C(val):
    # 阻塞的playwright同步操作
    return "processed:" + val

async def method_B(val):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, method_C, val)

关键说明 #

• 全局ThreadPoolExecutor是线程安全的，多个线程可以同时提交任务，线程池会自动调度并行执行。
• 每个请求在独立线程中处理时，asyncio.run会创建专属的事件循环，线程间的异步逻辑互不干扰，不会出现“所有线程等待第一个线程响应”的情况。
• 如果playwright支持异步API（它本身有async playwright），可以直接用异步方式调用，无需线程池，效率会更高：

  from playwright.async_api import async_playwright
  
  async def method_C(val):
      async with async_playwright() as p:
          browser = await p.chromium.launch()
          page = await browser.new_page()
          await page.goto(f"https://example.com/{val}")
          content = await page.content()
          await browser.close()
          return f"processed:{val}"
  
  async def method_B(val):
      return await method_C(val)
  

内容的提问来源于stack exchange，提问作者mang4521