Python数据API并发控制怎么实现

本文目录导读：

1. 使用信号量（Semaphore）控制并发数
2. 使用线程池控制并发
3. 使用限速器（Rate Limiter）
4. 使用第三方库（如 throttler）
5. 完整的并发控制实现
6. 使用连接池优化
7. 选择建议

在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

选择建议

1. 简单场景：使用asyncio.Semaphore控制并发数
2. 需要限速：结合Semaphore和自定义RateLimiter
3. 需要重试：使用自定义的重试逻辑
4. 高性能要求：使用连接池 + 异步IO
5. 简单快速实现：考虑第三方库如throttler、aiolimiter

根据你的具体需求选择合适的方案,通常建议从简单的Semaphore方案开始，然后根据实际需求逐步增加限速、重试等功能。