终结异步生态之争:用 AnyIO 统一 Python 的并发江湖

import anyioasync def main():    print("异步引擎启动，开始执行业务逻辑...")    # 模拟一个耗时的网络 I/O    await anyio.sleep(1)    print("业务逻辑执行完毕。")if __name__ == "__main__":    # 默认使用 asyncio 运行    anyio.run(main)    # 如果你想测试 trio 引擎的性能，只需一行代码切换    # 甚至不需要修改 main 函数里的任何逻辑    # anyio.run(main, backend="trio")

import anyioasync def worker(task_id, delay, should_crash=False):    print(f"任务 {task_id} 开始工作...")    await anyio.sleep(delay)    if should_crash:        raise RuntimeError(f"任务 {task_id} 发生致命错误！")    print(f"任务 {task_id} 顺利完成。")async def main():    try:        # 创建一个任务组作用域        async with anyio.create_task_group() as tg:            # 将任务丢给任务组管理            tg.start_soon(worker, "A", 2)            tg.start_soon(worker, "B", 4)            # 模拟任务 C 在 1 秒后崩溃            tg.start_soon(worker, "C", 1, True)    except RuntimeError as e:        print(f"捕获到任务组抛出的异常: {e}")        print("因为任务 C 崩溃，任务 A 和 B 已经被安全取消。")if __name__ == "__main__":    anyio.run(main)

import anyioasync def fragile_network_request():    print("发起远程请求...")    # 模拟一个卡住的请求，需要 10 秒才能返回    await anyio.sleep(10)    return "请求成功"async def main():    print("准备调用不稳定的接口...")    try:        # 限制这段代码块最多执行 3 秒        with anyio.fail_after(3):            result = await fragile_network_request()            print(result)    except TimeoutError:        print("接口响应太慢，已被强制取消并记录日志。")    # AnyIO 还提供 move_on_after，超时后不抛出异常，而是静默跳过    with anyio.move_on_after(2) as cancel_scope:        await anyio.sleep(5)    if cancel_scope.cancel_called:        print("静默超时触发，继续执行后续的主线逻辑。")if __name__ == "__main__":    anyio.run(main)

import anyioasync def write_and_read_file():    file_path = anyio.Path("test_data.txt")    # 异步写入文件，不阻塞事件循环    await file_path.write_text("Hello, 异步文件系统！\n这是 AnyIO 的实战演示。")    print("文件异步写入成功。")    # 异步读取文件    if await file_path.exists():        content = await file_path.read_text()        print("读取到的内容：")        print(content)    # 清理现场    await file_path.unlink()if __name__ == "__main__":    anyio.run(write_and_read_file)