同步代码里怎么调用异步函数？两种代码怎么共存？

简化版

同步和异步互调是不对称的：从异步调同步很容易（await asyncio.to_thread(sync_fn)），从同步调异步很麻烦——因为 await 只能出现在 async def 里，而调用一个协程必须有事件循环来驱动它。这就是著名的「函数着色问题」：一个函数一旦变成 async，所有调用它的函数都得跟着变成 async，这种「传染性」会一路蔓延到调用链顶端。从同步调异步有三种正确姿势，取决于「当前线程有没有正在运行的事件循环」：① 没有 loop（普通脚本、CLI、同步 Web 框架）——用 asyncio.run(coro())，它会创建一个新 loop、跑完、再关闭；② 有 loop 但你在另一个线程里——用 asyncio.run_coroutine_threadsafe(coro, loop)，它返回一个 concurrent.futures.Future，可以 .result(timeout) 阻塞等待；③ 当前线程就在 loop 里——你根本不该问这个问题，直接 await 即可（在运行中的 loop 里再调 asyncio.run() 会抛 RuntimeError: This event loop is already running）。Django/asgiref 提供了封装好的 async_to_sync() 和 sync_to_async()，它们额外处理了上下文传递和「同一个线程执行」的语义（thread_sensitive=True），是混合项目里最省事的选择。最该避免的是 nest_asyncio——它通过打补丁让事件循环可重入，能让 Jupyter 里的代码「跑起来」，但会破坏 asyncio 的执行模型、引发难以排查的问题。核心记忆：异步调同步用 to_thread；同步调异步先问「当前线程有没有 loop」——没有就 asyncio.run，有但在别的线程就 run_coroutine_threadsafe，就在 loop 里就直接 await。

详细版

从同步调异步的三种情况：

当前线程状态	正确做法	注意
没有运行中的 loop	asyncio.run(coro())	每次创建/销毁 loop，开销大
loop 在另一个线程	asyncio.run_coroutine_threadsafe(coro, loop)	返回 concurrent.futures.Future
就在 loop 所在线程	✅ 直接 await	调 asyncio.run 会 RuntimeError
混合框架（Django）	asgiref.sync.async_to_sync(f)(args)	处理上下文与线程语义

import asyncio, threading, concurrent.futures

# ① ★同步 → 异步（无 loop）：asyncio.run★
def sync_main():
    result = asyncio.run(fetch("http://x"))     # ★创建 loop → 跑 → 关闭★
    return result
# ✗ 别在循环里反复调：每次都新建/销毁 loop（★几十毫秒开销 + 连接池全丢★）
# ✓ 一次 run 里跑完所有异步逻辑：
def sync_main_better(urls):
    async def all_work():
        return await asyncio.gather(*(fetch(u) for u in urls))
    return asyncio.run(all_work())              # ★只创建一次 loop★

# ② ★同步 → 异步（loop 在别的线程）：run_coroutine_threadsafe★
class AsyncBridge:
    """在后台线程里跑一个长期存活的事件循环"""
    def __init__(self):
        self.loop = asyncio.new_event_loop()
        self.thread = threading.Thread(target=self._run, daemon=True)
        self.thread.start()

    def _run(self):
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    def call(self, coro, timeout=30):
        fut = asyncio.run_coroutine_threadsafe(coro, self.loop)   # ★线程安全★
        return fut.result(timeout)               # ★concurrent.futures.Future★

    def close(self):
        self.loop.call_soon_threadsafe(self.loop.stop)
        self.thread.join()

bridge = AsyncBridge()
data = bridge.call(fetch("http://x"))            # ★同步代码里拿到异步结果★

# ③ ★异步 → 同步：to_thread（方向反过来就简单多了）★
async def handler():
    result = await asyncio.to_thread(blocking_io)        # ★3.9+★
    # 或 await loop.run_in_executor(pool, blocking_io)

# ④ ★asgiref：Django 生态的标准方案★
from asgiref.sync import async_to_sync, sync_to_async
sync_fetch = async_to_sync(fetch)                # ★异步函数 → 同步函数★
data = sync_fetch("http://x")

async_query = sync_to_async(Model.objects.get, thread_sensitive=True)
obj = await async_query(pk=1)                    # ★同步 ORM → 可 await★
# ★thread_sensitive=True：在同一个线程里执行（Django ORM 的连接是 thread-local）

# ⑤ ★错误示范★
async def wrong():
    return asyncio.run(other())          # ✗ RuntimeError: already running
def wrong2():
    loop = asyncio.get_event_loop()      # ★3.12+ 无运行 loop 时会 DeprecationWarning★
    return loop.run_until_complete(f())  # 老写法，容易出问题

# ⑥ 从异步里安全地调"可能是同步也可能是异步"的函数
import inspect
async def call_any(fn, *a, **kw):
    if inspect.iscoroutinefunction(fn):
        return await fn(*a, **kw)
    return await asyncio.to_thread(fn, *a, **kw)   # ★同步的丢线程池★

