asyncio与非asyncio上下文通信架构的合理性及优化方案咨询

我正在开发一个同时支持req/rep（请求/响应）和push/pull（推送/拉取）模式的TCP客户端，尝试采用asyncio及其transport/protocol结构处理底层传输操作，但供外部调用的API运行在非asyncio上下文环境中。为此我设计了如下架构：在独立线程中运行asyncio，通过asyncio.run_coroutine_threadsafe提交请求并使用future等待响应，代码实现如下：

import asyncio as aio
from threading import Thread

class MyProtocol(aio.Protocol):
    ...

async def aio_main(drv, ip, port):
    loop = aio.get_running_loop()
    on_connection_lost = loop.create_future()
    transport, protocol = loop.create_connection(
        lambda: MyProtocol(drv, on_connection_lost ), ip, port
    )
    drv._transport = transport
    drv._protocol = protocol
    drv._loop = loop
    try:
        await on_connect_lost
    finally:
        transport.close()

class Driver:
    def __init__(self):
        self._thread = Thread(target=lambda: aio.run(aio_main(self, '192.168.0.1', 8888))
        self._thread.start()

    async def _request(self, data):
        loop = aio.get_running_loop()
        fut = loop.create_future()
        self._protocol._rsp = fut
        self._transport.write(data)
        return await fut # which is set in MyProtocol

    def request(self, data):
        coro = self._request(b'HELLO')
        fut = aio.run_coroutine_threadsafe(coro, self._loop)
        return fut.result(1.0) # 1s timeout

请问该架构是否属于良好实践？是否有更合适的设计模式来处理此类场景？

现有架构的合理性与问题 #

• 合理性：这种「异步线程+线程安全提交协程」的思路是处理「非异步上下文调用异步TCP客户端」场景的常规方案之一，核心逻辑是把异步IO的执行隔离在独立线程，通过run_coroutine_threadsafe实现线程间的协程调度和结果同步，能满足基础的req/rep需求。
• 潜在问题：
  1. 请求并发冲突：MyProtocol._rsp是单例属性，若多个线程同时调用request，会出现future被覆盖的情况，导致请求与响应完全不匹配，必须为每个请求绑定唯一标识，在MyProtocol中根据标识分发响应到对应的future。
  2. 资源管理漏洞：代码中存在拼写错误（on_connect_lost应为on_connection_lost），且线程退出逻辑不清晰，若Driver被销毁时未正确停止asyncio循环和线程，会导致套接字、线程等资源泄漏。
  3. push/pull模式适配缺失：当前架构仅考虑了主动发起的req/rep请求，对于服务端主动推送的消息，需要额外的线程安全回调机制（比如用线程安全队列传递推送消息），否则非异步上下文无法安全接收推送数据。

更合适的设计模式 #

1. 线程隔离异步循环+请求ID映射（成熟优化方案） #

在现有架构基础上补全并发处理和推送支持，是这类场景的标准实践：

• 给每个请求分配唯一ID，_request协程创建future后存入线程安全的字典（key为请求ID），发送数据时带上ID；
• MyProtocol收到响应后根据ID找到对应的future并设置结果；
• 对于推送消息，在MyProtocol中把消息放入线程安全队列，Driver提供同步方法供外部获取推送内容。

核心优化代码片段：

import queue
import uuid

class Driver:
    def __init__(self):
        self._pending_requests = {}  # 需用锁保护，或用`collections.defaultdict`+线程锁
        self._push_queue = queue.Queue()
        self._lock = Thread.Lock()
        # ... 原有初始化代码

    async def _request(self, data):
        req_id = uuid.uuid4().hex
        fut = self._loop.create_future()
        with self._lock:
            self._pending_requests[req_id] = fut
        # 发送带请求ID的数据（需和服务端约定格式）
        self._transport.write(f"{req_id}:{data}".encode())
        try:
            return await fut
        finally:
            with self._lock:
                del self._pending_requests[req_id]

    def get_push_message(self, timeout=None):
        """同步获取服务端推送的消息"""
        return self._push_queue.get(timeout=timeout)

class MyProtocol(aio.Protocol):
    def __init__(self, drv, on_connection_lost):
        self.drv = drv
        self.on_connection_lost = on_connection_lost

    def data_received(self, data):
        # 解析数据：区分响应和推送（需和服务端约定格式）
        if data.startswith(b"RESP:"):
            _, req_id, rsp_data = data.split(b":", 2)
            req_id = req_id.decode()
            with self.drv._lock:
                fut = self.drv._pending_requests.pop(req_id, None)
            if fut:
                fut.set_result(rsp_data)
        elif data.startswith(b"PUSH:"):
            _, push_data = data.split(b":", 1)
            self.drv._push_queue.put(push_data)

2. 简化场景：用asyncio.to_thread替代独立线程（Python 3.9+） #

如果你的TCP客户端是短连接场景，不需要长期保持连接，可直接用asyncio.to_thread把同步API的调用转成异步执行，无需手动管理独立线程和循环。但对于长连接的TCP客户端，独立线程运行异步循环的效率更高。

3. 标准范式：异步核心+同步门面 #

采用分层设计：

• 异步核心：完全用asyncio实现TCP客户端的req/rep和push/pull逻辑，处理连接、消息收发、超时等细节；
• 同步门面（即Driver类）：负责在独立线程启动异步循环，通过run_coroutine_threadsafe或loop.call_soon_threadsafe实现线程间通信，对外暴露同步API。这种分层方式让代码职责更清晰，便于维护和扩展。

总结 #

你的初始架构方向是正确的，但需要解决并发请求冲突、资源管理和推送消息适配的问题。优化后的「线程隔离异步循环+请求ID映射+线程安全队列」方案是这类场景的成熟实践，能稳定支持req/rep和push/pull模式。

内容的提问来源于stack exchange，提问作者Holmes Conan