asyncio与multiprocessing不能直接混用，因前者基于单线程事件循环，后者创建隔离进程且无默认事件循环；须在子进程中独立调用asyncio.run()启动异步任务，避免共享loop或传递协程对象。
asyncio 和 multiprocessing 不能直接混用
Python 的 asyncio 基于单线程事件循环，而 multiprocessing 创建的是完全隔离的进程，每个子进程默认没有运行事件循环。直接在子进程中调用 asyncio.run() 或试图共享 loop 会失败，常见报错是 RuntimeError: There is no current event loop in thread 或 ValueError: set_event_loop() must be called from the same thread。
关键判断：不是“怎么混用”，而是“在哪一层做分工”——multiprocessing 负责跨核/跨机器的并行调度（Worker 进程），每个 Worker 内部用 asyncio 处理 I/O 密集型任务（如 HTTP 请求、数据库查询）。
- 不要在主进程启动
asyncio.run()后 fork 出子进程；子进程必须自己初始化自己的事件循环 - 避免尝试把
async函数传给multiprocessing.Process或Pool；它不支持可序列化协程对象 - 如果用
concurrent.futures.ProcessPoolExecutor，记得所有提交的函数必须是普通同步函数，内部再启动自己的asyncio
每个 Worker 进程内独立运行 asyncio.run()
每个子进程应作为独立的异步工作单元启动，典型结构是定义一个入口函数，在其中调用 asyncio.run(main())。这样能确保事件循环在该进程的主线程中创建和关闭。
示例 Worker 入口：
def worker_entrypoint(task_id: int):
import asyncio
async def main():
# 模拟异步 I/O 工作
await asyncio.sleep(1)
print(f"Worker {task_id} done")
asyncio.run(main()) # ✅ 正确：每个进程有自己的 loop
- 切勿在
worker_entrypoint中使用async def并试图被Process直接执行——Process(target=...)只接受同步 callable - 若需传参（如配置、队列句柄），用普通参数传递；不要传
asyncio.Queue或loop实例，它们无法跨进程序列化 - Windows/macOS 上注意
spawn启动方式（默认），会重新导入模块，确保入口函数在if __name__ == "__main__":之外定义
进程间通信推荐用 multiprocessing.Queue 或 pipes
asyncio.Queue 是线程/协程安全的，但不能跨进程；multiprocessing.Queue 是进程安全的，且支持在子进程中被 asyncio 协程通过 loop.run_in_executor 非阻塞读写。
快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。
例如 Worker 从队列取任务：
def worker_with_queue(task_queue: multiprocessing.Queue, result_queue: multiprocessing.Queue):
import asyncio
async def handle_task():
loop = asyncio.get_running_loop()
# 在 executor 中同步读 queue，避免阻塞 event loop
task = await loop.run_in_executor(None, task_queue.get)
# ... 处理 task
await asyncio.sleep(0.1)
await loop.run_in_executor(None, result_queue.put, f"done-{task}")
asyncio.run(handle_task())
- 不要在协程里直接调用
task_queue.get(block=True)—— 这会阻塞整个 event loop -
run_in_executor是桥接同步阻塞 I/O 和异步逻辑的关键；建议用None（默认 ThreadPoolExecutor）或自定义ProcessPoolExecutor（仅当 queue 操作本身很重时） - 如果要用更高效通信（如大量小消息），考虑
multiprocessing.Pipe+asyncio.StreamReader/StreamWriter封装，但复杂度显著上升
分布式扩展需替换本地 multiprocessing 为网络协议
真分布式（非单机多核）时，multiprocessing 失效。此时应放弃 Process/Queue，改用轻量协议对接 Worker：比如每个 Worker 是一个独立 Python 进程，暴露 HTTP 接口（用 httpx.AsyncClient 或 FastAPI），主节点用 asyncio.gather 并发请求。
或者用现成消息队列：redis 的 list/pubsub + aioredis，或 RabbitMQ + aio-pika。Worker 进程启动后，只做一件事：监听队列、拉取任务、执行异步逻辑、回传结果。
- 别为了“分布”硬套 multiprocessing 模块；它的设计目标就是本地进程管理
- 单机多 Worker 场景下，
multiprocessing+asyncio是合理组合；跨机器时，网络延迟和序列化开销远大于进程开销，通信模型必须升级 - 调试时容易忽略：不同 Worker 进程的日志可能交错输出，建议在日志中打上
os.getpid()或task_id
实际最难的部分不是写异步逻辑，而是设计好进程边界——哪些状态必须隔离（如数据库连接、全局缓存），哪些要通过序列化传递（如任务参数、超时配置）。跨进程不共享内存，这点比跨线程更彻底。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