⚠️ 三个必须记住的点：① asyncio.run() 不能在已经运行的事件循环里调用——会抛 RuntimeError: asyncio.run() cannot be called from a running event loop。这是最常见的错误，出现在「在 async 函数里想调一个封装了 asyncio.run 的同步库函数」的场景。判断方法是 asyncio.get_running_loop()（有 loop 就返回，没有就抛 RuntimeError）——但更好的做法是在设计时就明确「这个函数是同步的还是异步的」，而不是运行时猜。② asyncio.run() 每次都会创建并销毁一个新的事件循环：不仅有几十毫秒的开销，更要命的是连接池、DNS 缓存、aiohttp.ClientSession 这些绑定在 loop 上的资源全都会被丢弃——在循环里反复调用 asyncio.run(fetch(url)) 会让每个请求都重建连接，性能比同步版本还差。正确做法是把所有异步逻辑收拢到一次 asyncio.run() 里。③ run_coroutine_threadsafe 返回的是 concurrent.futures.Future 而不是 asyncio.Future：它的 .result(timeout) 是阻塞的（适合在同步线程里调用），而且它是唯一线程安全地向事件循环提交协程的方式——从别的线程直接调 loop.create_task() 是不安全的（asyncio 的绝大多数 API 都不是线程安全的，只有 call_soon_threadsafe 和 run_coroutine_threadsafe 例外）。

完整版教学

一、函数着色：为什么互调这么麻烦

★ 核心不对称：
  异步 → 同步：★容易★
    async def h():
        await asyncio.to_thread(sync_fn)     # 丢到线程池，不阻塞循环
  同步 → 异步：★麻烦★
    def sync_fn():
        result = fetch()      # ✗ fetch() 只是创建了协程对象，什么都没执行
        result = await fetch()# ✗ SyntaxError：await 只能在 async def 里

★ 为什么：协程需要"驱动者"
  一个协程对象本身★什么都不做★——它需要事件循环反复调用它的 send()
  才能推进到下一个 await 点。
  同步函数里没有事件循环在跑 → ★没人驱动它★
  → 所以必须"借"一个 loop：新建一个（asyncio.run）
    或把协程送到别的线程的 loop 里（run_coroutine_threadsafe）

★ "函数着色问题"（function coloring）：
  一个函数变成 async → 调用它必须 await → 调用者也得是 async
  → ★传染性一路蔓延到调用链顶端★
    async def db_query()           ← 改成异步
      ↑ async def get_user()       ← 被迫改
        ↑ async def handle()       ← 被迫改
          ↑ async def main()       ← 被迫改
  → 这就是为什么"给老项目加异步"往往变成大规模重构

★ 三种应对策略：
  ① ★全异步★：整条链路都改成 async（最干净，但改造成本高）
  ② ★全同步 + 线程池★：不用 asyncio，用 ThreadPoolExecutor 处理并发
     （对中等并发完全够用，★复杂度低得多★）
  ③ ★边界隔离★：在明确的边界上做转换（网关层异步、内部同步，或反过来）
     → ★最现实的做法★

★ 一个重要的判断：你真的需要 asyncio 吗？
  asyncio 的价值在"★海量并发连接★"（几千到几十万）
  → QPS 几百、并发几十的服务，★线程池完全够用且简单得多★
  → 不要为了"看起来现代"而引入异步（★着色问题的成本很实在★）

理解互调要先看核心的不对称：异步调同步容易（await asyncio.to_thread(sync_fn) 丢到线程池即可），同步调异步麻烦——因为协程对象本身什么都不做，它需要事件循环反复调用 send() 来驱动，而同步函数里没有正在运行的 loop。这引出了著名的**「函数着色问题」**：一个函数变成 async 后，调用它必须 await，于是调用者也得是 async——传染性一路蔓延到调用链顶端，这正是「给老项目加异步」常常变成大规模重构的原因。三种应对策略：全异步（最干净但成本高）、全同步 + 线程池（对中等并发完全够用且复杂度低得多）、边界隔离（在明确的边界上做转换，最现实）。这里还有个值得反思的判断：asyncio 的价值在于「海量并发连接」（几千到几十万），QPS 几百、并发几十的服务用线程池完全够用且简单得多——不要为了「看起来现代」而引入异步，着色问题的成本很实在。

二、同步调异步：先问「当前线程有没有 loop」

★ 决策树（★这是本题的核心★）：
  ┌─────────────────────────────────────────────────────┐
  │ 当前线程有正在运行的事件循环吗？                        │
  │   （asyncio.get_running_loop() 不抛异常就是有）        │
  ├──────────────┬──────────────────────────────────────┤
  │ ★没有★       │ ★asyncio.run(coro())★               │
  │              │ （脚本、CLI、同步 Web 框架的视图）       │
  ├──────────────┼──────────────────────────────────────┤
  │ ★有，但在别的│ ★asyncio.run_coroutine_threadsafe(   │
  │  线程里★     │    coro, loop).result(timeout)★      │
  ├──────────────┼──────────────────────────────────────┤
  │ ★有，就是当前│ ★直接 await★                         │
  │  这个线程★   │ （调 asyncio.run 会 RuntimeError）    │
  └──────────────┴──────────────────────────────────────┘

