实际示例
现实世界中的协程管道应用
网页抓取管道
import asyncio
import aiohttp
from bs4 import BeautifulSoup
async def url_generator():
urls = [
'https://example.com/page1',
'https://example.com/page2',
'https://example.com/page3'
]
for url in urls:
yield url
async def fetch_page(session, url):
async with session.get(url) as response:
return await response.text()
async def parse_content(html):
soup = BeautifulSoup(html, 'html.parser')
## 提取特定数据
return soup.find_all('div', class_='content')
async def data_processor(content):
## 处理和转换提取的数据
processed_data = [item.text for item in content]
return processed_data
async def web_scraping_pipeline():
async with aiohttp.ClientSession() as session:
url_source = url_generator()
async def pipeline_stage():
async for url in url_source:
html = await fetch_page(session, url)
content = await parse_content(html)
processed_data = await data_processor(content)
yield processed_data
async for result in pipeline_stage():
print(f"抓取到的数据: {result}")
async def main():
await web_scraping_pipeline()
asyncio.run(main())
日志处理管道
import asyncio
import re
async def log_file_reader(filename):
async with aiofiles.open(filename, mode='r') as file:
async for line in file:
yield line
async def log_parser(log_line):
## 使用正则表达式解析日志行
pattern = r'(\d{4}-\d{2}-\d{2}) (\w+): (.+)'
match = re.match(pattern, log_line)
if match:
return {
'date': match.group(1),
'level': match.group(2),
'message': match.group(3)
}
return None
async def log_filter(parsed_log):
## 根据特定标准过滤日志
if parsed_log and parsed_log['level'] == 'ERROR':
yield parsed_log
async def log_processing_pipeline(filename):
log_source = log_file_reader(filename)
async def pipeline():
async for line in log_source:
parsed_log = await log_parser(line)
if parsed_log:
async for filtered_log in log_filter(parsed_log):
yield filtered_log
async for result in pipeline():
print(f"过滤后的日志: {result}")
async def main():
await log_processing_pipeline('system.log')
asyncio.run(main())
数据转换管道
import asyncio
import pandas as pd
async def data_source():
## 模拟数据生成
data = [
{'id': 1, 'value': 10},
{'id': 2, 'value': 20},
{'id': 3, 'value': 30}
]
for item in data:
yield item
async def transform_stage(source):
async for item in source:
## 复杂的转换逻辑
transformed = {
'id': item['id'],
'squared_value': item['value'] ** 2,
'is_even': item['value'] % 2 == 0
}
yield transformed
async def aggregation_stage(source):
aggregated_data = []
async for item in source:
aggregated_data.append(item)
## 转换为DataFrame进行高级处理
df = pd.DataFrame(aggregated_data)
return df
async def data_pipeline():
source = data_source()
transformed = transform_stage(source)
final_df = await aggregation_stage(transformed)
print("处理后的DataFrame:")
print(final_df)
async def main():
await data_pipeline()
asyncio.run(main())
管道性能比较
graph LR
A[顺序处理] --> B[性能开销]
C[协程管道] --> D[高效]
E[并行处理] --> F[最佳性能]
用例场景
| 场景 |
协程管道的优势 |
| 网络I/O |
减少等待时间 |
| 数据处理 |
并发转换 |
| 微服务 |
高效通信 |
高级技术
在LabEx，我们建议探索：
错误处理与弹性
async def resilient_pipeline_stage(source):
async for item in source:
try:
processed = await process_with_retry(item)
yield processed
except Exception as e:
logging.error(f"管道阶段错误: {e}")
通过掌握这些实际示例，开发者可以使用Python协程管道构建健壮、高效且可扩展的异步数据处理系统。
