大家好,我是木木。今天给大家分享一个神奇的 Python 库,Trio。
Trio 是简化异步编程的 Python 库,核心在于提供了一种更清晰、更可靠的方式来编写异步代码。Trio 的接口设计更为直观,侧重于可维护性和性能。

Trio的核心特点
-
简化异步 -
Trio 使用“nursery”模式来简化异步任务的管理,使得启动和监控子任务更为直接和安全。 -
同步风格 -
它提供了一种类似于同步代码的方式来处理并发,这减少了编程的复杂性并提高了代码的可读性。 -
强大的取消机制 -
Trio 内建了强大的任务取消机制,可以非常方便地停止正在运行的异步操作,这在很多异步框架中是一个棘手的问题。
最佳实践
项目地址:https://github.com/python-trio/trio
安装 Trio 非常简单,只需要运行以下命令:
pip install trio
功能一:异步文件操作
Trio 提供了一系列用于异步文件操作的 API。例如,你可以使用 trio.open_file
高效地进行文件读写操作。
以下是一个使用 Trio 读取文件的示例代码:
import trio
# 异步读取文件的任务
async def async_read_file(nursery, file_path):
async with trio.open_file(file_path, 'r') as file:
print(f"Reading {file_path} started!")
content = await file.read()
print(f"Reading {file_path} completed!")
nursery.cancel_scope.cancel() # 读完文件后取消其他任务
async def parent():
print("parent: started!")
async with trio.open_nursery() as nursery:
print("parent: spawning file reader for file1.txt...")
nursery.start_soon(async_read_file, nursery, 'file1.txt')
print("parent: spawning file reader for file2.txt...")
nursery.start_soon(async_read_file, nursery, 'file2.txt')
print("parent: waiting for file reading to finish...")
# -- we exit the nursery block here --
print("parent: all done!")
trio.run(parent)

在 parent
函数中,我们设立了一个 nursery(托儿所),这是 Trio 管理子任务的方式。我们启动了两个子任务,每个都是异步读取一个文件。父任务在 nursery
块中等待所有子任务完成,实现了优雅的并发处理。
功能二:异步回声服务器
这个示例展示了如何用 Trio
实现一个异步回声服务器。服务器的作用是接收来自客户端的数据,然后原样发送回去,这个过程称为“回声”。这对于测试网络连接和学习异步网络编程非常有用。
import trio
from itertools import count
# 端口是任意选择的,但需要注意:
# - 必须在1024到65535之间
# - 不能被电脑上的其他程序使用
# - 必须与客户端设置的端口相匹配
PORT = 12345
CONNECTION_COUNTER = count()
async def echo_server(server_stream):
# 为每个连接分配一个唯一编号,便于在有多个同时连接时理解调试信息。
ident = next(CONNECTION_COUNTER)
print(f"echo_server {ident}: started")
try:
async for data in server_stream:
print(f"echo_server {ident}: received data {data!r}")
await server_stream.send_all(data)
print(f"echo_server {ident}: connection closed")
except Exception as exc:
# 未处理的异常会传播到父级并关闭整个程序。如果异常是 KeyboardInterrupt,
# 这是我们想要的,但其他情况可能不是...
print(f"echo_server {ident}: crashed: {exc!r}")
async def main():
await trio.serve_tcp(echo_server, PORT)
trio.run(main)
高级功能:在 asyncio 中嵌套运行 Trio
在 asyncio
主循环中集成和运行一个 Trio 程序,这种技术可以在混合异步环境中非常有用,特别是在需要同时使用多个异步库的场合。
import asyncio, trio
# trio程序
async def trio_main():
for _ in range(5):
print("Hello from Trio!")
# 在trio里面,所以要用trio的api
await trio.sleep(1)
return "trio done!"
# 作为client运行在asyncio里面
async def asyncio_main():
asyncio_loop = asyncio.get_running_loop()
def run_sync_soon_threadsafe(fn):
asyncio_loop.call_soon_threadsafe(fn)
done_fut = asyncio_loop.create_future()
def done_callback(trio_main_outcome):
print("all done from %s" % trio_main_outcome)
done_fut.set_result(trio_main_outcome)
trio.lowlevel.start_guest_run(
trio_main,
run_sync_soon_threadsafe=run_sync_soon_threadsafe,
done_callback=done_callback,
host_uses_signal_set_wakeup_fd=True
)
# 等待guest完成
trio_main_outcome = await done_fut
# 从guest返回里面获取exception或者通过
return trio_main_outcome.unwrap()
asyncio.run(asyncio_main())

小总结
通过上述示例,我们可以看到 Trio 如何通过简化的异步模式和强大的取消机制,提高 Python 异步编程的效率和可靠性。无论是文件操作、网络请求还是信号处理,Trio 都表现出了其优越的设计理念。
—— End ——
【限时福利】专为粉丝打造的《Python入门到入魔保姆级教程》来啦!
现在扫码或者添加 vx:257735 即可领取
原文始发于微信公众号(木木夕咦):Trio,一个有趣的python库
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
文章由极客之家整理,本文链接:https://www.bmabk.com/index.php/post/287151.html