★ 情况一：asyncio.run（没有 loop 时）
  def main():
      return asyncio.run(async_main())
  ★ 它做的事：新建 loop → 运行协程 → 取消剩余任务 → 关闭异步生成器 → 关闭 loop
  ★ 三个注意：
    ① ★不能嵌套★（运行中的 loop 里调用会 RuntimeError）
    ② ★每次都新建/销毁 loop★：几十毫秒开销 + ★loop 相关的资源全丢★
       （aiohttp.ClientSession、连接池、DNS 缓存都绑在 loop 上）
    ③ ★一个进程里反复 asyncio.run 是反模式★
       ✗ for url in urls: asyncio.run(fetch(url))     # ★每次重建连接★
       ✓ asyncio.run(fetch_all(urls))                  # ★只建一次 loop★

  ★ 3.11+ 的 asyncio.Runner：可以复用 loop
    with asyncio.Runner() as runner:
        r1 = runner.run(task1())
        r2 = runner.run(task2())      # ★同一个 loop，资源可复用★

★ 情况二：run_coroutine_threadsafe（loop 在别的线程）
  典型场景：同步的主程序 + 后台线程里跑着一个长期存活的 loop
  fut = asyncio.run_coroutine_threadsafe(coro, loop)
  result = fut.result(timeout=30)     # ★阻塞等待，返回 concurrent.futures.Future★
  ★ 关键点：
    - ★这是唯一线程安全的"向 loop 提交协程"的方式★
    - ★asyncio 的其他 API 都不是线程安全的★
      （从别的线程调 loop.create_task() → ★未定义行为★）
    - 另一个线程安全的 API 是 loop.call_soon_threadsafe(callback)
    - fut.result() 会阻塞调用线程 → ★别在事件循环线程里调它（自锁）★

★ 情况三：已经在 loop 里 → 直接 await
  ✗ async def h(): return asyncio.run(g())    # RuntimeError
  ✓ async def h(): return await g()
  ★ 如果 g 是"同步函数内部封装了 asyncio.run"的第三方库 →
    ✓ await asyncio.to_thread(g)     # ★丢到线程里，那里没有运行中的 loop★

★ 判断当前状态的正确方式：
  try:
      loop = asyncio.get_running_loop()    # ★3.7+，推荐★
      in_loop = True
  except RuntimeError:
      in_loop = False
  ★ 不要用 asyncio.get_event_loop()：
    3.10 起在没有运行 loop 时会 DeprecationWarning，
    ★3.12 起行为进一步收紧★，语义混乱（有时创建新的、有时报错）

同步调异步的关键是先问「当前线程有没有正在运行的事件循环」，三种情况对应三种做法。没有 loop 时用 asyncio.run()——但要注意它每次都新建并销毁 loop，不仅有几十毫秒开销，更要命的是 aiohttp.ClientSession、连接池、DNS 缓存这些绑在 loop 上的资源全都会丢失，所以「在 for 循环里反复 asyncio.run(fetch(url))」是典型反模式，正确做法是把所有异步逻辑收进一次 run（3.11+ 还可以用 asyncio.Runner 复用 loop）。loop 在别的线程时用 run_coroutine_threadsafe——它是唯一线程安全地向事件循环提交协程的方式（asyncio 的其他 API 都不是线程安全的，从别的线程调 loop.create_task() 是未定义行为），返回的是 concurrent.futures.Future、.result(timeout) 会阻塞调用线程。已经在 loop 里就直接 await；如果要调的是「内部封装了 asyncio.run 的同步库函数」，就 await asyncio.to_thread(g) 把它丢到线程里执行。最后记住判断状态要用 asyncio.get_running_loop()，不要用语义混乱的 get_event_loop()。

三、在后台线程里跑事件循环

★ 场景：主程序是同步的（如 GUI、老框架、脚本），但要用异步库
  → 在后台线程里启动一个★长期存活★的事件循环，同步代码通过桥接调用

  class AsyncBridge:
      def __init__(self):
          self._loop = asyncio.new_event_loop()
          self._ready = threading.Event()
          self._thread = threading.Thread(target=self._runner, daemon=True)
          self._thread.start()
          self._ready.wait()                 # ★等 loop 真正跑起来★

      def _runner(self):
          asyncio.set_event_loop(self._loop)
          self._loop.call_soon(self._ready.set)
          self._loop.run_forever()           # ★一直跑，等待被提交协程★

      def run(self, coro, timeout=None):
          fut = asyncio.run_coroutine_threadsafe(coro, self._loop)
          return fut.result(timeout)         # ★阻塞当前（同步）线程★

      def submit(self, coro):
          return asyncio.run_coroutine_threadsafe(coro, self._loop)  # ★不阻塞★

      def close(self):
          self._loop.call_soon_threadsafe(self._loop.stop)   # ★线程安全地停★
          self._thread.join()
          self._loop.close()

★ 好处：
  ✓ ★loop 长期存活★ → 连接池、session、DNS 缓存都能复用
  ✓ 同步代码可以多次调用，不用反复建销 loop
  ✓ 可以并发提交多个协程（submit 不阻塞）

