Python怎么使用async/await_Python异步编程async/await入门

Python使用

async/await

async def

await

在Python中，

async/await

async def

await

async def

await

await

一个最基本的例子可能长这样：

import asyncio

async def say_hello(delay, message):
    print(f"[{asyncio.current_task().get_name()}] 开始等待 {delay} 秒...")
    await asyncio.sleep(delay) # 这是一个异步的非阻塞等待
    print(f"[{asyncio.current_task().get_name()}] {message}!")

async def main():
    # 创建并运行两个协程任务
    task1 = asyncio.create_task(say_hello(3, "Hello from Task 1"), name="Task-1")
    task2 = asyncio.create_task(say_hello(1, "Hi from Task 2"), name="Task-2")

    # 等待这两个任务完成
    await task1
    await task2
    print("所有任务都完成了。")

if __name__ == "__main__":
    asyncio.run(main())

在这个例子里，

say_hello

main

say_hello

await

asyncio.sleep

Task-1

Task-2

立即学习“Python免费学习笔记（深入）”；

Python异步编程的核心优势是什么？它与传统同步编程模式有何不同？

在我看来，Python异步编程最核心的优势在于它能够以极低的资源开销实现高并发，尤其是在I/O密集型任务中。我们都知道，CPU的速度远超硬盘读写或网络传输。在传统的同步编程模式下，当程序发起一个网络请求或文件读取时，它会傻傻地等待，直到操作完成才能继续执行。这段等待时间，CPU其实是闲置的，什么也没做。这就像你点了一份外卖，然后就坐在门口，什么也不干，直到外卖送到。

异步编程改变了这种模式。它允许程序在等待I/O操作时“暂停”当前任务，将CPU的控制权交还给一个调度器（也就是事件循环），让调度器去运行其他已经准备好的任务。当I/O操作完成后，调度器会通知之前暂停的任务继续执行。这就像你点完外卖后，可以去洗个澡、看会儿电视，甚至再点一份水果，等外卖快到了再准备开门。这样，你的时间就得到了更有效的利用。

与传统同步编程模式的主要区别：

• 执行模型：
  • 同步： 任务按顺序执行，一个任务必须完全结束后，下一个任务才能开始。如果其中一个任务阻塞（等待I/O），整个程序都会停滞。
  • 异步： 任务可以在等待I/O时“挂起”，让出CPU，允许其他任务运行。当挂起的任务I/O就绪后，它会被事件循环重新调度执行。它实现的是“协作式多任务”，而不是操作系统层面的抢占式多任务。
• 资源利用：
  • 同步： 为了实现并发，通常需要创建多个线程或进程。每个线程/进程都有自己的内存开销和上下文切换成本。
  • 异步： 通常在一个单线程内实现高并发。通过协程的切换，避免了线程/进程的创建和销毁开销，以及昂贵的上下文切换。这使得它在处理大量并发连接时，内存占用和CPU开销都更低。
• 复杂度：
  • 同步： 逻辑相对直接，易于理解和调试，因为代码的执行顺序就是编写顺序。
  • 异步： 引入了事件循环、协程、

    async/await

    登录后复制
    等概念，初期学习曲线可能更陡峭。调试时也需要适应其非线性的执行流程。但一旦掌握，它能解决同步编程难以高效处理的并发问题。

我个人的体会是，很多人初次接触异步编程会觉得它很“魔幻”，甚至有点反直觉。但一旦你理解了它不是为了让CPU计算得更快，而是为了让CPU在等待外部资源时也能保持“忙碌”，去做一些有意义的事情，那么它的价值就显而易见了。它不是万能药，计算密集型任务依然需要多进程来利用多核CPU，但对于网络服务、爬虫、数据处理管道等I/O密集型应用，异步编程无疑是提升性能和响应速度的利器。

在Python中，如何启动和有效地管理

asyncio

登录后复制

事件循环？

启动和管理

asyncio

asyncio

1. 使用

asyncio.run()

这是目前最简洁、最推荐的启动事件循环的方式。

asyncio.run()

import asyncio

async def main_coroutine():
    print("主协程开始")
    await asyncio.sleep(1)
    print("主协程结束")

if __name__ == "__main__":
    print("启动事件循环...")
    asyncio.run(main_coroutine())
    print("事件循环已关闭。")

