如何在Python同步代码中运行无阻塞异步解析代码（无需事件循环）

首先明确：只要代码里有await，就必须在异步上下文（事件循环）里执行——哪怕所有被await的函数都不阻塞。不过Trio提供了极轻量的方式运行异步代码，完全不需要复杂的启动流程，而且对无阻塞的场景来说，开销几乎可以忽略。

1. 实现无阻塞的异步Reader #

自己写个简单的异步版BytesIO就行，read()方法直接返回预存的数据，因为数据在内存里，内部完全是同步操作，不会挂起：

class AsyncBytesIO:
    def __init__(self, data: bytes):
        self.buffer = data
        self.pos = 0

    async def read(self, n: int) -> bytes:
        result = self.buffer[self.pos:self.pos+n]
        self.pos += len(result)
        return result

2. 在同步代码里运行异步parse函数 #

用Trio的trio.run()就行，它会启动一个轻量事件循环，运行你的异步函数直到完成，开销极小：

import trio

async def parse(reader):
    d = await reader.read(2)
    length = int.from_bytes(d, byteorder='big')
    d2 = await reader.read(length)
    return {'length': length, 'data': d2}

# 同步调用的函数
def sync_parse(data: bytes):
    reader = AsyncBytesIO(data)
    return trio.run(parse, reader)

# 测试
test_data = b'\x00\x04hell'
print(sync_parse(test_data))  # 输出: {'length': 4, 'data': b'hell'}

能不能完全不用事件循环？ #

严格来说不行。Python的协程依赖事件循环调度，哪怕所有协程都不阻塞，也需要事件循环驱动执行。不过上面的trio.run()已经足够轻量，完全没必要纠结。

如果非要搞个hack，也可以手动驱动协程，但这种方法不推荐——绕过了异步框架的安全机制，以后如果parse函数加了真正的异步操作，直接就报错：

def sync_parse_hack(data: bytes):
    reader = AsyncBytesIO(data)
    coro = parse(reader)
    try:
        while True:
            coro.send(None)
    except StopIteration as e:
        return e.value

总之，用trio.run()是最靠谱、最符合规范的方式。

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