如何使用asyncio进行异步编程？

asyncio通过协程实现单线程并发，适用于I/O密集型任务。使用async/await定义和调用协程，通过事件循环调度执行。可用asyncio.run()启动主协程，create_task()并发运行多个协程，gather()等待所有协程完成。异常处理需在await时捕获，未处理异常会存储于Task中。避免阻塞事件循环：使用异步I/O、将CPU密集型任务放入线程或进程池、用wait_for()设置超时、定期调用sleep(0)让出控制权。相比线程和进程，asyncio轻量高效，适合I/O密集场景；CPU密集任务应选进程或多线程；混合任务可结合使用三种方式。

Asyncio允许你编写并发代码，而无需线程或进程的复杂性。它本质上是一个单线程、事件循环的并发框架，通过协程（coroutines）实现非阻塞操作，从而提升I/O密集型任务的效率。

使用asyncio进行异步编程，简单来说，就是用

async

和

await

关键字定义和调用协程，并将它们注册到事件循环中执行。

解决方案：

1. 定义协程： 使用

   async def

   定义一个协程函数。协程可以暂停执行，等待I/O操作完成，而不会阻塞整个程序。

   import asyncio
   async def my_coroutine(delay):
   print(f"协程开始，等待 {delay} 秒")
   await asyncio.sleep(delay)
   print("协程结束")
   return f"等待了 {delay} 秒的结果"

2. 创建事件循环： asyncio的核心是事件循环，它负责调度和执行协程。

   loop = asyncio.get_event_loop()

3. 运行协程： 可以使用

   loop.run_until_complete()

   或

   asyncio.run()

   来运行协程。

   asyncio.run()

   简化了事件循环的创建和关闭，更适合简单的脚本。

   async def main():
   result = await my_coroutine(2)
   print(result)
   asyncio.run(main())

4. 使用

   await

   关键字：

   await

   关键字用于等待一个

   awaitable

   对象（例如另一个协程、

   Future

   或

   Task

   ）完成。

   await

   只能在

   async

   函数中使用。

   async def another_coroutine():
   await asyncio.sleep(1)
   return "另一个协程的结果"
   async def main():
   result = await another_coroutine()
   print(result)
   asyncio.run(main())

5. 创建Tasks：

   asyncio.create_task()

   函数可以创建一个

   Task

   对象，它代表一个正在运行的协程。 这允许你并发地运行多个协程。

   async def main():
   task1 = asyncio.create_task(my_coroutine(1))
   task2 = asyncio.create_task(my_coroutine(2))
   await task1
   await task2
   print("所有任务完成")
   asyncio.run(main())

6. 并发执行多个协程： 使用

   asyncio.gather()

   可以并发地运行多个协程，并等待它们全部完成。

   async def main():
   results = await asyncio.gather(
   my_coroutine(1),
   my_coroutine(2),
   another_coroutine()
   )
   print(results)
   asyncio.run(main())

如何处理asyncio中的异常？

在asyncio中处理异常与在常规Python代码中类似，但需要注意协程的特性。可以使用

try...except

块来捕获协程中抛出的异常。

import asyncio
async def might_fail():
await asyncio.sleep(0.5)
raise ValueError("Something went wrong!")
async def main():
try:
await might_fail()
except ValueError as e:
print(f"Caught an error: {e}")
asyncio.run(main())

如果一个任务（Task）内部抛出了未捕获的异常，该异常会被存储在Task对象中。当使用

await

等待该Task完成时，异常会被重新抛出。如果Task永远没有被

await

，则异常会被事件循环记录下来，但不会中断程序。

import asyncio
async def task_that_fails():
await asyncio.sleep(0.5)
raise ValueError("Task failed!")
async def main():
task = asyncio.create_task(task_that_fails())
await asyncio.sleep(1) # 让任务有时间完成
try:
await task # 重新抛出异常
except ValueError as e:
print(f"Caught error from task: {e}")
asyncio.run(main())

可以使用

task.result()

方法来获取Task的结果或异常。如果Task已经完成并且没有抛出异常，则返回结果。如果Task抛出了异常，则该方法会重新抛出异常。

import asyncio
async def task_that_fails():
await asyncio.sleep(0.5)
raise ValueError("Task failed!")
async def main():
task = asyncio.create_task(task_that_fails())
await asyncio.sleep(1)
try:
task.result() # 重新抛出异常
except ValueError as e:
print(f"Caught error from task: {e}")
asyncio.run(main())

asyncio中如何避免阻塞事件循环？

避免阻塞asyncio事件循环是保证程序响应性的关键。事件循环是单线程的，如果一个协程执行了耗时的同步操作，它会阻塞整个事件循环，导致其他协程无法运行。