★ 五个注意事项：
  ① ★启动时要等 loop 真正 running★（用 Event 同步），
     否则 run_coroutine_threadsafe 可能在 loop 还没跑起来时提交
  ② ★所有跨线程操作只能用两个 API★：
     run_coroutine_threadsafe / call_soon_threadsafe
  ③ ★asyncio 的同步原语（Lock/Queue）绑定在 loop 上★
     → 别在同步线程里直接操作它们
  ④ ★关闭要用 call_soon_threadsafe(loop.stop)★，
     不能直接 loop.stop()（不是线程安全的）
  ⑤ daemon=True 只是兜底，★仍应显式 close★（否则连接不会正常关闭）

★ 什么时候用这个方案：
  ✓ 同步框架（Flask/Django WSGI）里要调异步 SDK
  ✓ GUI 程序（PyQt/Tkinter）主循环是同步的
  ✓ Jupyter 之外的交互式场景
  ✗ ★如果整个程序可以是异步的，就别绕这一圈★

★ 反向：在异步程序里跑同步代码（已有专题，这里只强调边界）
  await asyncio.to_thread(fn, *args)              # ★3.9+，会传 contextvars★
  await loop.run_in_executor(pool, fn, *args)     # ★不传 contextvars★
  ★ CPU 密集要用 ProcessPoolExecutor（线程池挡不住 GIL）
  ★ 线程池大小有限（默认 min(32, cpu+4)），大量阻塞调用会排队

当主程序是同步的（GUI、老框架、脚本）但要用异步库时，标准方案是在后台线程里跑一个长期存活的事件循环。它的最大好处是 loop 长期存活，连接池、session、DNS 缓存都能复用——避免了反复 asyncio.run 的开销。实现有五个注意点：① 启动时要用 Event 等 loop 真正 running（否则提交的协程可能丢失）；② 所有跨线程操作只能用 run_coroutine_threadsafe 和 call_soon_threadsafe 这两个 API；③ asyncio 的 Lock/Queue 绑定在 loop 上，别在同步线程里直接操作；④ 关闭必须用 call_soon_threadsafe(loop.stop)（直接 loop.stop() 不是线程安全的）；⑤ 即使设了 daemon=True 也应显式 close（否则连接不会正常关闭）。反方向（异步里跑同步代码）已有专题，这里只强调两个边界：to_thread 会传递 contextvars 而 run_in_executor 不会，以及 CPU 密集必须用进程池（线程池挡不住 GIL）。

四、asgiref：Django 生态的标准答案

★ asgiref 提供两个转换器（Django 内部大量使用）：

  ① ★async_to_sync(async_fn)★ → 得到一个同步函数
     from asgiref.sync import async_to_sync
     result = async_to_sync(fetch)("http://x")
     ★ 内部逻辑：
       - 当前没有 loop → 新建一个跑（类似 asyncio.run）
       - 当前★已经在 loop 里★ → ★在一个新线程里跑新 loop★（避免 RuntimeError）
       - ★传递 contextvars★（比裸 asyncio.run 更完善）

  ② ★sync_to_async(sync_fn, thread_sensitive=True)★ → 得到一个异步函数
     from asgiref.sync import sync_to_async
     obj = await sync_to_async(Model.objects.get)(pk=1)
     ★ thread_sensitive 的含义（★重点★）：
       True（默认）：★所有调用在同一个线程里执行★
         → 因为 Django ORM 的数据库连接是 ★thread-local★ 的，
           如果每次都在不同线程执行，会创建大量连接、事务也会错乱
       False：丢到普通线程池（★可以并行，但不能用 thread-local 的资源★）

★ 为什么 Django 需要这套：
  Django 同时支持 WSGI（同步）和 ASGI（异步）
  → 异步视图里调用同步 ORM → sync_to_async
  → 同步中间件里调用异步组件 → async_to_sync
  → ★asgiref 帮你处理了上下文、线程亲和性、异常传播★

★ Django 里的实际用法：
  # 异步视图里用同步 ORM
  async def view(request):
      user = await sync_to_async(User.objects.get)(pk=1)
      # ★Django 4.1+ 有原生异步 ORM：await User.objects.aget(pk=1)★
      return JsonResponse({...})

  # 同步代码里调异步
  from asgiref.sync import async_to_sync
  async_to_sync(channel_layer.send)("group", {"type": "msg"})

★ 三个坑：
  ① ★thread_sensitive=False 时不能碰 thread-local 资源★
     （Django 连接、事务、request 上下文）
  ② ★async_to_sync 在已有 loop 时会新起一个线程 + 新 loop★
     → 不能共享原 loop 的资源（连接池等）→ ★有性能代价★
  ③ ★嵌套转换会层层起线程★：sync_to_async(async_to_sync(...)) → 性能灾难
     → ★设计上避免来回横跳★

★ 与裸 asyncio 的对比：
  ┌──────────────────────┬──────────────────────────────────┐
  │ asyncio.run          │ 简单；★不能嵌套★；不传 contextvars │
  │ ★asgiref★           │ ★能嵌套（起新线程）★+ 传上下文     │
  │                      │ + thread_sensitive 语义           │
  │ run_coroutine_threadsafe│ 需要自己维护一个后台 loop        │
  └──────────────────────┴──────────────────────────────────┘
  → ★Django 项目用 asgiref；纯 asyncio 项目用原生 API★

