当前位置：首页 > news >正文

Python异步编程模式：从同步到异步的演进

news 2026/6/17 18:23:55

Python异步编程模式：从同步到异步的演进

引言

在Python开发中，异步编程模式是构建高性能应用的关键。作为一名从Rust转向Python的后端开发者，我深刻体会到异步编程在处理高并发场景时的优势。本文将深入探讨Python中的异步编程模式及其最佳实践。

异步编程核心概念

同步 vs 异步

特性	同步编程	异步编程
执行方式	阻塞等待	非阻塞回调
资源利用	线程阻塞	事件循环复用
并发模型	多线程/多进程	协程
适用场景	CPU密集型	IO密集型
编程复杂度	简单	较高

异步编程模式分类

回调模式：传统异步编程方式
协程模式：使用async/await
生产者-消费者模式：队列驱动
管道模式：数据流处理

环境搭建与基础配置

异步Hello World

import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") asyncio.run(main())

事件循环配置

import asyncio async def main(): loop = asyncio.get_event_loop() print(f"Loop running: {loop.is_running()}") policy = asyncio.get_event_loop_policy() print(f"Policy: {policy}") asyncio.run(main())

异步编程模式实战

模式一：回调模式

import asyncio def callback(future): print(f"Callback: {future.result()}") async def compute(): await asyncio.sleep(1) return 42 async def main(): loop = asyncio.get_event_loop() future = loop.create_task(compute()) future.add_done_callback(callback) await future asyncio.run(main())

模式二：协程模式

import asyncio async def fetch_data(url): print(f"Fetching {url}") await asyncio.sleep(1) return f"Data from {url}" async def main(): results = await asyncio.gather( fetch_data("https://example.com"), fetch_data("https://google.com"), fetch_data("https://github.com") ) for result in results: print(result) asyncio.run(main())

模式三：生产者-消费者模式

import asyncio import random async def producer(queue): for i in range(10): await asyncio.sleep(random.random()) item = f"Item {i}" await queue.put(item) print(f"Produced: {item}") async def consumer(queue, name): while True: item = await queue.get() await asyncio.sleep(random.random()) print(f"Consumer {name} consumed: {item}") queue.task_done() async def main(): queue = asyncio.Queue() producer_task = asyncio.create_task(producer(queue)) consumer_tasks = [ asyncio.create_task(consumer(queue, "A")), asyncio.create_task(consumer(queue, "B")) ] await producer_task await queue.join() for task in consumer_tasks: task.cancel() asyncio.run(main())

模式四：管道模式

import asyncio async def stage1(input_data): await asyncio.sleep(0.5) return input_data * 2 async def stage2(input_data): await asyncio.sleep(0.5) return input_data + 10 async def stage3(input_data): await asyncio.sleep(0.5) return f"Result: {input_data}" async def pipeline(data): result = await stage1(data) result = await stage2(result) result = await stage3(result) return result async def main(): results = await asyncio.gather( pipeline(1), pipeline(2), pipeline(3) ) for result in results: print(result) asyncio.run(main())

实际业务场景

场景一：API网关

import asyncio import aiohttp async def handle_request(request): service_tasks = [ fetch_user_service(request), fetch_product_service(request), fetch_order_service(request) ] results = await asyncio.gather(*service_tasks) return { "user": results[0], "product": results[1], "order": results[2] }

场景二：数据处理流水线

import asyncio async def read_file(file_path): with open(file_path, 'r') as f: for line in f: yield line.strip() await asyncio.sleep(0) async def process_lines(lines): async for line in lines: processed = line.upper() yield processed async def write_file(lines, output_path): with open(output_path, 'w') as f: async for line in lines: f.write(line + '\n') await asyncio.sleep(0) async def main(): lines = read_file('input.txt') processed = process_lines(lines) await write_file(processed, 'output.txt') asyncio.run(main())

性能优化

使用异步上下文管理器

import asyncio class AsyncResource: async def __aenter__(self): await self.connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.disconnect() async def connect(self): await asyncio.sleep(0.1) print("Connected") async def disconnect(self): await asyncio.sleep(0.1) print("Disconnected") async def main(): async with AsyncResource() as resource: print("Using resource") asyncio.run(main())

使用异步迭代器

import asyncio class AsyncDataStream: def __init__(self, count): self.count = count self.index = 0 def __aiter__(self): return self async def __anext__(self): if self.index >= self.count: raise StopAsyncIteration await asyncio.sleep(0.1) self.index += 1 return self.index async def main(): async for item in AsyncDataStream(5): print(item) asyncio.run(main())