异步自由线程的 Python¶
asyncio 使用事件循环作为调度器，通过在任务之间切换以允许非阻塞输入/输出操作，从而实现高效的并发。这使得它在输入/输出范围用例中具有更好的性能。它还允许将 CPU 密集型工作卸载到线程或进程池中，但这仍然受到 CPython 中 全局解释器锁 的限制。
然而，在 自由线程 Python 中，全局解释器锁被禁用，Python 可以运行真正的多线程代码。这意味着 asyncio 现在可以利用多个 CPU 核心，而不受全局解释器锁的限制。
从Python 3.14开始，asyncio对自由线程Python提供了一流的支持，并且asyncio的实现可以在多线程环境中安全使用。
单个核心上的单个事件循环可以并发处理许多连接，但是用于处理每个连接的 Python 代码仍然串行执行。一旦请求涉及大量的每个请求计算，该处理就会成为瓶颈，单个核心就无法跟上。在这里，将 asyncio 与线程结合使用最为有用：通过每个线程运行一个事件循环，不同请求的处理可以在多个 CPU 核心上并行运行。当你需要从 asyncio 应用程序运行阻塞或 CPU 密集型代码时，它也很有用。
参见
在自由线程 Python 上扩展 asyncio，Kumar Aditya 撰写的博客文章，解释了为使 asyncio 在自由线程 Python 下安全且高效所做的内部变更，并附有相应改进效果的基准测试数据。
线程安全注意事项¶
虽然asyncio被设计为在自由线程Python环境中是线程安全的，但在将asyncio与线程一起使用时，仍有一些注意事项需要记住：
事件循环：每个线程都应该有自己的事件循环，不应该在线程之间共享。这可以确保事件循环可以管理自己的任务和回调函数，而不会受到其他线程的干扰。
任务管理：在一个线程中创建的任务和未来不应等待或从另一个线程进行操作。
线程安全 API：当从多个线程与 asyncio 交互时，使用 asyncio 提供的线程安全 API 非常重要，例如
asyncio.run_coroutine_threadsafe()用于从另一个线程向事件循环提交协程。如果你需要从不同的线程调用回调函数，你可以使用loop.call_soon_threadsafe()来安全地调度它。同步：asyncio 提供的同步原语（如
asyncio.Lock和asyncio.Event）并非设计为跨线程使用。如果需要在线程之间同步，你应该改用threading模块中的同步原语。
在线程中使用asyncio¶
asyncio 支持每个线程运行一个事件循环，这允许你在自由线程 Python 环境中利用多个 CPU 核心。每个线程都可以运行自己的事件循环，任务可以在这些循环上独立调度。
下面是如何在线程中使用异步的示例:
import asyncio
import threading
async def worker(name: str) -> None:
print(f"Worker {name} starting")
await asyncio.sleep(1)
print(f"Worker {name} done")
def run_loop(name: str) -> None:
asyncio.run(worker(name))
threads = [
threading.Thread(target=run_loop, args=(f"T{i}",))
for i in range(4)
]
for t in threads:
t.start()
for t in threads:
t.join()
在这个例子中，每个线程都使用 asyncio.run() 创建自己的事件循环，并在其上运行一个协程。这些线程并发执行，并且在自由线程构建版本中，它们可以并行运行在不同的 CPU 核心上。
跨线程的生产者/使用者¶
当一个常规 (非 asyncio) 线程需要将工作交给在另一个线程中运行的 asyncio 事件循环时，请使用例如 queue.Queue 这样的线程安全原语，而不是 asyncio.Queue，后者仅在单个事件循环内是安全的。:
import asyncio
import queue
import threading
def producer(q: queue.Queue[int]) -> None:
for i in range(5):
print(f"Producing {i}")
q.put(i)
q.shutdown()
async def consumer(q: queue.Queue[int]) -> None:
while True:
try:
item = q.get_nowait()
except queue.Empty:
await asyncio.sleep(0.1)
continue
except queue.ShutDown:
break
print(f"Consumed {item}")
await asyncio.sleep(item)
q: queue.Queue[int] = queue.Queue()
consumer_thread = threading.Thread(
target=lambda: asyncio.run(consumer(q))
)
consumer_thread.start()
producer(q)
consumer_thread.join()
生产者在主线程上运行，而使用者在它自己的线程内的事件循环中运行，然而它们通过 queue.Queue 安全地进行通信。当队列为空时，使用者会短暂休眠并重试。当生产者完成时，它会调用 shutdown()，这会导致后续对 get_nowait() 的调用引发 queue.ShutDown，以便使用者可以干净地退出。