如何同步运行异步NATS函数？Python模块RPC适配问题

你的核心需求是封装Python模块，让本地同步调用与NATS RPC异步调用的使用方式完全一致，但直接用多次asyncio.run()包裹异步操作会出现请求超时、事件循环已关闭的错误。

为什么多次调用asyncio.run()不可行？ #

• asyncio.run()每次调用都会创建全新的事件循环，协程执行完毕后会自动关闭该循环并清理资源。
• 你的NATS客户端nc是全局实例，第一次connect()在第一个事件循环中建立的连接，和后续asyncio.run()创建的新循环完全无关。调用request()时，NATS客户端试图在新循环中处理响应，但该循环并未绑定已建立的连接，导致响应无法被接收，最终超时。
• 最后调用close()时，第一个事件循环已经被asyncio.run()关闭，客户端实例绑定的循环不存在，因此抛出Event loop is closed错误。

解决方案 #

方案1：后台线程常驻事件循环（最适配你的场景） #

创建后台线程运行事件循环，所有NATS异步操作都提交到该循环执行，同步等待结果。这样Service类的方法可保持同步调用形式，完全屏蔽异步细节。

RPC版本的Service类示例：

from nats.aio.client import Client as NATS
import asyncio
import threading

class Service:
    def __init__(self):
        self.nc = NATS()
        self.loop = asyncio.new_event_loop()
        # 启动后台线程运行事件循环
        self.thread = threading.Thread(target=self._run_loop, daemon=True)
        self.thread.start()
        # 同步等待连接完成
        self._run_coroutine_sync(self._connect())

    def _run_loop(self):
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    def _run_coroutine_sync(self, coro):
        # 提交协程到后台循环并同步等待结果
        future = asyncio.run_coroutine_threadsafe(coro, self.loop)
        return future.result()

    async def _connect(self):
        await self.nc.connect("nats://localhost:4222")

    def add(self, a, b):
        # 封装异步request为同步调用
        result = self._run_coroutine_sync(self._nc_request("service.add", f"{a},{b}".encode()))
        return int(result.data.decode())

    def sub(self, a, b):
        result = self._run_coroutine_sync(self._nc_request("service.sub", f"{a},{b}".encode()))
        return int(result.data.decode())

    async def _nc_request(self, subject, payload):
        msg = await self.nc.request(subject, payload, timeout=5)
        return msg

    def __del__(self):
        # 关闭连接和事件循环
        self._run_coroutine_sync(self.nc.close())
        self.loop.call_soon_threadsafe(self.loop.stop)
        self.thread.join()

本地版本的Service类可直接编写同步方法，用户调用代码无需修改：

# mylib.py
class Service:
    def add(self, a, b):
        return a + b
    def sub(self, a, b):
        return a - b

方案2：复用单例事件循环 #

若不想使用线程，可手动管理全局事件循环，避免每次asyncio.run()创建新循环。注意该方式仅适用于单一主线程场景，不能与其他异步代码冲突：

from nats.aio.client import Client as NATS
import asyncio

nc = NATS()
_loop = None

def _get_loop():
    global _loop
    if _loop is None or _loop.is_closed():
        _loop = asyncio.new_event_loop()
        asyncio.set_event_loop(_loop)
    return _loop

class Service:
    def __init__(self):
        # 同步执行连接
        _get_loop().run_until_complete(nc.connect("nats://localhost:4222"))

    def add(self, a, b):
        result = _get_loop().run_until_complete(self._request("service.add", f"{a},{b}".encode()))
        return int(result.data.decode())

    def sub(self, a, b):
        result = _get_loop().run_until_complete(self._request("service.sub", f"{a},{b}".encode()))
        return int(result.data.decode())

    async def _request(self, subject, payload):
        msg = await nc.request(subject, payload)
        return msg

关键注意点 #

• 确保NATS客户端生命周期与事件循环绑定，避免循环被提前关闭。
• 后台线程方案需设置daemon=True，保证主程序退出时线程自动结束。
• 同步等待异步结果时，需处理超时、连接失败等异常，提升模块健壮性。

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