title: 深入理解Python协程：从基础到实战
date: 2024/4/27 16:48:43
updated: 2024/4/27 16:48:43
categories:
tags:
协程（Coroutines）是一种特殊的软件构造，它允许程序在执行过程中暂停并恢复执行，而不会丢失当前的执行上下文。与线程和进程不同，协程在单个线程中运行，通过调度机制实现并发，降低了上下文切换的开销，提高了程序的执行效率。协程通常用于处理I/O密集型任务，如网络请求、文件读写等。
生成器（Generators）是Python中实现协程的一种方式，它通过内置的yield关键字来暂停和恢复执行。当函数遇到yield
时，会暂停执行并返回一个值，下次调用时会从上次暂停的地方继续执行。yield
实际上是一个特殊的return语句，它会保存当前的状态（包括局部变量和执行上下文），当再次调用时，这些状态会被恢复。
def coroutine_example():
value = yield 0
print(f'Received value: {value}')
value = yield 1
print(f'Received value: {value}')
c = coroutine_example()
next(c) # 输出 'Received value: 0'
print(c.send(2)) # 输出 'Received value: 1'
yield关键字。next()或send()方法开始执行，直到遇到yield。yield时，函数暂停，保存当前状态。send()方法传入值，函数从上次暂停的地方继续执行。yield可执行，或遇到return语句时，协程结束。asyncio是 Python 中用于编写异步代码的标准库，它提供了一组工具和API来管理和调度协程。通过asyncio，可以轻松创建、执行和管理异步任务。
import asyncio
async def async_function():
await asyncio.sleep(1)
print("Async function executed")
asyncio.run(async_function())
async关键字用于定义异步函数，await关键字用于暂停异步函数的执行，等待另一个异步任务完成。
import asyncio
async def async_function():
await asyncio.sleep(1)
print("Async function executed")
asyncio.run(async_function())
asyncio提供了事件循环（Event Loop）来调度协程的执行。事件循环负责管理和调度所有的协程任务，确保它们按照正确的顺序执行。
import asyncio
async def task():
print("Task executed")
async def main():
await asyncio.gather(task(), task())
asyncio.run(main())
import asyncio
import aiohttp
async def fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
urls = ['http://example.com', 'http://example.org']
tasks = [fetch(url) for url in urls]
responses = await asyncio.gather(*tasks)
for response in responses:
print(response)
asyncio.run(main())
这个例子演示了如何使用asyncio和aiohttp库进行异步的网络请求处理。fetch()函数负责发送 HTTP
请求并返回响应内容，main()函数创建了多个任务，并使用asyncio.gather()并行执行这些任务，最后输出每个请求的响应内容。
异步编程是一种编程范式，允许程序在等待某些操作完成的同时继续执行其他任务，而不会被阻塞。相比于传统的同步编程方式，异步编程具有以下优势：
协程是一种轻量级的线程，可以在执行过程中暂停并恢复。在Python中，协程通过async和await
关键字实现，是异步编程的关键技术之一。协程的实现原理包括以下几个关键点：
async def定义的函数可以在函数内部使用await关键字来挂起函数的执行，等待异步操作完成。asyncio库提供了事件循环的支持。异步事件循环是异步编程中的核心概念，负责协调和调度异步任务的执行。其原理包括以下几个关键点：
异步事件循环的作用在于提供一个统一的调度器，使得异步任务能够在不同的协程之间切换执行，实现非阻塞的并发处理。
任务池是一种管理和调度异步任务的机制，用于管理大量的异步任务并控制其并发执行。任务池在异步编程中具有以下重要性：
任务池在异步编程中扮演着重要的角色，能够有效管理和调度大量的异步任务，提高程序的效率和可靠性。
下面是一个简单的示例，演示如何使用asyncio库创建和管理任务集合：
import asyncio
async def task(num):
print(f"Task {num} started")
await asyncio.sleep(1)
print(f"Task {num} completed")
async def main():
tasks = [task(i) for i in range(3)] # 创建多个任务
await asyncio.gather(*tasks) # 等待所有任务完成
if __name__ == "__main__":
asyncio.run(main()) # 运行主函数
在这个示例中，我们定义了一个异步任务task，然后在main函数中创建了多个任务，并使用asyncio.gather
来等待所有任务完成。最后通过asyncio.run来运行主函数。这样就实现了使用asyncio库创建和管理任务集合的功能。
协程池是一种用于管理和调度协程执行的机制，可以控制并发度、减少资源占用和提高程序性能。协程池在并发编程中的作用和优化策略包括：
优化策略包括合理设置协程池的大小、避免阻塞操作、及时处理协程的返回值等，以提高程序的效率和性能。
资源管理在并发编程中非常重要，可以避免资源泄露和提高程序的稳定性。避免资源泄露的方法包括：
with语句可以确保资源在使用完毕后及时释放。良好的资源管理能够避免资源泄露和提高程序的稳定性，确保程序的正常运行。
在异步编程中，管理协程的取消和异常处理是非常重要的，可以提高程序的健壮性。有效管理协程的取消和异常处理包括：
asyncio.Task.cancel()方法可以取消正在执行的协程，避免不必要的资源消耗。try-except语句捕获异常，并根据实际情况处理异常，避免程序崩溃。asyncio.create_task()创建任务，并在任务中统一处理异常，以确保程序的稳定性。通过合理取消协程和处理异常，可以有效管理协程的执行过程，提高程序的可靠性和健壮性。
异步编程在Web服务器中的应用可以显著提高性能，因为它允许服务器在等待客户端响应时处理其他请求，而不是阻塞。这种方式提高了服务器的并发处理能力，使得在高负载情况下也能保持良好的响应速度。
aiohttp是一个用于构建高性能HTTP/HTTPS服务器和客户端的Python库，它非常适合异步IO操作。下面是一个简单的aiohttp异步Web服务器示例：
import asyncio
from aiohttp import web
runner = None # 定义全局变量 runner
async def handle_request(request):
name = request.match_info.get('name', 'World')
text = f'Hello, {name}!'
return web.Response(text=text)
async def run_app(app):
global runner # 声明使用全局变量 runner
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, '127.0.0.1', 8080)
await site.start()
async def main():
app = web.Application()
app.router.add_get('/{name}', handle_request)
try:
print('Server started at http://127.0.0.1:8080')
await run_app(app)
except KeyboardInterrupt:
pass
finally:
if runner is not None: # 检查 runner 是否已初始化
await runner.cleanup() # 使用 runner.cleanup() 替代 runner.shutdown()
if __name__ == '__main__':
asyncio.run(main()) # 使用 asyncio.run() 简化事件循环管理
在这个例子中，handle_request函数是协程，它接收一个请求，处理并返回响应。app()函数创建了一个应用实例，添加路由，并启动一个事件循环来监听请求。
web.Request对象和web.View接口都是异步的，通过async def定义的函数处理请求，可以在处理过程中执行其他协程，提高效率。asyncio.get_event_loop()获取事件循环，它负责调度协程的执行，当有新的请求到达时，它会将请求添加到任务队列中，等待调度。通过这种方式，aiohttp可以实现高效的Web服务器，提高并发处理能力，同时避免了阻塞，使得服务器在高负载下仍能保持良好的性能。
在异步IO中，文件操作和Socket编程是常见的任务，可以通过协程实现异步处理以提高效率。
import asyncio
async def read_file_async(file_path):
async with open(file_path, 'r') as file:
data = await file.read()
return data
async def write_file_async(file_path, data):
async with open(file_path, 'w') as file:
await file.write(data)
# 使用示例
async def main():
data = await read_file_async('example.txt')
await write_file_async('example_copy.txt', data)
asyncio.run(main())
import asyncio
async def handle_client(reader, writer):
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print(f"Received {message} from {addr}")
print(f"Send: {message}")
writer.write(data)
await writer.drain()
print("Closing the connection")
writer.close()
async def main():
server = await asyncio.start_server(
handle_client, '127.0.0.1', 8888)
addr = server.sockets[0].getsockname()
print(f'Serving on {addr}')
async with server:
await server.serve_forever()
asyncio.run(main())
数据库操作通常涉及磁盘IO和网络IO，因此异步编程在此领域尤为重要。常见的数据库操作库如asyncpg、aiomysql等都提供了异步接口。
import asyncio
import asyncpg
async def fetch_data():
conn = await asyncpg.connect(user='user', password='password',
database='database', host='127.0.0.1')
values = await conn.fetch('''SELECT * FROM table''')
await conn.close()
return values
async def main():
data = await fetch_data()
print(data)
asyncio.run(main())
import asyncio
import asyncpg
async def fetch_data_and_write_to_file():
conn = await asyncpg.connect(user='user', password='password',
database='database', host='127.0.0.1')
values = await conn.fetch('''SELECT * FROM table''')
await conn.close()
async with open('database_data.txt', 'w') as file:
for row in values:
file.write(str(row) + '\n')
async def main():
await fetch_data_and_write_to_file()
asyncio.run(main())
在这个示例中，我们连接到数据库，从表中检索数据，然后将数据写入到文件中。所有这些操作都是异步的，通过协程实现了非阻塞的数据库操作和文件IO。
在协程中，为了避免并发访问共享资源时出现数据竞争的情况，可以使用锁（Lock）等同步原语来实现线程间的互斥。
import asyncio
async def task(lock):
async with lock:
# 访问共享资源的代码
print("Accessing shared resource")
await asyncio.sleep(1)
print("Finished accessing shared resource")
async def main():
lock = asyncio.Lock()
tasks = [task(lock) for _ in range(5)]
await asyncio.gather(*tasks)
asyncio.run(main())
在上面的示例中，通过asyncio.Lock()创建了一个锁对象，然后在协程中使用async with lock来获取锁。这样可以保证同一时刻只有一个协程可以访问共享资源。
在Python的asyncio模块中，并发控制通常通过asyncio.Lock来实现，而不是使用传统的指针锁。asyncio.Lock
是基于协程的锁，可以在协程中使用async with lock语法来实现锁定和释放。
import asyncio
async def task(lock):
async with lock:
# 访问共享资源的代码
print("Accessing shared resource")
await asyncio.sleep(1)
print("Finished accessing shared resource")
async def main():
lock = asyncio.Lock()
tasks = [task(lock) for _ in range(5)]
await asyncio.gather(*tasks)
asyncio.run(main())
import asyncio
shared_resource = 0
lock = asyncio.Lock()
async def update_shared_resource():
global shared_resource
async with lock:
shared_resource += 1
async def main():
tasks = [update_shared_resource() for _ in range(10)]
await asyncio.gather(*tasks)
print(f"Final shared resource value: {shared_resource}")
asyncio.run(main())
在这个示例中，多个协程同时更新共享资源shared_resource，通过asyncio.Lock实现并发控制，确保共享资源的安全访问。最终输出的共享资源值应为10，每个协程更新一次。
协程链（Coroutine
Chain）是一种将多个协程按照顺序连接起来的并发编程模式，每个协程负责处理一部分任务。流水线模式（Pipeline）是协程链的一种特例，它将数据流通过一系列协程进行处理，每个协程只负责处理特定的数据处理步骤。
import asyncio
async def coroutine1(data):
# 处理数据
print(f"Coroutinue1: {data}")
await asyncio.sleep(0.1)
return data * 2
async def coroutine2(data):
# 处理数据
print(f"Coroutinue2: {data}")
await asyncio.sleep(0.1)
return data ** 2
async def main():
data = 1
coroutines = [coroutine1, coroutine2]
for coroutine in coroutines:
data = await coroutine(data)
print(f"Final result: {data}")
asyncio.run(main())
在上面的示例中，我们创建了两个协程coroutine1和coroutine2，将它们按照顺序连接起来，形成一个协程链。数据在协程链中流动，每个协程负责处理特定的数据处理步骤。
事件驱动架构（Event-Driven Architecture）是一种基于事件的并发编程模式，它将应用程序分解为多个独立的事件处理器，每个事件处理器负责处理特定的事件。当事件发生时，事件处理器会被激活并执行相应的处理逻辑。
import asyncio
async def handle_event(event):
# 处理事件
print(f"Handling event: {event}")
await asyncio.sleep(0.1)
async def main():
events = ["event1", "event2", "event3"]
tasks = [handle_event(event) for event in events]
await asyncio.gather(*tasks)
asyncio.run(main())
在上面的示例中，我们定义了一个handle_event协程来处理事件。在main函数中，我们创建了三个事件event1、event2和event3
，然后为每个事件创建一个任务，并使用asyncio.gather同时运行这些任务。这样，基于协程的事件驱动架构可以实现并发处理多个事件。
基于协程的实时数据处理是一种利用协程实现数据流处理的并发编程模式，可以实现高效的数据处理和实时响应。
import asyncio
async def process_data(data):
# 处理数据
print(f"Processing data: {data}")
await asyncio.sleep(0.1)
return data.upper()
async def main():
data_stream = ["data1", "data2", "data3"]
tasks = [process_data(data) for data in data_stream]
processed_data = await asyncio.gather(*tasks)
print(f"Processed data: {processed_data}")
asyncio.run(main())
在上面的示例中，我们定义了一个process_data协程来处理数据。在main函数中，我们创建了一个数据流data_stream
，并为每个数据创建一个处理任务。使用asyncio.gather可以同时运行这些处理任务，并等待它们完成。最终，我们可以得到处理后的数据流。
在爬虫中使用协程可以提高爬取效率，协程调度可以使爬虫程序更加高效地处理多个任务。以下是一个简单的爬虫示例，使用协程和异步IO库aiohttp：
import asyncio
import aiohttp
async def fetch_url(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
urls = ["http://example.com/page1", "http://example.com/page2", "http://example.com/page3"]
tasks = [fetch_url(url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
asyncio.run(main())
在上面的示例中，我们定义了一个fetch_url协程来获取URL的内容。在main
函数中，我们创建了多个URL的任务，并使用asyncio.gather同时运行这些任务。这样，爬虫可以并发地获取多个URL的内容，提高爬取效率。
使用协程可以构建高性能的Web服务器，以下是一个简单的基于协程的Web服务器示例：
from aiohttp import web
async def handle(request):
return web.Response(text="Hello, World!")
app = web.Application()
app.router.add_get('/', handle)
web.run_app(app)
在上面的示例中，我们定义了一个处理函数handle来处理HTTP请求，并创建了一个web.Application应用。通过app.router.add_get
将处理函数绑定到根路径'/'，最后使用web.run_app来运行Web服务器。
构建一个简单的异步HTTP客户端可以帮助我们实现高效的HTTP请求。以下是一个简单的异步HTTP客户端示例：
import aiohttp
import asyncio
async def fetch_url(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
url = "http://example.com"
response = await fetch_url(url)
print(response)
asyncio.run(main())
在上面的示例中，我们定义了一个fetch_url协程来获取URL的内容。在main函数中，我们发起了一个HTTP
GET请求，并等待响应。这样，我们可以实现异步地获取URL的内容。
Python 3.7版本及以上版本对async/await语法进行了改进，使得使用协程更加方便和高效。以下是一些Python
3.7及以上版本中async/await语法的改进：
async with和async for语句，使得使用协程可以更加方便和高效。async for语句的async for ... of语法，使得在协程中使用生成器更加简单和高效。async def语句，使得定义协程更加简单和直观。await语句的await expression语法，使得在协程中等待异步操作更加简单和高效。asyncio库中的asyncio.run函数，使得运行协程更加简单和高效。在现代Python生态系ystem中，协程已经成为一个非常重要的并发编程模型。以下是协程在现代Python生态系统中的一些角色：
在本文中，我们介绍了协程的基本概念和使用方法，并结合实际案例展示了协程在实际应用中的优势和应用场景。如果您想进一步学习协程，可以参考以下资源：
asyncio和aiohttp章节。首页 | 一个覆盖广泛主题工具的高效在线平台(amd794.com)
asyncio是Python 3.4版本引入的一个标准库，用于实现异步IO操作和并发编程。asyncio
基于协程实现，提供了许多高级API和工具，使得开发人员可以快速构建高效的异步IO应用。
aiohttp是一个基于asyncio实现的异步HTTP客户端和服务器库。aiohttp支持协程，提供了许多高级API和工具，使得开发人员可以快速构建高效的异步Web应用。
trio是一个基于协程实现的异步IO操作和并发编程库，与asyncio类似，但提供了更加简单和高效的API和工具。trio
支持多个事件循环，可以更加灵活和高效地管理协程。
curio是一个基于协程实现的异步IO操作和并发编程库，与asyncio类似，但提供了更加简单和高效的API和工具。curio
支持多个事件循环，可以更加灵活和高效地管理协程。
Sanic是一个基于aiohttp实现的异步Web框架，支持协程，提供了许多高级API和工具，使得开发人员可以快速构建高效的异步Web应用。
调试协程可能会比调试同步代码更加复杂，因为协程的执行流程更加复杂。以下是一些调试协程的技巧和工具：
pdb调试器：pdb是Python的标准调试器，可以用于调试协程。asyncio提供的asyncio.get_event_loop()函数获取当前事件循环，并使用loop.run_until_complete()函数运行协程。asyncio提供的asyncio.create_task()函数创建一个新的任务，并使用asyncio.gather()函数等待所有任务完成。asyncio提供的asyncio.as_completed()函数按照完成顺序获取任务的结果。asyncio提供的asyncio.wait()函数等待所有任务完成，并获取完成和未完成的任务列表。优化协程的性能可能会比优化同步代码更加复杂，因为协程的执行流程更加复杂。以下是一些优化协程性能的技巧和工具：
asyncio.gather()函数并行执行多个任务，提高IO操作的效率。asyncio.sleep()函数减少CPU占用，提高IO操作的效率。asyncio.wait()函数并行执行多个任务，并获取完成和未完成的任务列表，提高IO操作的效率。asyncio.as_completed()函数按照完成顺序获取任务的结果，提高IO操作的效率。asyncio.Queue和asyncio.Semaphore限制并发数，提高IO操作的效率。协程是一种轻量级的线程，可以在单个线程中实现多个任务的并发执行。
使用协程可以实现高效的异步IO操作和并发编程，提高IO操作的效率。
使用协程需要使用async和await关键字，定义一个协程函数，并使用asyncio库中的asyncio.run()函数运行协程。
使用await关键字可以在协程中等待异步操作，直到操作完成。
使用asyncio.create_task()函数可以在协程中创建一个新的任务。
使用asyncio.gather()函数可以在协程中等待多个任务完成。
使用asyncio.as_completed()函数可以在协程中按照完成顺序获取任务的结果。
使用asyncio.Queue和asyncio.Semaphore可以在协程中限制并发数。
使用pdb调试器、asyncio.get_event_loop()函数、asyncio.create_task()函数、asyncio.gather()
函数、asyncio.as_completed()函数和asyncio.wait()函数可以调试协程。
使用asyncio.gather()函数、asyncio.sleep()函数、asyncio.wait()函数、asyncio.as_completed()函数和asyncio.Queue
和asyncio.Semaphore可以优化协程性能。
使用try和except语句可以在协程中处理异常。如果在协程中发生异常，可以使用asyncio.exceptions.AsyncioFuture.get_result()
函数获取异常信息。
使用asyncio.wait_for()函数可以在协程中实现超时。如果在超时时间内未完成，可以使用asyncio.wait_for()函数中的timeout
参数设置超时时间。
使用asyncio.create_task()函数和asyncio.sleep()
函数可以在协程中实现定时任务。可以在协程中创建一个新的任务，并使用asyncio.sleep()函数设置定时时间。
使用asyncio.create_task()函数和asyncio.sleep()
函数可以在协程中实现循环任务。可以在协程中创建一个新的任务，并使用asyncio.sleep()函数设置循环时间。
使用asyncio.Semaphore可以在协程中实现并发限制。可以在协程中创建一个asyncio.Semaphore
对象，并使用asyncio.Semaphore.acquire()函数获取信号量，使用asyncio.Semaphore.release()函数释放信号量。
使用asyncio.PriorityQueue可以在协程中实现任务优先级。可以在协程中创建一个asyncio.PriorityQueue
对象，并使用asyncio.PriorityQueue.put()函数添加任务，使用asyncio.PriorityQueue.get()函数获取优先级最高的任务。
使用asyncio.create_task()函数和asyncio.Task.cancel()
函数可以在协程中实现任务取消。可以在协程中创建一个新的任务，并使用asyncio.Task.cancel()函数取消任务。
使用asyncio.wait_for()函数和asyncio.Task.cancel()
函数可以在协程中实现任务超时。可以在协程中创建一个新的任务，并使用asyncio.wait_for()
函数设置超时时间，如果在超时时间内未完成，可以使用asyncio.Task.cancel()函数取消任务。
使用asyncio.Queue可以在协程中实现任务队列。可以在协程中创建一个asyncio.Queue对象，并使用asyncio.Queue.put()
函数添加任务，使用asyncio.Queue.get()函数获取任务。
使用asyncio.gather()函数可以在协程中实现任务分组。可以在协程中使用asyncio.gather()
函数分组多个任务，并使用asyncio.gather()函数中的return_exceptions参数设置是否返回异常信息。
在 Python 3.4 Python 引入了一个非常有用的特性——协程，在本篇文章当中我们将详细介绍一下 Python 协程的原理以及虚拟机具体的实现协程的方式。
super 是 Python 面向对象编程当中非常重要的一部分内容，在本篇文章当中详细介绍了 super 内部的工作原理和 CPython 内部部分源代码分析了 super 的具体实现。
在本篇文章当中主要分析的生成器内部实现原理和相关的两个重要的字节码，分析了生成器能够停下来还能够恢复执行的原因，深入剖析的生成器的原理的各个细节。