子进程中不能直接调用 asyncio.run()，因 spawn/fork 启动的新进程无默认事件循环且策略未初始化；应手动创建、设置并关闭事件循环，或改用 ProcessPoolExecutor 配合同步入口函数。
不能直接在子进程中运行 asyncio.run() 或启动事件循环，除非显式配置——因为 multiprocessing 默认 spawn/fork 启动的新进程不带主线程的事件循环，且 asyncio 的默认策略（DefaultEventLoopPolicy）在子进程中未初始化。
为什么 asyncio.run() 在子进程里会报 RuntimeError: asyncio.run() cannot be called from a running event loop
常见错误现象是：你在 worker 函数里直接调用 asyncio.run(some_coro())，结果抛出 RuntimeError: asyncio.run() cannot be called from a running event loop ——这通常不是因为你“嵌套调用”，而是因为在某些平台（尤其是 Windows/macOS 使用 spawn 启动方式时），子进程导入模块阶段可能意外触发了某个库的异步初始化（比如某些 HTTP 客户端、日志 handler 或第三方包的惰性 import），悄悄创建了一个已关闭/半初始化的循环。
-
spawn方式下，子进程从头执行if __name__ == '__main__'之外的所有顶层代码，任何提前 import 的异步相关模块都可能埋雷 - 即使你没显式 import
asyncio，aiohttp、httpx、aiomysql等库内部也可能触发循环创建逻辑 - 解决核心：确保子进程的
worker函数中，所有asyncio调用都发生在干净、未污染的上下文中
正确做法：在 worker 函数内部手动新建并关闭事件循环
不要依赖 asyncio.run()，改用底层 API 显式管理循环生命周期，避免策略冲突：
import asyncio
import multiprocessing
<p>def worker(task_id):</p><h1>每个子进程必须自己新建事件循环</h1><pre class="brush:php;toolbar:false;">loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(fetch_data(task_id))
return result
finally:
loop.close() # 必须关闭，否则资源泄漏async def fetch_data(task_id): await asyncio.sleep(1) # 模拟 I/O return f"done-{task_id}"
- 必须调用
asyncio.set_event_loop(loop)，否则run_until_complete会找不到当前循环 -
loop.close()不可省略，尤其在频繁启停子进程的场景下，否则 fd 泄漏会导致 OSError: Too many open files - 该写法在 Linux（fork/spawn）、Windows（spawn）、macOS（spawn）上均稳定
何时该用 ProcessPoolExecutor 而非 Pool + manual loop
如果你的任务本质是「CPU 密集 + 少量异步 I/O」（例如：解析大文件 → 异步上传结果 → 返回状态），用 concurrent.futures.ProcessPoolExecutor 更安全：
SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的 双功能合体技能 ，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途 ：；核心要点📝 题解模式 ：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式 ：输入 Python 代码，自动添加详细中。它将相关步骤、
from concurrent.futures import ProcessPoolExecutor import asyncio <p>def sync_worker(task_id):</p><h1>这里可以安全调用 asyncio.run()，因为它是顶层同步函数</h1><pre class="brush:php;toolbar:false;">return asyncio.run(fetch_data(task_id))
if name == 'main': with ProcessPoolExecutor(max_workers=4) as executor: results = list(executor.map(sync_worker, range(8)))
-
ProcessPoolExecutor隐式处理了子进程的模块重载和环境隔离，比裸multiprocessing.Pool更少踩坑 - 它天然兼容
asyncio.run()，前提是sync_worker是纯同步入口，且不跨进程复用 event loop - 注意：不能把协程对象（如
fetch_data(1)）直接传给executor.submit，必须包装成同步函数
真正需要警惕的陷阱：共享状态与 pickle 限制
asyncio 对象（Task、Future、EventLoop）无法被 pickle，因此绝不能作为参数传入 Pool.map 或通过 Queue 发送给子进程：
- 错误示例：
p.map(worker, [asyncio.create_task(...)])→TypeError: can't pickle _asyncio.Task objects - 所有跨进程数据必须是纯 Python 基本类型（
dict、list、int、str）或numpy.ndarray（需额外处理） - 如果需要传递配置，用
json.dumps()序列化后再传，子进程里再json.loads()
最易被忽略的一点：子进程里的 asyncio 行为完全独立于父进程，哪怕你用了 asyncio.Queue 或 asyncio.Event，它们只在单个进程内有效——跨进程通信仍得靠 multiprocessing.Queue、multiprocessing.Pipe 或文件/数据库等外部媒介。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
