本文目录导读：
在Python中实现数据API的并发控制,主要有以下几种常见方案：
使用信号量（Semaphore）控制并发数
import asyncio
import aiohttp
from asyncio import Semaphore
class APIClient:
def __init__(self, max_concurrent=10):
self.semaphore = Semaphore(max_concurrent)
async def fetch_data(self, session, url):
async with self.semaphore: # 控制并发数
async with session.get(url) as response:
return await response.json()
async def batch_fetch(self, urls):
async with aiohttp.ClientSession() as session:
tasks = [self.fetch_data(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# 使用示例
async def main():
client = APIClient(max_concurrent=5)
urls = [f"http://api.example.com/data/{i}" for i in range(20)]
results = await client.batch_fetch(urls)
使用线程池控制并发
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
class ThreadPoolAPIClient:
def __init__(self, max_workers=10):
self.executor = ThreadPoolExecutor(max_workers=max_workers)
def fetch_single(self, url):
response = requests.get(url)
return response.json()
def batch_fetch(self, urls):
futures = [self.executor.submit(self.fetch_single, url) for url in urls]
results = []
for future in as_completed(futures):
try:
result = future.result()
results.append(result)
except Exception as e:
print(f"Error: {e}")
return results
# 使用示例
client = ThreadPoolAPIClient(max_workers=5)
urls = [f"http://api.example.com/data/{i}" for i in range(20)]
results = client.batch_fetch(urls)
使用限速器（Rate Limiter）
import asyncio
import time
from collections import deque
class RateLimiter:
def __init__(self, max_calls, period=1.0):
self.max_calls = max_calls
self.period = period
self.calls = deque()
async def acquire(self):
while True:
now = time.time()
# 移除过期的调用记录
while self.calls and now - self.calls[0] > self.period:
self.calls.popleft()
if len(self.calls) < self.max_calls:
self.calls.append(now)
return
else:
# 等待直到下一个可用时间槽
wait_time = self.period - (now - self.calls[0])
if wait_time > 0:
await asyncio.sleep(wait_time)
class RateLimitedAPIClient:
def __init__(self, max_calls_per_second=10):
self.rate_limiter = RateLimiter(max_calls_per_second)
async def fetch_data(self, session, url):
await self.rate_limiter.acquire() # 限速
async with session.get(url) as response:
return await response.json()
使用第三方库（如 throttler）
from throttler import throttle
import asyncio
import aiohttp
class ThrottledAPIClient:
@throttle(rate_limit=10, period=1) # 每秒最多10次请求
async def fetch_data(self, session, url):
async with session.get(url) as response:
return await response.json()
async def batch_fetch(self, urls):
async with aiohttp.ClientSession() as session:
tasks = [self.fetch_data(session, url) for url in urls]
return await asyncio.gather(*tasks)
完整的并发控制实现
import asyncio
import aiohttp
from asyncio import Semaphore
from typing import List, Any, Optional
import time
class AdvancedAPIClient:
def __init__(
self,
max_concurrent: int = 10,
rate_limit: Optional[int] = None,
retry_count: int = 3
):
self.semaphore = Semaphore(max_concurrent)
self.rate_limit = rate_limit
self.retry_count = retry_count
self._last_request_time = 0
async def _rate_limit_wait(self):
"""限速等待"""
if self.rate_limit:
elapsed = time.time() - self._last_request_time
min_interval = 1.0 / self.rate_limit
if elapsed < min_interval:
await asyncio.sleep(min_interval - elapsed)
self._last_request_time = time.time()
async def fetch_with_retry(self, session, url, **kwargs):
"""带重试的请求"""
for attempt in range(self.retry_count):
try:
async with self.semaphore:
await self._rate_limit_wait()
async with session.get(url, **kwargs) as response:
if response.status == 200:
return await response.json()
elif response.status == 429: # Too Many Requests
retry_after = int(response.headers.get('Retry-After', 1))
await asyncio.sleep(retry_after)
else:
response.raise_for_status()
except Exception as e:
if attempt == self.retry_count - 1:
raise e
await asyncio.sleep(2 ** attempt) # 指数退避
return None
async def batch_fetch(self, urls: List[str], **kwargs) -> List[Any]:
"""批量获取数据"""
async with aiohttp.ClientSession() as session:
tasks = [
self.fetch_with_retry(session, url, **kwargs)
for url in urls
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理异常
processed_results = []
for result in results:
if isinstance(result, Exception):
processed_results.append(None)
print(f"Request failed: {result}")
else:
processed_results.append(result)
return processed_results
# 使用示例
async def main():
client = AdvancedAPIClient(
max_concurrent=5,
rate_limit=10, # 每秒10次
retry_count=3
)
urls = [f"http://api.example.com/data/{i}" for i in range(50)]
results = await client.batch_fetch(urls)
# 处理结果
for url, result in zip(urls, results):
if result:
print(f"Success: {url}")
else:
print(f"Failed: {url}")
# 运行
asyncio.run(main())
使用连接池优化
import aiohttp
from aiohttp import ClientTimeout
class ConnectionPoolClient:
def __init__(self, max_connections=20):
self.connector = aiohttp.TCPConnector(
limit=max_connections, # 连接池大小
limit_per_host=max_connections // 2, # 每个主机的最大连接数
ttl_dns_cache=300, # DNS缓存时间
enable_cleanup_closed=True
)
self.timeout = ClientTimeout(total=30) # 总超时时间
async def create_session(self):
return aiohttp.ClientSession(
connector=self.connector,
timeout=self.timeout
)
async def batch_fetch(self, urls):
results = []
async with await self.create_session() as session:
tasks = [self.fetch_url(session, url) for url in urls]
for task in asyncio.as_completed(tasks):
result = await task
results.append(result)
return results
选择建议
- 简单场景：使用
asyncio.Semaphore控制并发数 - 需要限速：结合
Semaphore和自定义RateLimiter - 需要重试：使用自定义的重试逻辑
- 高性能要求：使用连接池 + 异步IO
- 简单快速实现：考虑第三方库如
throttler、aiolimiter
根据你的具体需求选择合适的方案,通常建议从简单的Semaphore方案开始，然后根据实际需求逐步增加限速、重试等功能。
