1. 引言
在现代软件开发中，处理高并发任务已成为常态。Python，作为一种广泛使用的高级编程语言，提供了强大的并发模型，其中协程是关键组件。本文将深入探讨Python中的协程，从基础概念到高级应用，以及它们在实际开发中的使用。
2. 协程的基本概念
在深入探讨Python协程之前，我们需要理解一些基本的概念。协程是一种程序组件，它允许挂起和恢复执行，这使得协程非常适合处理并发任务，尤其是在I/O密集型应用中。
2.1 协程与线程的区别
在传统的多线程编程中，操作系统负责线程的调度，这涉及到上下文切换的开销。协程则由程序自身管理，因此上下文切换的开销要小得多。协程通常用于协程之间不会阻塞对方的任务，而线程则适用于CPU密集型任务。
2.2 协程的工作原理
协程通过yield关键字实现，Python 2中的生成器就是协程的一种形式。在Python 3中，引入了async和await关键字，使得协程的编写更加直观和方便。
2.3 Python中的yield和yield from
在Python 2中，yield关键字用于创建生成器，这些生成器可以被用作协程。而在Python 3中，yield from允许一个生成器代理另一个生成器的值，这在编写协程时非常有用。
def simple_coroutine():
print('-> Coroutine started')
x = yield
print('-> Coroutine received:', x)
my_coro = simple_coroutine()
next(my_coro) # 启动协程
my_coro.send(42) # 向协程发送值
2.4 使用async和await
Python 3.5及以后版本中，可以使用async def来定义协程函数，使用await来挂起当前协程的执行，等待另一个协程完成。
async def hello():
print('Hello')
async def main():
await hello() # 等待hello协程完成
import asyncio
asyncio.run(main())
2.5 协程的执行流程
协程的执行流程通常由事件循环控制。事件循环负责分发事件和任务，以及管理协程的执行。
async def count_down(seconds):
while seconds > 0:
print(seconds)
seconds -= 1
await asyncio.sleep(1) # 模拟I/O操作，挂起当前协程
asyncio.run(count_down(5))
2.6 协程的异常处理
异常处理在协程中同样重要，可以使用传统的try/except语句来捕获和处理异常。
async def might_fail():
try:
# 模拟可能失败的操作
raise ValueError('Something went wrong')
except ValueError as e:
print(f'Caught an exception: {e}')
asyncio.run(might_fail())
3. Python协程的历史
Python协程的发展是一个逐渐演进的过程，它与Python语言的发展紧密相连。了解这一历史背景有助于我们更好地理解协程的概念及其在现代Python编程中的应用。
3.1 早期的生成器和迭代器
在Python 2时代，生成器（generator）是协程的前身。它们通过yield语句产生一个值，然后挂起，等待下一次迭代。生成器主要用于简化迭代逻辑，同时节省内存。
def simple_generator():
yield 1
yield 2
yield 3
for value in simple_generator():
print(value)
3.2 yield from的引入
Python 3引入了yield from语法，它允许一个生成器将另一个生成器的输出作为自己的输出。这为构建更复杂的生成器提供了便利。
def nested_generator():
yield from [1, 2, 3]
for value in nested_generator():
print(value)
3.3 asyncio模块的诞生
随着Python 3.3的发布，asyncio模块正式成为Python标准库的一部分。asyncio是一个用于编写单线程并发代码的库，它使用async和await语法来定义和调度协程。
import asyncio
async def async_task():
print('Task started')
await asyncio.sleep(1)
print('Task completed')
loop = asyncio.get_event_loop()
loop.run_until_complete(async_task())
3.4 async和await的语法糖
Python 3.5进一步简化了协程的编写，引入了async和await作为语法糖。这使得编写异步代码更加直观和易于理解。
async def async_greet(name):
await asyncio.sleep(1) # 模拟异步操作
print(f'Hello, {name}!')
async def main():
await async_greet('World')
asyncio.run(main())
3.5 协程的演进
随着Python版本的更新，协程的功能也在不断增强。例如，Python 3.6引入了async for和async with语法，使得异步迭代和上下文管理更加方便。
async def async_iter():
yield 1
yield 2
yield 3
async def main():
async for value in async_iter():
print(value)
asyncio.run(main())
3.6 社区和生态系统的发展
随着Python协程的普及，社区中涌现出许多优秀的第三方库，如aiohttp用于异步HTTP请求，SQLAlchemy的异步支持等。这些库进一步扩展了协程的应用范围。
4. 创建协程
在Python中，创建协程是一个简单的过程，但要充分利用它们，就需要深入理解async和await的使用。本节将详细介绍如何定义协程，以及如何通过asyncio模块来运行它们。
4.1 定义协程函数
使用async def关键字可以定义一个协程函数。这个关键字定义了一个异步函数，它可以包含await表达式。
async def simple_coroutine():
print('Coroutine started')
await asyncio.sleep(1) # 模拟异步操作
print('Coroutine finished')
4.2 启动和运行协程
定义协程后，你需要通过事件循环来启动和运行它。asyncio.run()是一个方便的函数，用于运行最高级别的协程。
import asyncio
asyncio.run(simple_coroutine())
4.3 协程的await表达式
await关键字用于挂起协程的执行，直到等待的协程或异步操作完成。这允许事件循环运行其他任务。
async def coroutine_with_await():
print('Before await')
await asyncio.sleep(2)
print('After await')
4.4 协程的组合
你可以使用asyncio.gather()来同时运行多个协程，这在并行处理多个异步任务时非常有用。
async def coroutine_a():
await asyncio.sleep(1)
return 'Result A'
async def coroutine_b():
await asyncio.sleep(2)
return 'Result B'
async def main():
results = await asyncio.gather(coroutine_a(), coroutine_b())
print(results)
asyncio.run(main())
4.5 错误处理
在协程中处理错误与同步代码类似，使用try/except语句来捕获异常。
async def coroutine_with_errors():
try:
raise ValueError('An error occurred')
except ValueError as e:
print(f'Caught an exception: {e}')
asyncio.run(coroutine_with_errors())
4.6 协程的取消
协程可以被取消，如果协程中的任务无法完成或者不再需要。
async def cancellable_coroutine():
try:
while True:
print('Loop running')
await asyncio.sleep(1)
except asyncio.CancelledError:
print('Coroutine cancelled')
async def main():
task = asyncio.create_task(cancellable_coroutine())
await asyncio.sleep(5)
task.cancel()
asyncio.run(main())
4.7 使用asyncio.wait_for超时控制
asyncio.wait_for可以用来设置协程的超时时间，如果协程在指定时间内未完成，则会抛出asyncio.TimeoutError。
async def long_running_coroutine():
await asyncio.sleep(10) # 模拟长时间运行的任务
async def main():
try:
await asyncio.wait_for(long_running_coroutine(), timeout=2)
except asyncio.TimeoutError:
print('Coroutine timed out')
asyncio.run(main())
5. 协程的执行控制
在Python中，协程的执行控制是通过asyncio模块实现的，该模块提供了一套完整的框架来处理异步任务的调度和执行。本节将详细介绍事件循环、任务(Task)、Future对象以及异常处理等概念，并提供丰富的示例。
5.1 事件循环
事件循环是asyncio模块的核心，负责运行和管理协程的执行。事件循环会不断地检查协程的状态，并在适当的时候执行它们。
import asyncio
async def say_after(delay, what):
await asyncio.sleep(delay)
print(what)
async def main():
task1 = asyncio.create_task(say_after(1, 'hello'))
task2 = asyncio.create_task(say_after(2, 'world'))
await task1
await task2
asyncio.run(main())
5.2 创建和管理任务(Task)
任务是协程的一个封装，它允许将协程作为后台任务运行，并可以查询任务的状态或取消任务。
async def main():
task = asyncio.create_task(asyncio.sleep(5))
print(f'Task is running: {task.running()}')
await asyncio.sleep(3)
task.cancel()
try:
await task
except asyncio.CancelledError:
print('Task was cancelled')
asyncio.run(main())
5.3 Future对象
Future是一个特殊类型的asyncio对象，代表一个尚未完成的操作。它可以在操作完成时存储结果或异常。
async def main():
future = asyncio.Future()
async def set_value():
await asyncio.sleep(1)
future.set_result('Future result')
asyncio.create_task(set_value())
result = await future
print(result)
asyncio.run(main())
5.4 异常处理
在协程中处理异常与同步代码类似，使用try/except语句来捕获和处理异常。
async def main():
try:
await asyncio.sleep(1)
raise ValueError('An error occurred')
except ValueError as e:
print(f'Caught an exception: {e}')
asyncio.run(main())
5.5 协程的超时控制
使用asyncio.wait_for可以为协程设置超时时间，如果协程在指定时间内未完成，则会抛出asyncio.TimeoutError。
async def long_running_task():
await asyncio.sleep(10)
async def main():
try:
await asyncio.wait_for(long_running_task(), timeout=2)
except asyncio.TimeoutError:
print('Task timed out')
asyncio.run(main())
5.6 协程的并发执行
asyncio.gather可以用来并发执行多个协程，并等待它们全部完成。
async def task1():
await asyncio.sleep(2)
return 'result1'
async def task2():
await asyncio.sleep(1)
return 'result2'
async def main():
results = await asyncio.gather(task1(), task2())
print(results)
asyncio.run(main())
5.7 协程的优先级调度
asyncio模块允许设置协程的优先级，优先级高的协程会更频繁地被事件循环调度。
async def high_priority_task():
for _ in range(5):
print('High priority task')
await asyncio.sleep(0.2)
async def low_priority_task():
for _ in range(5):
print('Low priority task')
await asyncio.sleep(0.2)
async def main():
loop = asyncio.get_running_loop()
loop.create_task(high_priority_task(), name='high_priority')
loop.create_task(low_priority_task(), name='low_priority')
asyncio.run(main())
6. 协程的高级用法
在Python中，协程的高级用法涉及到更复杂的异步编程模式，包括协程的组合、并发执行、状态管理以及与同步代码的交互。本节将深入探讨这些高级概念，并提供丰富的示例。
6.1 协程的组合：asyncio.gather
asyncio.gather是处理多个协程并发执行的强大工具。它可以等待多个协程完成，并返回所有协程的结果。
async def fetch_data(source):
await asyncio.sleep(1)
return f"Data from {source}"
async def main():
results = await asyncio.gather(
fetch_data('source1'),
fetch_data('source2'),
fetch_data('source3')
)
print(results)
asyncio.run(main())
6.2 使用asyncio.wait
与asyncio.gather不同，asyncio.wait可以用来等待多个协程中的一个或多个完成，而不必等待所有协程。
async def main():
tasks = [asyncio.create_task(fetch_data(source)) for source in ['source1', 'source2', 'source3']]
done, _ = await asyncio.wait(tasks, timeout=2, return_when=asyncio.FIRST_COMPLETED)
for task in done:
print(await task)
asyncio.run(main())
6.3 协程与同步函数的交互
在某些情况下，我们需要在协程中调用同步函数。asyncio.to_thread函数可以用来在协程中运行同步函数，而不会阻塞事件循环。
import asyncio
import time
def sync_function():
time.sleep(2) # 同步阻塞调用
return "Synchronous result"
async def main():
result = await asyncio.to_thread(sync_function)
print(result)
asyncio.run(main())
6.4 异步上下文管理器：asyncio.Lock
在并发编程中，经常需要对共享资源进行同步访问。asyncio.Lock提供了一种异步的互斥机制，以确保同时只有一个协程可以访问特定的代码段。
async def main():
lock = asyncio.Lock()
async def task(name):
async with lock:
print(f"Task {name} is running")
await asyncio.sleep(1)
print(f"Task {name} is done")
tasks = [asyncio.create_task(task(i)) for i in range(5)]
await asyncio.gather(*tasks)
asyncio.run(main())
6.5 异步生成器
Python 3.6引入了异步生成器，允许协程在执行过程中产生值，而不必一次性产生所有值。
async def async_generator():
for i in range(3):
yield f"Value {i}"
await asyncio.sleep(1)
async def main():
async for value in async_generator():
print(value)
asyncio.run(main())
6.6 异步迭代器和async for
与异步生成器相对应，async for可以用来迭代异步生成器产生的值。
# 继续使用上面定义的 async_generator 函数
asyncio.run(main())
6.7 任务的取消和超时
任务的取消和超时是协程执行控制的重要方面。可以通过设置超时或在事件循环中显式取消任务来处理长时间运行的任务。
async def main():
task = asyncio.create_task(asyncio.sleep(10))
asyncio.create_task(asyncio.sleep(1)).cancel() # 立即取消
try:
await asyncio.wait_for(task, timeout=5)
except asyncio.TimeoutError:
print("Task was cancelled due to timeout")
task.cancel()
try:
await task
except asyncio.CancelledError:
print("Task was successfully cancelled")
asyncio.run(main())