Python异步函数实现：同时等待Socket读取或事件触发终止

要实现同时监听流读取和trio.Event触发、优先响应先完成操作的逻辑，Trio提供了trio.wait_any API（Trio 0.22及以上版本支持），可以直接模拟Go语言中select语句的行为。

实现代码 #

import trio
# 请替换为实际的UStreams和MAX定义的导入路径
from your_module import UStreams, MAX

async def read_data(self, stream: UStreams, evt: trio.Event) -> None:
    while True:
        # 同时等待流读取和事件触发
        completed_result = await trio.wait_any(
            stream.read(MAX),
            evt.wait()
        )
        
        # 判断哪个操作先完成
        if completed_result is evt.wait():
            # 事件触发，终止循环退出函数
            break
        
        # 流读取完成，处理字节数据
        read_bytes = completed_result
        # 这里添加你的字节处理逻辑
        # 示例：print(f"读取到{len(read_bytes)}字节数据")

逻辑说明 #

• trio.wait_any会同时挂起传入的多个异步可等待对象，仅等待第一个完成的操作，并返回该操作的结果
• 当evt.wait()先完成时，直接退出循环终止函数
• 当stream.read(MAX)先完成时，立即获取读取到的字节进行处理，不会阻塞等待事件触发

兼容老版本Trio（0.22以下） #

如果使用的是较早版本的Trio，没有trio.wait_any，可以通过Nursery和MultiError实现相同逻辑：

import trio
from your_module import UStreams, MAX

async def read_data(self, stream: UStreams, evt: trio.Event) -> None:
    while True:
        async with trio.Nursery() as nursery:
            # 启动两个并行任务
            read_task = nursery.start_soon(stream.read, MAX)
            evt_task = nursery.start_soon(evt.wait)

            # 等待任意一个任务完成后取消另一个
            try:
                await trio.wait([read_task, evt_task])
            except trio.MultiError as e:
                # 过滤掉取消异常，只保留完成的任务结果
                e.filter(lambda exc: isinstance(exc, trio.Cancelled))
            
            # 判断哪个任务完成
            if evt_task.done():
                break
            read_bytes = read_task.result()
            # 处理字节数据...

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