如何在 Python 中让两个异步函数无限期运行

在 Python 的世界里,异步编程 早已不是什么新鲜的“黑魔法”,但在 2026 年,随着 AI 原生应用高并发 IO 密集型任务 的普及,掌握如何高效地“同时做多件事”变得比以往任何时候都重要。我们经常需要编写一些能够永久运行的后台任务——比如监听消息队列、处理 WebSocket 数据流,或者是维持一个与 LLM(大型语言模型)的长连接会话。

在本文中,我们将不仅回顾 Python asyncio 库的基础用法,还会结合我们在现代开发(特别是 AI 辅助编程)中的实战经验,深入探讨如何构建健壮、可维护的“永久运行”异步系统。

核心概念回顾

在开始之前,让我们快速梳理一下构建异步系统的基石。理解这些对于后续编写高性能代码至关重要。

  • 协程: 我们使用 async 关键字定义的函数。这不仅仅是语法糖,它告诉 Python 解释器:“这个函数会在执行过程中暂停。”
  • 事件循环: 它是异步心脏的起搏器。它负责调度所有的协程,当一个任务因等待网络 I/O(如请求 OpenAI API)而挂起时,事件循环会利用这段时间去执行其他任务。
  • asyncio.sleep(): 这是模拟异步操作的黄金标准。与 time.sleep() 不同,它不会冻结整个程序,而是把控制权交还给事件循环。

示例:基础异步运行

Python


CODEBLOCK_281ac390

解释: 当你运行这段代码时,INLINECODEfa87974c 会创建一个新的事件循环,运行 INLINECODEd61bb662,并在结束后关闭循环。await asyncio.sleep(1) 是关键,它告诉循环:“这一秒钟我没空,去干点别的吧。”

同时运行两个异步函数

在实际开发中,我们很少只运行一个任务。比如,你可能需要一边从传感器收集数据,一边在本地保存日志。让我们来看看如何并发运行两个函数。

Python


CODEBLOCK_27e35513

输出

!OutputUsing asyncio.gather()

解释: 在这个例子中,INLINECODE22b515b3 发挥了调度器的作用。它同时启动 INLINECODEd7604334 和 fun2。注意看输出的顺序,它们是交替进行的,而不是一个做完再做另一个。这就是并发的威力:总耗时约为 2.5 秒(5次 * 0.5秒),而不是串行运行时的 5 秒。

深入探讨:如何让两个异步函数永远运行

现在,我们进入今天的核心话题:无限期运行。在构建像 AI Agent(AI 代理) 这样的后台服务时,我们需要确保任务一旦启动,除非遇到致命错误,否则绝不停止。

为了实现这一点,我们可以结合 INLINECODE6fcaa390 循环和 INLINECODEee5bc928。但在 2026 年的工程实践中,我们不仅要让它“跑起来”,还要考虑它如何“优雅地停下来”。

生产级代码示例:

Python


CODEBLOCK_3142037d

输出

!OutputUsing while

解释: 这段代码展示了我们在生产环境中的思考方式。我们不仅使用了 INLINECODE2bfcc871,还引入了一个全局标志 INLINECODEc0fd9cbb 来控制退出。更重要的是,我们在 INLINECODEbf152e61 函数中处理了 INLINECODEadd278ac(即 Ctrl+C)。在 2026 年的云原生环境下,容器编排系统(如 Kubernetes)在停止 Pod 前会发送信号,如果你的代码不处理这种“优雅退出”,可能会导致数据丢失或 AI 上下文损坏。

2026 年视角:现代开发范式与技术趋势

仅仅让代码跑起来是不够的。作为现代开发者,我们需要结合最新的工具和理念来提升效率。以下是我们团队在使用 CursorGitHub Copilot 等 AI IDE 开发异步系统时的一些最佳实践。

#### 1. 智能化容错与重试机制

在 AI 时代,我们的代码经常依赖不稳定的网络(调用 LLM API)。简单的无限循环可能会因为一次网络超时而崩溃。我们可以结合 Tenacity 库或自定义装饰器来增强 INLINECODE371622cf 和 INLINECODE94485f58。

让我们思考一下这个场景:如果你的 fun2 是一个需要连接外部服务的 Agent,网络抖动是常态。我们建议这样改进:

Python


CODEBLOCK_9778dc30

这种模式在我们的 Agentic AI 工作流中非常关键,确保 AI 代理在面对外部服务波动时具有韧性。

