让我们探索 Python 中协程和结构化并发的令人兴奋的世界。这些强大的功能彻底改变了我们编写并发代码的方式，使其更加高效且更易于管理。
协程是特殊函数，可以暂停执行并将控制权交给其他协程。它们是使用 async 关键字定义的，并且可以使用await 关键字等待。这是一个简单的例子：
async def greet(name):
print(f"Hello, {name}!")
await asyncio.sleep(1)
print(f"Goodbye, {name}!")
async def main():
await greet("Alice")
await greet("Bob")
asyncio.run(main())
在此代码中，greet 函数是一个协程，它打印问候语，等待一秒钟，然后说再见。 main函数调用了两次greet，我们使用asyncio.run来执行主协程。
但是是什么让协程如此特别呢？它们允许我们编写看起来和行为都像同步代码的并发代码，但实际上可以同时执行多个操作。这对于 I/O 密集型任务特别有用，例如网络操作或文件处理。
让我们更深入地了解 asyncio 库，它为 Python 中的异步编程提供了基础。其核心是事件循环，它管理协程的执行。您可以将其视为决定接下来运行哪个协程的调度程序。
以下是我们如何通过 asyncio 创建和使用任务：
import asyncio
async def fetch_data(url):
print(f"Fetching data from {url}")
await asyncio.sleep(2) # Simulating network delay
return f"Data from {url}"
async def main():
urls = ['http://example.com', 'http://example.org', 'http://example.net']
tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
asyncio.run(main())
在此示例中，我们模拟同时从多个 URL 获取数据。 asyncio.create_task 函数将我们的协程转换为任务，然后使用 asyncio.gather 并发执行。
现在，我们来谈谈结构化并发。这是一个旨在使并发代码更可预测、更容易推理的范例。 Python 3.11 引入了一些新功能来支持结构化并发，例如任务组。
以下是我们如何使用任务组：
import asyncio
async def process_item(item):
await asyncio.sleep(1)
return f"Processed {item}"
async def main():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(process_item("A"))
task2 = tg.create_task(process_item("B"))
task3 = tg.create_task(process_item("C"))
print(task1.result())
print(task2.result())
print(task3.result())
asyncio.run(main())
TaskGroup 确保在我们继续之前完成（或取消）所有任务。这有助于防止忘记任务或并发操作之间意外交互等问题。
协程最强大的方面之一是它们有效处理 I/O 操作的能力。让我们看一个简单的异步 Web 服务器的示例：
import asyncio
from aiohttp import web
async def handle(request):
name = request.match_info.get('name', "Anonymous")
text = f"Hello, {name}!"
return web.Response(text=text)
async def main():
app = web.Application()
app.add_routes([web.get('/', handle),
web.get('/{name}', handle)])
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, 'localhost', 8080)
await site.start()
print("Server started at http://localhost:8080")
await asyncio.Event().wait()
asyncio.run(main())
由于协程的强大功能，该服务器可以同时处理多个连接。每个请求都在自己的协程中处理，使服务器即使在高负载下也能保持响应。
让我们探索一些更高级的概念。在处理并发操作时，取消是一个重要的特性。有时我们需要在任务完成之前停止它。我们可以这样做：
async def greet(name):
print(f"Hello, {name}!")
await asyncio.sleep(1)
print(f"Goodbye, {name}!")
async def main():
await greet("Alice")
await greet("Bob")
asyncio.run(main())
在此示例中，我们创建一个长时间运行的任务并在 5 秒后取消它。该任务捕获 CancelledError 并在退出之前执行任何必要的清理。
另一个强大的功能是创建自定义事件循环的能力。虽然默认的事件循环足以满足大多数情况，但有时我们需要更多的控制。这是自定义事件循环的简单示例：
import asyncio
async def fetch_data(url):
print(f"Fetching data from {url}")
await asyncio.sleep(2) # Simulating network delay
return f"Data from {url}"
async def main():
urls = ['http://example.com', 'http://example.org', 'http://example.net']
tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
asyncio.run(main())
这是一个非常基本的自定义事件循环，但它演示了原理。您可以扩展它以添加功能，例如更好的调度、监控或与其他系统的集成。
让我们讨论一下使用协程和结构化并发时的一些最佳实践。首先，始终使用 async with 来管理异步上下文管理器。即使发生异常，这也能确保正确的设置和拆卸：
import asyncio
async def process_item(item):
await asyncio.sleep(1)
return f"Processed {item}"
async def main():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(process_item("A"))
task2 = tg.create_task(process_item("B"))
task3 = tg.create_task(process_item("C"))
print(task1.result())
print(task2.result())
print(task3.result())
asyncio.run(main())
第二，要小心阻塞操作。如果您需要执行 CPU 密集型任务，请考虑使用 asyncio.to_thread 在单独的线程中运行它：
import asyncio
from aiohttp import web
async def handle(request):
name = request.match_info.get('name', "Anonymous")
text = f"Hello, {name}!"
return web.Response(text=text)
async def main():
app = web.Application()
app.add_routes([web.get('/', handle),
web.get('/{name}', handle)])
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, 'localhost', 8080)
await site.start()
print("Server started at http://localhost:8080")
await asyncio.Event().wait()
asyncio.run(main())
第三，当您需要对一组任务进行更多控制时，请使用 asyncio.wait。它允许您等待第一个任务完成，或设置超时：
import asyncio
async def long_running_task():
try:
while True:
print("Working...")
await asyncio.sleep(1)
except asyncio.CancelledError:
print("Task was cancelled")
async def main():
task = asyncio.create_task(long_running_task())
await asyncio.sleep(5)
task.cancel()
try:
await task
except asyncio.CancelledError:
print("Main: task was cancelled")
asyncio.run(main())
调试并发代码可能具有挑战性。 Python 的 asyncio 附带了一些有用的工具。您可以启用调试模式以获得更详细的日志记录：
import asyncio
class MyEventLoop(asyncio.BaseEventLoop):
def __init__(self):
self._running = False
self._ready = asyncio.Queue()
def run_forever(self):
self._running = True
while self._running:
coro = self._ready.get_nowait()
if coro:
coro.send(None)
def stop(self):
self._running = False
def call_soon(self, callback, *args):
self._ready.put_nowait(callback(*args))
# Usage
loop = MyEventLoop()
asyncio.set_event_loop(loop)
async def my_coroutine():
print("Hello from my coroutine!")
loop.call_soon(my_coroutine)
loop.run_forever()
您还可以使用 aiodebug 库来实现更高级的调试功能。
让我们看一个更复杂的示例：并行数据处理管道。这对于处理大型数据集或处理流数据等任务非常有用：
async with aiohttp.ClientSession() as session:
async with session.get('http://example.com') as response:
html = await response.text()
此管道演示了我们如何使用队列在不同处理阶段之间传递数据，所有阶段都同时运行。
协程和结构化并发为 Python 编程开辟了新的可能性。它们使我们能够编写更易于推理和维护的高效并发代码。无论您是构建 Web 服务器、数据处理管道还是响应式 GUI，这些工具都可以帮助您创建强大的高性能应用程序。
记住，掌握这些概念的关键是练习。从简单的示例开始，逐渐构建更复杂的用例。注意错误处理和取消，因为这些对于构建可靠的异步系统至关重要。不要害怕深入研究 asyncio 源代码 - 这是加深您对这些强大功能在幕后如何工作的理解的好方法。
当您继续探索协程和结构化并发时，您将发现新的模式和技术，可以使您的代码更加高效和富有表现力。这是 Python 开发中一个令人兴奋的领域，并且是一个不断发展的领域。因此，继续学习，继续尝试，享受异步编程世界的旅程！
一定要看看我们的创作：
投资者中心 | 智能生活 | 时代与回声 | 令人费解的谜团 | 印度教 | 精英开发 | JS学校
科技考拉洞察 | 时代与回响世界 | 投资者中央媒体 | 令人费解的谜团 | 科学与时代媒介 | 现代印度教
以上是掌握 Python 协程：提高代码效率和性能的详细内容。更多信息请关注PHP中文网其他相关文章！