asyncio.run()

• 创建一个新的事件循环。
• 运行传入的协程直到完成。
• 关闭事件循环。
• 处理一些清理工作，比如取消所有未完成的任务。

对于大多数简单的脚本或应用程序，

asyncio.run()

2. 手动管理事件循环 (适用于特定场景或旧代码)

在Python 3.7之前，或者当你需要在更复杂的环境中（例如，在现有线程中集成

asyncio

import asyncio

async def my_task(name):
    print(f"任务 {name} 开始")
    await asyncio.sleep(2)
    print(f"任务 {name} 结束")

def manual_loop_management():
    loop = asyncio.get_event_loop() # 获取当前线程的事件循环，如果没有则创建
    try:
        print("手动管理：运行任务 A")
        loop.run_until_complete(my_task("A")) # 运行一个协程直到完成

        print("手动管理：运行多个任务")
        task_b = loop.create_task(my_task("B")) # 创建一个任务
        task_c = loop.create_task(my_task("C"))
        loop.run_until_complete(asyncio.gather(task_b, task_c)) # 等待多个任务完成
    finally:
        loop.close() # 确保事件循环被关闭

if __name__ == "__main__":
    # asyncio.run(main_coroutine()) # 使用推荐方式
    manual_loop_management() # 使用手动方式

手动管理涉及以下步骤：

豆包AI编程

豆包推出的AI编程助手

483

查看详情

• asyncio.get_event_loop()

  登录后复制
  ：获取当前线程的事件循环。如果当前线程没有事件循环，它会创建一个新的。

• loop.run_until_complete(coroutine)

  登录后复制
  ：运行一个协程直到它完成。

• loop.close()

  登录后复制
  ：关闭事件循环。这会清除所有待处理的回调和任务，并释放资源。非常重要，否则可能导致资源泄露。

3. 有效地管理并发任务：

asyncio.create_task()

asyncio.gather()

仅仅启动事件循环是不够的，你还需要有效地管理多个并发执行的协程。

• asyncio.create_task(coroutine)

  登录后复制
  ： 当你想要一个协程在后台“独立”运行，而不是被

  await

  登录后复制
  立即阻塞时，你需要将它包装成一个

  Task

  登录后复制
  。

  create_task

  登录后复制
  会安排协程在事件循环中运行，并立即返回一个Task对象。你可以稍后

  await

  登录后复制
  这个Task对象来获取其结果或等待其完成。

  async def fetch_data(url):
      print(f"正在从 {url} 获取数据...")
      await asyncio.sleep(2) # 模拟网络延迟
      print(f"从 {url} 获取数据完成。")
      return f"Data from {url}"
  
  async def main_tasks():
      task1 = asyncio.create_task(fetch_data("http://example.com/api/1"))
      task2 = asyncio.create_task(fetch_data("http://example.com/api/2"))
  
      # 此时 task1 和 task2 已经开始在后台运行了
      print("任务已创建，正在等待结果...")
  
      result1 = await task1 # 等待 task1 完成
      result2 = await task2 # 等待 task2 完成
  
      print(f"收到结果: {result1}, {result2}")
  
  # asyncio.run(main_tasks())

  登录后复制

• *`asyncio.gather(coros_or_futures, loop=None, return_exceptions=False)`：** 这是一个非常实用的工具，用于同时运行并等待多个可等待对象（协程或Future）完成。它会收集所有结果，并按传入顺序返回。

  async def main_gather():
      results = await asyncio.gather(
          fetch_data("http://example.com/api/3"),
          fetch_data("http://example.com/api/4"),
          fetch_data("http://example.com/api/5")
      )
      print(f"所有数据都已获取: {results}")
  
  # asyncio.run(main_gather())

  登录后复制

  asyncio.gather()

  登录后复制
  在等待多个任务时非常方便，它能确保所有任务都被调度并等待其完成。

我的经验告诉我，事件循环是

asyncio

create_task

gather

async/await

登录后复制

在实际应用中，如何处理并发任务的异常和取消操作？

在实际的异步编程中，仅仅启动和等待任务是不够的，我们还需要妥善处理可能出现的异常和任务取消。这就像在现实世界中，计划总赶不上变化，我们需要有应对突发状况的机制。