#### 2. 结构化并发与任务组

从 Python 3.11 开始,我们引入了 INLINECODE97ec9391。这是对传统 INLINECODE03718dc6 和 create_task 的重大升级,它提供了更好的异常处理机制——如果组内有一个任务失败,TaskGroup 会自动取消其他任务,防止“僵尸进程”的出现。这符合 2026 年对代码安全性的要求。

Python


CODEBLOCK_06e19998

#### 3. 可观测性:看见不可见的

在运行永久后台任务时,最大的痛点往往是“它死了吗?”。在 2026 年,我们不仅要打印日志,还要集成 OpenTelemetry。虽然这超出了基础教程的范畴,但我们在编写代码时应预留接口。

例如,在 INLINECODE88e4e700 循环中,不要只打印 INLINECODE8e6c1824,而是使用结构化日志:

Python


CODEBLOCK_d44ee7ef

总结与最佳实践

在 2026 年,编写“两个永远运行的异步函数”不再仅仅是关于 while True 的技巧。它是关于构建一个 具有韧性、可观测且易于维护的微服务单元

让我们回顾一下关键点:

  • 使用 asyncio.run() 作为入口,管理事件循环生命周期。
  • 使用 INLINECODE8331a673 或 INLINECODE0b497a9c 来并发运行任务。在 Python 3.11+ 中,优先选择 TaskGroup 以获得更安全的异常处理。
  • 永远不要使用阻塞代码:在 INLINECODE7e5a0e6a 函数中,严禁使用 INLINECODE7d41ed1d 或同步的文件/网络 IO,否则会阻塞整个事件循环。
  • 优雅退出:总是要考虑到你的程序如何停止。处理 SIGINT 并在退出前清理资源,这是区分脚本和服务的分水岭。

在你最近的一个项目中,如果你需要同时运行一个数据监听器和一个 AI 模型推理服务,不妨尝试上述的 TaskGroup 模式。你会发现代码不仅更整洁,而且在面对异常时更加健壮。

常见问题 (FAQ)

#### 如何在 Python 中一次运行两个异步函数?

> 最现代且推荐的方法是使用 INLINECODEa481b1ce (Python 3.11+)。如果你使用的是旧版本,INLINECODEf1e459a5 依然是可靠的选择。方法如下:

>

>

> import asyncio
> 
> async def first_function():
>     print("First function running")
>     await asyncio.sleep(1)
> 
> async def second_function():
>     print("Second function running")
>     await asyncio.sleep(1)
> 
> async def main():
>     async with asyncio.TaskGroup() as tg:
>         tg.create_task(first_function())
>         tg.create_task(second_function())
> 
> asyncio.run(main())
> 

#### 如何在 Python 中多次运行一个异步函数?

> 要并发运行一个异步函数多次,通常是为了进行压力测试或并行处理。我们可以在列表推导式中生成任务列表:

>

>

> import asyncio
> 
> async def async_task(n):
>     print(f"Task {n} started")
>     await asyncio.sleep(1)
>     print(f"Task {n} finished")
> 
> async def main():
>     # 使用 TaskGroup 批量创建 5 个任务
>     async with asyncio.TaskGroup() as tg:
>         for i in range(5):
>             tg.create_task(async_task(i))
> 
> asyncio.run(main())
> 

#### 如何在 Python 中使函数异步化?

> 将同步函数改造为异步函数不仅仅是加上 INLINECODE6304d944 关键字。你需要找到函数中的 I/O 瓶颈(如数据库查询、HTTP 请求、文件读写),并将它们替换为对应的异步库(如 INLINECODE0bb1e998, INLINECODEb201a6f0, INLINECODEbad7aafc)。

>

> 错误示范:

>

> async def bad_async():
>     time.sleep(1) # 这会阻塞整个循环!
> 

>

> 正确示范:

>

> async def good_async():
>     await asyncio.sleep(1) # 交出控制权
> 

通过掌握这些技术,你已经准备好在 2026 年构建下一代 Python 后端服务了。快乐编码!

声明:本站所有文章,如无特殊说明或标注,均为本站原创发布。任何个人或组织,在未征得本站同意时,禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益,可联系我们进行处理。如需转载,请注明文章出处豆丁博客和来源网址。https://shluqu.cn/21878.html
点赞
0.00 平均评分 (0% 分数) - 0