asgiref 是 Django 生态的标准答案，提供两个转换器。async_to_sync(async_fn) 把异步函数变成同步函数——它比裸 asyncio.run 完善的地方在于：当前已经在 loop 里时，它会在一个新线程里跑新 loop（从而避免 RuntimeError），并且会传递 contextvars。sync_to_async(sync_fn, thread_sensitive=True) 把同步函数变成可 await 的——thread_sensitive=True（默认）意味着所有调用在同一个线程里执行，这是因为 Django ORM 的数据库连接是 thread-local 的，每次换线程会创建大量连接、事务也会错乱。三个坑要注意：thread_sensitive=False 时不能碰 thread-local 资源、async_to_sync 在已有 loop 时会新起线程和新 loop（无法共享原 loop 的连接池，有性能代价）、以及嵌套转换会层层起线程（sync_to_async(async_to_sync(...)) 是性能灾难）——设计上要避免同步异步来回横跳。

五、常见陷阱

★ 陷阱一：在运行中的 loop 里调 asyncio.run★
  async def handler():
      data = legacy_sync_api()      # 它内部调了 asyncio.run
  → RuntimeError: asyncio.run() cannot be called from a running event loop
  ✓ await asyncio.to_thread(legacy_sync_api)   # ★丢到线程，那里没有运行的 loop★

★ 陷阱二：Jupyter 里的 "event loop is already running"★
  Jupyter/IPython ★本身就跑在事件循环里★（为了支持 %autoawait）
  → 在 cell 里调 asyncio.run() 会报错
  ✓ ★直接在 cell 里 await★（Jupyter 支持顶层 await）
  ✓ 或 asyncio.get_running_loop().create_task(...)
  ✗ ★nest_asyncio★（打补丁让 loop 可重入）
    → 能"跑起来"，但★破坏了 asyncio 的执行模型★：
      重入的循环会打乱任务调度、超时和取消语义可能失效、
      与某些库（uvloop、aiohttp）不兼容、★问题极难排查★
    → ★只在临时探索时用，绝不进生产★

★ 陷阱三：反复 asyncio.run 导致性能崩塌★
  ✗ def fetch_all(urls):
        return [asyncio.run(fetch(u)) for u in urls]
     → ★每次新建 loop + 新建连接 + TLS 握手★
     → 比同步版本还慢，且完全没有并发
  ✓ asyncio.run(gather_all(urls))

★ 陷阱四：跨 loop 使用 asyncio 对象★
  loop1 里创建的 asyncio.Lock / Queue / Future
  → 在 loop2 里 await → ★RuntimeError 或未定义行为★
  ★ 3.10 起同步原语移除了 loop 参数，但★仍然绑定到"首次使用时的 loop"★
  ✓ 在协程内部创建，或每个 loop 一套

★ 陷阱五：从别的线程直接操作 loop★
  ✗ loop.create_task(coro)        # ★不是线程安全的★
  ✗ loop.stop()
  ✓ asyncio.run_coroutine_threadsafe(coro, loop)
  ✓ loop.call_soon_threadsafe(loop.stop)

★ 陷阱六：库同时提供同步和异步 API 时选错★
  requests（同步）vs aiohttp/httpx（异步）
  → ★在协程里用 requests = 阻塞整个事件循环★
  ✓ 异步项目里统一用异步客户端；
    实在没有异步版本 → to_thread 包一层

★ 陷阱七：忘记 loop 生命周期，资源没关★
  asyncio.run 结束时会关闭 loop → 之后再用绑定在它上面的 session
  → RuntimeError: Event loop is closed
  ✓ 用 async with 管理 session，在同一个 run 内完成所有工作

★ 陷阱八：signal handler 里调异步★
  signal 处理器是同步的、可能在任意字节码之间执行
  ✓ 用 loop.add_signal_handler（Unix）
  ✓ 或在 handler 里 loop.call_soon_threadsafe(event.set)

八个陷阱里最常见的三个：① 在运行中的 loop 里调 asyncio.run（解法是 await asyncio.to_thread(那个同步函数)，因为线程里没有运行中的 loop）；② Jupyter 里的 “event loop is already running”——Jupyter 本身就跑在事件循环里，正确做法是直接在 cell 里 await（它支持顶层 await），而**nest_asyncio 虽然能让代码跑起来，但它通过打补丁让循环可重入、破坏了 asyncio 的执行模型**（打乱任务调度、超时和取消语义可能失效、与 uvloop 等不兼容），只适合临时探索、绝不能进生产；③ 反复 asyncio.run 导致性能崩塌（每次新建 loop、新建连接、重做 TLS 握手，比同步版本还慢且毫无并发）。其余几个：跨 loop 使用 asyncio 对象（Lock/Queue 绑定到首次使用时的 loop）、从别的线程直接调 loop.create_task()（不是线程安全的）、在协程里用 requests（直接阻塞事件循环）、以及 asyncio.run 结束后再用绑定在旧 loop 上的 session。

六、渐进式迁移策略

★ 现实问题：老项目全是同步代码，想用异步怎么办？

