本文目录导读：
在Python中实现数据API的异步加载，主要有以下几种流行的方法,我会从简单到复杂依次介绍。
使用 aiohttp（最常用）
这是Python异步HTTP请求的首选库，配合 asyncio 使用。
基本示例：
import aiohttp
import asyncio
async def fetch_data(session, url):
async with session.get(url) as response:
return await response.json()
async def main():
urls = [
'https://api.example.com/data1',
'https://api.example.com/data2',
'https://api.example.com/data3'
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_data(session, url) for url in urls]
results = await asyncio.gather(*tasks)
return results
# 运行
results = asyncio.run(main())
带错误处理和限速的完整示例：
import aiohttp
import asyncio
from typing import List, Dict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AsyncAPIClient:
def __init__(self, base_url: str, max_concurrent: int = 5):
self.base_url = base_url
self.semaphore = asyncio.Semaphore(max_concurrent)
async def fetch_single(self, session: aiohttp.ClientSession, endpoint: str) -> Dict:
url = f"{self.base_url}/{endpoint}"
async with self.semaphore: # 控制并发数
try:
async with session.get(url, timeout=10) as response:
response.raise_for_status()
return await response.json()
except aiohttp.ClientError as e:
logger.error(f"Error fetching {url}: {e}")
return None
async def fetch_multiple(self, endpoints: List[str]) -> List[Dict]:
async with aiohttp.ClientSession() as session:
tasks = [self.fetch_single(session, ep) for ep in endpoints]
return await asyncio.gather(*tasks)
# 使用示例
async def main():
client = AsyncAPIClient("https://api.example.com", max_concurrent=10)
endpoints = ["users", "posts", "comments"]
results = await client.fetch_multiple(endpoints)
return results
results = asyncio.run(main())
使用 httpx（现代替代方案）
httpx 提供了同步和异步两种接口,API设计更友好。
import httpx
import asyncio
async def fetch_data_httpx():
async with httpx.AsyncClient() as client:
# 单个请求
response = await client.get('https://api.example.com/data')
return response.json()
# 并发请求
urls = ['https://api.example.com/data1', 'https://api.example.com/data2']
async with httpx.AsyncClient() as client:
async with client as client:
responses = await asyncio.gather(
*[client.get(url) for url in urls]
)
return [r.json() for r in responses]
# 更实用的批量请求示例
async def batch_fetch_httpx(endpoints: List[str]):
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
results = {}
async for endpoint in endpoints:
response = await client.get(f"/{endpoint}")
results[endpoint] = response.json()
return results
使用 concurrent.futures（线程池方式）
对于某些库不支持异步的情况,可以使用线程池来实现并发：
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def fetch_url(url):
response = requests.get(url)
return response.json()
def concurrent_fetch(urls):
with ThreadPoolExecutor(max_workers=10) as executor:
future_to_url = {executor.submit(fetch_url, url): url for url in urls}
results = []
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
results.append(data)
except Exception as e:
print(f"Error fetching {url}: {e}")
return results
# 使用
urls = ['http://api.example.com/data1', 'http://api.example.com/data2']
results = concurrent_fetch(urls)
结合数据库的异步加载模式
在实际应用中,常需要异步加载数据到数据库：
import aiohttp
import asyncio
import asyncpg # 异步PostgreSQL驱动
from typing import List, Dict
class AsyncDataPipeline:
def __init__(self, api_base_url: str, db_config: Dict):
self.api_base_url = api_base_url
self.db_config = db_config
async def fetch_data(self, session: aiohttp.ClientSession, endpoint: str) -> List[Dict]:
url = f"{self.api_base_url}/{endpoint}"
async with session.get(url) as resp:
return await resp.json()
async def save_to_db(self, conn, data: List[Dict]):
# 批量插入示例
await conn.executemany(
"INSERT INTO data_table (field1, field2) VALUES ($1, $2)",
[(item['field1'], item['field2']) for item in data]
)
async def process_pipeline(self, endpoints: List[str]):
# 建立数据库连接
conn = await asyncpg.connect(**self.db_config)
try:
async with aiohttp.ClientSession() as session:
for endpoint in endpoints:
# 获取数据
data = await self.fetch_data(session, endpoint)
# 保存到数据库
await self.save_to_db(conn, data)
print(f"Processed {endpoint}: {len(data)} records")
finally:
await conn.close()
# 使用示例
async def main():
db_config = {
'host': 'localhost',
'port': 5432,
'database': 'mydb',
'user': 'user',
'password': 'password'
}
pipeline = AsyncDataPipeline('https://api.example.com', db_config)
await pipeline.process_pipeline(['users', 'orders', 'products'])
使用 asyncio 的事件循环控制
对于更精细的控制：
import asyncio
from datetime import datetime
class AsyncDataLoader:
def __init__(self):
self.rate_limiter = asyncio.Semaphore(10) # 每秒最多10个请求
self.last_request_time = {}
async def rate_limited_fetch(self, url: str):
async with self.rate_limiter:
# 速率限制逻辑
now = datetime.now()
if url in self.last_request_time:
elapsed = (now - self.last_request_time[url]).total_seconds()
if elapsed < 0.1: # 同一URL请求间隔至少100ms
await asyncio.sleep(0.1 - elapsed)
self.last_request_time[url] = now
# 实际请求
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
async def batch_load(self, urls: List[str]) -> List[Dict]:
tasks = [self.rate_limited_fetch(url) for url in urls]
return await asyncio.gather(*tasks, return_exceptions=True)
最佳实践总结
-
选择合适的库：
- 新建项目推荐
httpx（API更友好） - 成熟项目常用
aiohttp - 简单场景可用
concurrent.futures
- 新建项目推荐
-
并发控制：
- 使用
asyncio.Semaphore控制并发数 - 避免同时发送过多请求导致服务器拒绝
- 使用
-
错误处理：
- 超时设置
- 重试机制
- 异常捕获
-
性能优化：
- 连接池复用
- 合理设置
max_connections - 使用
gather批量处理
-
资源管理：
- 正确关闭session
- 使用上下文管理器
- 注意内存使用
选择哪种方式取决于你的具体需求：是否需要与现有代码集成、对性能的要求、以及团队对异步编程的熟悉程度。