1. 使用异步I/O操作： 这是最基本也是最重要的原则。使用asyncio提供的异步I/O操作，例如

   asyncio.sleep()

   、

   asyncio.open_connection()

   、

   aiohttp

   （异步HTTP客户端）等。这些操作会在等待I/O完成时释放事件循环的控制权，允许其他协程运行。

2. 避免CPU密集型任务： 如果需要执行CPU密集型任务（例如图像处理、加密解密等），应该将这些任务放到单独的进程或线程中执行，避免阻塞事件循环。可以使用

   asyncio.to_thread()

   或

   concurrent.futures

   模块来实现。

   import asyncio
   import concurrent.futures
   import time
   def cpu_bound_task(n):
   time.sleep(n)  # 模拟耗时操作
   return f"CPU密集型任务完成，耗时 {n} 秒"
   async def main():
   loop = asyncio.get_running_loop()
   with concurrent.futures.ProcessPoolExecutor() as pool:
   result = await loop.run_in_executor(pool, cpu_bound_task, 2) # 放在进程池中运行
   print(result)
   asyncio.run(main())

3. 限制协程的运行时间： 使用

   asyncio.wait_for()

   可以限制一个协程的运行时间。如果协程在指定时间内没有完成，

   asyncio.wait_for()

   会抛出一个

   asyncio.TimeoutError

   异常，可以捕获并处理该异常。

   import asyncio
   async def long_running_task():
   await asyncio.sleep(5)
   return "任务完成"
   async def main():
   try:
   result = await asyncio.wait_for(long_running_task(), timeout=2)
   print(result)
   except asyncio.TimeoutError:
   print("任务超时")
   asyncio.run(main())

4. 使用异步库： 尽量使用异步库来替代同步库。例如，使用

   aiohttp

   替代

   requests

   ，使用

   asyncpg

   或

   aiosqlite

   替代

   psycopg2

   或

   sqlite3

   。

5. 定期释放控制权： 如果一个协程需要执行大量的计算或I/O操作，可以定期使用

   await asyncio.sleep(0)

   来释放事件循环的控制权，让其他协程有机会运行。

   import asyncio
   async def large_task():
   for i in range(100000):
   # 模拟大量计算
   _ = i * i
   if i % 1000 == 0:
   await asyncio.sleep(0)  # 释放控制权
   return "任务完成"
   async def main():
   result = await large_task()
   print(result)
   asyncio.run(main())

如何选择合适的并发策略：asyncio vs 线程 vs 进程？

选择合适的并发策略取决于任务的性质和程序的需求。Asyncio、线程和进程各有优缺点，适用于不同的场景。

• Asyncio：

  • 优点：
    • 轻量级： 基于协程，切换开销小，资源占用少。
    • 高效I/O： 擅长处理I/O密集型任务，例如网络请求、文件读写等。
    • 单线程： 避免了线程锁的复杂性，减少了死锁和竞争条件。
  • 缺点：
    • CPU密集型： 不适合CPU密集型任务，因为单线程无法充分利用多核CPU。
    • 阻塞操作： 如果有阻塞操作，会阻塞整个事件循环，影响程序的响应性。
  • 适用场景：
    • Web服务器、网络爬虫、聊天服务器、实时数据处理等I/O密集型应用。

• 线程：

  • 优点：
    • 并发执行： 可以并发执行CPU密集型任务，充分利用多核CPU。
    • 共享内存： 线程之间可以共享内存，方便数据交换。
  • 缺点：
    • 线程锁： 需要使用线程锁来保护共享资源，增加了编程复杂性。
    • GIL限制： 在CPython中，由于全局解释器锁（GIL）的存在，同一时刻只能有一个线程执行Python字节码，限制了CPU密集型任务的并发性能。
    • 上下文切换： 线程切换开销比协程大。
  • 适用场景：
    • GUI程序、多线程下载器、需要并发执行CPU密集型任务的应用。

• 进程：

  • 优点：
    • 真正的并行： 可以利用多核CPU实现真正的并行，不受GIL限制。
    • 隔离性： 进程之间相互隔离，一个进程崩溃不会影响其他进程。
  • 缺点：
    • 资源占用： 进程创建和销毁开销大，资源占用多。
    • 通信开销： 进程间通信（IPC）开销比线程大。
  • 适用场景：
    • 科学计算、图像处理、视频编码等CPU密集型应用，需要高可靠性和隔离性的应用。

总结：

• I/O密集型任务： 优先选择asyncio，如果asyncio遇到阻塞操作，可以考虑使用线程池或进程池来执行阻塞操作。
• CPU密集型任务： 优先选择进程，如果需要共享内存，可以考虑使用线程。
• 混合型任务： 可以结合使用asyncio、线程和进程，例如使用asyncio处理I/O，使用线程或进程处理CPU密集型任务。