★ 策略一：★边界隔离（推荐）★
  在明确的边界上转换，内部保持单一风格：
    方案 A：外层异步 + 内层同步
      异步 Web 框架（FastAPI）接收请求
      → 同步的业务逻辑用 ★await asyncio.to_thread(...)★ 包一层
      → 好处：能享受异步框架的高并发接入能力，业务代码不用改
      → 代价：线程池大小限制了实际并发（★默认 min(32, cpu+4)，要调大★）
    方案 B：外层同步 + 内层异步
      同步框架里用 ★后台 loop + run_coroutine_threadsafe★
      → 适合"只有某几个下游调用需要高并发"的场景

★ 策略二：★从叶子往根改（自底向上）★
  先把最底层的 IO 封装（HTTP 客户端、DB 访问）改成异步
  → 再逐层向上，每层用 to_thread/sync_to_async 临时桥接
  → ★好处：每一步都可运行、可回滚★
  ✗ 反向（从入口往下改）会导致一大段时间里系统跑不起来

★ 策略三：★双 API（库作者的做法）★
  同时提供 fetch() 和 afetch()，共享底层实现
  → httpx、redis-py、SQLAlchemy 都是这么做的
  ✗ 维护成本翻倍；★别在业务代码里这么干★

★ 策略四：★干脆不迁移★
  先问：★瓶颈真的是并发连接数吗？★
  - QPS 几百、并发几十 → ★线程池完全够用★，改异步收益很小
  - 瓶颈在 CPU / 数据库 / 下游 → ★改异步一点用没有★
  - 团队不熟悉 asyncio → ★引入的 bug 可能比收益多★
  → ★"能不迁移就不迁移"是很多情况下的正确答案★

★ 迁移中的检查清单：
  □ 边界在哪里？（★写下来，团队达成一致★）
  □ 同步侧调用异步：用什么方式？（asyncio.run / 后台 loop / asgiref）
  □ 异步侧调用同步：★线程池大小够吗？CPU 密集是不是要进程池？★
  □ ★有没有来回横跳★（sync→async→sync→async，性能灾难）
  □ 上下文（trace_id、租户）怎么跨边界传递？（★contextvars + to_thread★）
  □ 事务、连接这些 thread-local 资源怎么处理？（thread_sensitive）
  □ 超时、取消语义在边界上还成立吗？（★线程里的任务取消不了★）
  □ 测试：同步侧和异步侧都要覆盖

★ 一个务实的总结：
  ★"同步和异步共存的成本很高（着色、线程池、上下文、取消语义都要处理），
    所以最好的策略是：要么整条链路统一，要么把转换收敛到一两个明确的边界上；
    最糟的是让 sync/async 在业务代码里来回横跳。"★

老项目迁移有四种策略。边界隔离是最推荐的：要么「外层异步 + 内层同步」（FastAPI 接请求、业务逻辑用 to_thread 包一层，代价是线程池大小限制了实际并发，默认 min(32, cpu+4) 需要调大），要么「外层同步 + 内层异步」（同步框架里用后台 loop）。从叶子往根改（自底向上）比反过来好，因为每一步都可运行、可回滚。双 API 是库作者的做法（httpx、redis-py 都提供两套），业务代码别这么干。而第四种策略——干脆不迁移——常常是正确答案：先问「瓶颈真的是并发连接数吗」，QPS 几百、并发几十的服务用线程池完全够用，瓶颈在 CPU 或数据库时改异步一点用没有。检查清单里最容易忽略的三条：有没有来回横跳（性能灾难）、上下文怎么跨边界传递（contextvars + to_thread）、取消语义在边界上还成立吗（丢进线程的任务是取消不了的）。

记忆钩子：「同步和异步互调是★不对称★的：异步调同步容易（await asyncio.to_thread(fn)），★同步调异步麻烦★——因为协程对象本身什么都不做，★需要事件循环反复 send() 来驱动★，而同步函数里没有 loop。这就是★函数着色问题★：一个函数变 async，调用者被迫全变 async，传染到调用链顶端。★同步调异步的决策树只有一条：先问『当前线程有没有正在运行的 loop』★（用 asyncio.get_running_loop() 判断，★别用语义混乱的 get_event_loop()★）——①★没有 loop★ → asyncio.run(coro())，但★它每次新建销毁 loop、连接池/session/DNS 缓存全丢★，所以『for 循环里反复 asyncio.run』是反模式（比同步还慢），要把异步逻辑收进一次 run（3.11+ 可用 asyncio.Runner 复用 loop）；②★loop 在别的线程★ → asyncio.run_coroutine_threadsafe(coro, loop).result(timeout)，★这是唯一线程安全地向 loop 提交协程的方式★（asyncio 其他 API 都不线程安全，另一个例外是 call_soon_threadsafe）；③★就在 loop 里★ → 直接 await（调 asyncio.run 会 RuntimeError；如果要调的是『内部封装了 asyncio.run 的同步库』，就 await to_thread(它)）。★Django 生态用 asgiref★：async_to_sync（已有 loop 时会★新起线程跑新 loop★，还会传 contextvars）和 sync_to_async（★thread_sensitive=True 让所有调用在同一线程，因为 Django ORM 连接是 thread-local★）。★nest_asyncio 绝不能进生产★——它打补丁让循环可重入，会破坏调度、超时和取消语义；Jupyter 里直接用顶层 await 即可。迁移策略：★边界隔离★（外层异步+内层 to_thread，注意线程池默认只有 min(32,cpu+4)）、自底向上改、或者★先问『瓶颈真的是并发连接数吗』——QPS 几百的服务线程池完全够用★。最糟的是让 sync/async 在业务代码里★来回横跳★。」