1. 异常处理

异步任务中的异常处理与同步代码类似，主要通过

try...except

• 捕获单个协程的异常： 你可以在

  await

  登录后复制
  一个可能抛出异常的协程时，使用

  try...except

  登录后复制
  来捕获它。

  import asyncio
  
  async def might_fail_task(task_id):
      if task_id % 2 != 0:
          raise ValueError(f"任务 {task_id} 故意失败了！")
      await asyncio.sleep(1)
      return f"任务 {task_id} 成功完成。"
  
  async def handle_single_task_error():
      try:
          result = await might_fail_task(1) # 这个会失败
          print(result)
      except ValueError as e:
          print(f"捕获到异常: {e}")
  
      try:
          result = await might_fail_task(2) # 这个会成功
          print(result)
      except ValueError as e:
          print(f"捕获到异常: {e}") # 不会执行
  
  # asyncio.run(handle_single_task_error())

  登录后复制

• asyncio.gather()

  登录后复制
  中的异常：

  asyncio.gather()

  登录后复制
  在处理多个任务时，默认行为是“快速失败”（fail fast）。如果它所等待的任何一个子任务抛出了异常，

  gather

  登录后复制
  会立即取消所有其他未完成的任务，并重新抛出第一个遇到的异常。

  async def handle_gather_error_default():
      try:
          results = await asyncio.gather(
              might_fail_task(1), # 会失败
              might_fail_task(2), # 会成功，但可能被取消
              might_fail_task(3)  # 会失败，但可能在第一个异常抛出前被取消
          )
          print(f"所有任务完成: {results}")
      except ValueError as e:
          print(f"asyncio.gather 捕获到异常: {e}")
          # 此时，其他任务可能已经被取消了
  # asyncio.run(handle_gather_error_default())

  登录后复制

• asyncio.gather()

  登录后复制
  与

  return_exceptions=True

  登录后复制
  ： 如果你希望

  gather

  登录后复制
  等待所有任务完成，即使其中有任务失败，并将异常作为结果返回而不是直接抛出，你可以设置

  return_exceptions=True

  登录后复制
  。

  async def handle_gather_error_return_exceptions():
      results = await asyncio.gather(
          might_fail_task(1), # 失败
          might_fail_task(2), # 成功
          might_fail_task(3), # 失败
          return_exceptions=True # 关键参数
      )
      print("所有任务都处理完毕（包括失败的）:")
      for i, res in enumerate(results):
          if isinstance(res, Exception):
              print(f"  任务 {i+1} 失败: {res}")
          else:
              print(f"  任务 {i+1} 成功: {res}")
  # asyncio.run(handle_gather_error_return_exceptions())

  登录后复制

  这种方式在处理多个独立且不互相依赖的任务时非常有用，比如批量处理数据，你希望即使某些数据处理失败，也不影响其他数据的处理。

2. 任务取消

在异步编程中，任务取消是一个重要的概念。你可能需要停止一个长时间运行的任务，例如用户关闭了一个页面，或者一个操作超时了。

• task.cancel()

  登录后复制
  ： 调用一个

  Task

  登录后复制
  对象的

  cancel()

  登录后复制
  方法，会向该任务发送一个取消请求。这并不会立即停止任务，而是在任务内部（通常是在下一个

  await

  登录后复制
  点）抛出一个

  asyncio.CancelledError

  登录后复制
  异常。

  async def long_running_task(task_name):
      print(f"[{task_name}] 任务开始...")
      try:
          for i in range(5):
              print(f"[{task_name}] 工作中... {i+1}/5")
              await asyncio.sleep(1) # 这是一个可能的取消点
      except asyncio.CancelledError:
          print(f"[{task_name}] 任务被取消了！执行清理工作...")
      finally:
          print(f"[{task_name}] 任务结束（无论成功或取消）。")
  
  async def main_cancel():
      task = asyncio.create_task(long_running_task("A"))
      await asyncio.sleep(2.5) # 等待任务运行一段时间
      print("主程序：请求取消任务 A...")
      task.cancel() # 发送取消请求
  
      try:
          await task # 等待

  登录后复制

以上就是Python怎么使用async/await_Python异步编程async/await入门的详细内容，更多请关注php中文网其它相关文章！