asyncio 本身不支持跨进程通信，需结合 multiprocessing 或 IPC 机制；推荐 Unix domain socket（低开销、原生异步支持）或 multiprocessing.Pipe 配合 asyncio.to_thread；禁用 multiprocessing.Queue 直接 await 及 asyncio.Queue 跨进程误用。
asyncio 本身不支持跨进程通信，必须搭配 multiprocessing 或其他 IPC 机制
asyncio 是单进程、单线程的异步运行时，async/await 无法直接跨越进程边界。想让两个 Python 进程（比如主协程进程 + 工作进程）异步通信，得靠外部 IPC 载体——asyncio 只负责在本进程中「非阻塞地读写」这个载体。
常见可行组合是：asyncio + multiprocessing.Pipe（需配合 asyncio.StreamReader/StreamWriter 封装）、asyncio + socket（本地 Unix domain socket 更轻量）、或 asyncio + asyncio.Queue（仅限同进程，跨进程无效，常被误用）。
最容易踩的坑是试图把 multiprocessing.Queue 直接 await：它不是可等待对象，也不提供 async 接口，强行调用会报 TypeError: object Queue is not awaitable。
用 asyncio.open_unix_connection 配合 Unix socket 实现低开销跨进程 IPC
Unix domain socket 是本地进程间最常用、零网络栈开销的 IPC 方式，且原生支持 asyncio 的流式 API，无需额外线程桥接。
立即学习“Python免费学习笔记（深入）”；
- 服务端进程用
await asyncio.start_unix_server()启动监听，每个连接由协程处理 - 客户端进程用
await asyncio.open_unix_connection(path)建立连接，之后用reader.read()/writer.write()异步收发 - 务必设置
path为文件系统中可写路径（如/tmp/myapp_ipc.sock），并注意权限和残留 socket 文件清理 - 消息需自行定义协议（如 JSON 行分隔、长度前缀），避免粘包；不要依赖
readline()除非确保发送端每条消息都带\n
示例片段（服务端）：
安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。
import asyncio
import json
<p>async def handle_client(reader, writer):
data = await reader.read(1024)
try:
msg = json.loads(data.decode())
result = {"status": "ok", "echo": msg.get("data")}
writer.write(json.dumps(result).encode() + b"\n")
await writer.drain()
except Exception as e:
writer.write(b'{"error": "invalid json"}\n')
await writer.drain()
finally:
writer.close()</p><p>async def main():
server = await asyncio.start_unix_server(handle_client, path="/tmp/pipe.sock")
async with server:
await server.serve_forever()</p><p>asyncio.run(main())
用 multiprocessing.Pipe + asyncio.to_thread 绕过阻塞读写（适合简单二进制通信）
如果已有基于 multiprocessing.Pipe 的代码，又不想改通信层，可用 asyncio.to_thread 把 conn.recv() / conn.send() 包装成可等待操作，避免阻塞事件循环。
-
to_thread在后台线程执行阻塞调用，适用于低频、小数据量场景（如控制信号、状态心跳） - 不能用于高吞吐场景——线程池有上限，且频繁创建线程开销大
-
Pipe的两端必须在 fork 后由父子进程各自持有，不能在 asyncio 任务中跨进程传递 conn 对象 - 接收端需加超时保护，否则
await asyncio.to_thread(conn.recv)可能永久挂起
关键调用模式：
try:
data = await asyncio.wait_for(
asyncio.to_thread(conn.recv),
timeout=5.0
)
except (EOFError, TimeoutError):
# 处理断连或超时
别碰 asyncio.Queue 和 multiprocessing.Manager —— 它们都不解决跨进程 async 问题
asyncio.Queue 是协程安全的内存队列，但只在单个事件循环内有效；multiprocessing.Manager().Queue() 虽能跨进程，但所有操作（get/put）都是同步阻塞的，没有 async 方法，也无法直接 await。
有人尝试用 loop.run_in_executor 包装 Manager.Queue 的方法，这理论上可行，但引入了不必要的线程调度和序列化开销，且容易因 manager 进程崩溃导致死锁。真实项目中，应优先选 socket 或 Pipe + to_thread，而非硬套 Queue。
真正麻烦的地方不在“怎么连”，而在“怎么断”：IPC 连接意外中断时，reader 可能静默卡在 read()，writer 可能积压未 flush 的数据。务必对每个 IO 操作设 timeout，并监听 ConnectionResetError 和 BrokenPipeError。