七、常见误区与追问

• 误区：在同步函数里调用异步函数，写成 result = async_fn() 就行。 调用一个 async def 函数只会返回一个协程对象，函数体一行都不会执行——它需要事件循环反复调用 send() 才能推进。所以这样写的结果是：业务逻辑完全没跑，而且 Python 会在协程对象被 GC 时发出 RuntimeWarning: coroutine 'async_fn' was never awaited（如果没开 -W error 很容易被忽略）。正确做法取决于当前线程有没有运行中的事件循环：没有就 asyncio.run(async_fn())、有但在别的线程就 asyncio.run_coroutine_threadsafe(...)、就在 loop 里则说明你的函数应该改成 async def 并直接 await。
• 误区：asyncio.run() 可以随便调，反正它会帮我管理事件循环。 有两个硬约束。① 不能嵌套：在已经运行的事件循环里调用会抛 RuntimeError: asyncio.run() cannot be called from a running event loop——这在「async 视图里调用了一个内部封装了 asyncio.run 的第三方同步库」时经常发生（解法是 await asyncio.to_thread(那个函数)，把它丢到没有 loop 的线程里）。② 每次都创建并销毁一个新的事件循环：几十毫秒的固定开销还是小事，真正致命的是所有绑定在 loop 上的资源都会被丢弃——aiohttp.ClientSession、连接池、DNS 缓存、SSL 会话复用全没了。所以 for url in urls: asyncio.run(fetch(url)) 这种写法会给每个请求重建连接和重做 TLS 握手，比纯同步版本还慢，而且完全没有并发。正确做法是把所有异步逻辑收进一次 asyncio.run(gather_all(urls))。
• 误区：从其他线程可以直接调用 loop.create_task() 把协程提交给事件循环。 asyncio 的绝大多数 API 都不是线程安全的，从非 loop 线程调用 loop.create_task()、loop.stop()、直接操作 asyncio.Queue 都是未定义行为——可能表现为任务丢失、数据竞争、或者难以复现的崩溃。整个 asyncio 只有两个明确线程安全的入口：asyncio.run_coroutine_threadsafe(coro, loop)（提交协程，返回 concurrent.futures.Future，可以 .result(timeout) 阻塞等待）和 loop.call_soon_threadsafe(callback)（提交一个同步回调）。停止后台 loop 也必须写成 loop.call_soon_threadsafe(loop.stop)。反过来也要注意：fut.result() 会阻塞调用线程，绝不能在事件循环线程里调用（那会自锁）。
• 误区：Jupyter 里报「event loop is already running」，装个 nest_asyncio 就解决了。 nest_asyncio 通过给事件循环打补丁使其可重入来绕过这个限制，代价很大：它破坏了 asyncio 的核心执行模型——重入的循环会打乱任务调度顺序，导致超时和取消的语义在某些情况下失效、TaskGroup/shield 之类的结构化并发行为异常、与 uvloop 和部分基于 C 的异步库不兼容，而且由此产生的 bug 极难排查（现象往往是「偶尔卡住」或「取消不生效」）。Jupyter 报这个错的根本原因是它本身就运行在事件循环里（为了支持 %autoawait），所以正确做法是直接在 cell 里写顶层 await（IPython 原生支持），或者用 asyncio.get_running_loop().create_task(...)。nest_asyncio 只适合临时探索，绝不能进生产代码。
• 误区：用 asgiref 的 sync_to_async 时，把 thread_sensitive 设成 False 能提高并发，应该总是这样设。 thread_sensitive=True（默认值）存在是有原因的：它保证所有被包装的同步调用在同一个线程里执行，因为 Django 的数据库连接、事务状态、以及很多框架的请求上下文都是 thread-local 的。设成 False 后调用会被分散到线程池的不同线程，后果是：每个线程各自创建数据库连接（连接数暴涨甚至打满数据库）、同一个事务的多次操作跑在不同线程上导致事务错乱、以及 thread-local 的上下文丢失。只有当被包装的函数确实与线程无关（纯计算、只读文件、不碰 ORM 和请求上下文）时，才适合用 thread_sensitive=False 来获得真正的并行。另外要避免嵌套转换（sync_to_async(async_to_sync(...))），那会层层创建线程，是性能灾难。
• 追问：为什么说「函数着色」是异步编程的固有成本？有办法绕开吗？ 「着色」指的是函数被分成了「同步」和「异步」两种颜色，且异步会传染：await 只能出现在 async def 里，所以一旦最底层的 IO 函数变成异步，它的所有调用者都必须变成异步，一路蔓延到 main()。这不是 Python 的设计缺陷，而是**「协作式并发」的必然结果**——协程必须显式标记出「我可能在这里让出控制权」，事件循环才知道在哪里切换；如果不需要标记（像 Go 的 goroutine 那样），就需要运行时提供可抢占的轻量级线程，那是另一种（成本更高的）实现路径。绕开的办法本质上都是「在边界上开销换便利」：asyncio.run()（每次建销 loop）、后台 loop + run_coroutine_threadsafe（跨线程通信开销）、asgiref（起新线程 + 上下文复制）、to_thread（线程池开销）——没有零成本的方案。所以最实际的建议是：要么整条链路统一颜色，要么把转换收敛到一两个明确的边界上，最糟的是让两种颜色在业务代码里来回横跳。
• 追问：asyncio.run()、asgiref.async_to_sync() 和 run_coroutine_threadsafe() 该怎么选？ 按「是否已有 loop」和「是否需要复用资源」两个维度选。asyncio.run()：适合程序入口（if __name__ == "__main__": asyncio.run(main())）和一次性脚本——简单，但不能嵌套、每次新建 loop、不传 contextvars。run_coroutine_threadsafe()：适合同步主程序 + 长期存活的后台 loop 的架构（GUI、同步 Web 框架、需要频繁调用异步 SDK 的场景）——loop 长期存活所以连接池等资源能复用，代价是要自己管理那个线程和 loop 的生命周期。asgiref.async_to_sync()：适合 Django/ASGI 生态——它比裸 asyncio.run 聪明的地方在于「已经在 loop 里时会另起线程跑新 loop」（从而不会 RuntimeError）并且会传递 contextvars，代价是新 loop 无法共享原 loop 的连接池、以及起线程的开销。一个实用的补充：3.11+ 的 asyncio.Runner 可以在多次调用之间复用同一个 loop，是 asyncio.run 的更好替代（with asyncio.Runner() as r: r.run(coro1()); r.run(coro2())）。
• 追问：把同步业务逻辑用 to_thread 包起来放进 FastAPI，能获得异步的性能优势吗？ 能获得一部分，但有明确的天花板。好处是：框架层（接收连接、解析请求、返回响应）是真异步的，所以能承载大量并发连接、不会因为连接多而耗尽线程；慢客户端、长连接、WebSocket 这些场景收益明显。天花板在于：业务逻辑仍然跑在线程池里，实际并发受限于线程池大小（asyncio.to_thread 用的是默认 executor，max_workers 默认只有 min(32, cpu_count + 4)，8 核机器上就是 12），超出的调用会排队。所以要做三件事：① 调大线程池（loop.set_default_executor(ThreadPoolExecutor(max_workers=N))，N 按「目标并发 × 平均耗时」估算）；② 区分任务类型——CPU 密集的必须用 ProcessPoolExecutor（线程池挡不住 GIL）；③ 注意取消语义——丢进线程的任务是无法取消的，客户端断开或超时后那个线程仍会跑到底，占着槽位。如果业务逻辑本身的瓶颈是数据库或下游 API，更彻底的收益还是要把那部分改成真正的异步驱动。

