怎样在Python中使用asyncio实现跨进程异步IPC通信？-Python教程-PHP中文网

asyncio 本身不支持跨进程通信，需结合 multiprocessing 或 IPC 机制；推荐 Unix domain socket（低开销、原生异步支持）或 multiprocessing.Pipe 配合 asyncio.to_thread；禁用 multiprocessing.Queue 直接 await 及 asyncio.Queue 跨进程误用。

怎样在python中使用asyncio实现跨进程异步ipc通信？

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，无需额外线程桥接。

服务端进程用 await asyncio.start_unix_server() 启动监听，每个连接由协程处理
客户端进程用 await asyncio.open_unix_connection(path) 建立连接，之后用 reader.read() / writer.write() 异步收发
务必设置 path 为文件系统中可写路径（如 /tmp/myapp_ipc.sock），并注意权限和残留 socket 文件清理
消息需自行定义协议（如 JSON 行分隔、长度前缀），避免粘包；不要依赖 readline() 除非确保发送端每条消息都带 \n

示例片段（服务端）：

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())
</p>

用 multiprocessing.Pipe + asyncio.to_thread 绕过阻塞读写（适合简单二进制通信）

如果已有基于 multiprocessing.Pipe 的代码，又不想改通信层，可用 asyncio.to_thread 把 conn.recv() / conn.send() 包装成可等待操作，避免阻塞事件循环。

Secure Python Code Sandbox

安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。

下载

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):
    # 处理断连或超时