八、加强记忆

同步和异步互调是不对称的：异步调同步容易（await asyncio.to_thread(fn)），同步调异步麻烦——因为协程对象本身什么都不做，需要事件循环反复 send() 来驱动，而同步函数里没有 loop。这就是函数着色问题：一个函数变成 async，调用者被迫全变 async，传染到调用链顶端（这是协作式并发的固有成本，没有零成本的绕开方案）。同步调异步的决策树只有一条：先问「当前线程有没有正在运行的 loop」（用 asyncio.get_running_loop() 判断，别用语义混乱的 get_event_loop()）——① 没有 loop → asyncio.run(coro())，但它每次新建销毁 loop，连接池、session、DNS 缓存全丢，所以「for 循环里反复 asyncio.run」是反模式（比同步还慢且毫无并发），要把异步逻辑收进一次 run（3.11+ 可用 asyncio.Runner 复用 loop）；② loop 在别的线程 → asyncio.run_coroutine_threadsafe(coro, loop).result(timeout)，这是唯一线程安全地向 loop 提交协程的方式（asyncio 其他 API 都不线程安全，另一个例外是 call_soon_threadsafe）；③ 就在 loop 里 → 直接 await（调 asyncio.run 会 RuntimeError；若要调的是「内部封装了 asyncio.run 的同步库」，就 await asyncio.to_thread(它)）。Django 生态用 asgiref：async_to_sync（已有 loop 时会新起线程跑新 loop，还会传 contextvars）和 sync_to_async（thread_sensitive=True 让所有调用在同一线程，因为 Django ORM 连接是 thread-local，设成 False 会导致连接暴涨和事务错乱）。nest_asyncio 绝不能进生产——它打补丁让循环可重入，会破坏调度、超时和取消语义；Jupyter 里直接用顶层 await 即可。迁移策略：边界隔离（外层异步 + 内层 to_thread，注意默认线程池只有 min(32, cpu+4)、CPU 密集要用进程池、丢进线程的任务无法取消）、自底向上改、或者先问「瓶颈真的是并发连接数吗」——QPS 几百的服务用线程池完全够用。最糟的做法是让 sync/async 在业务代码里来回横跳。