本文目录导读：
在Python中过滤协程重复请求任务,我提供几种常用且有效的方案：
基于集合去重（简单任务）
import asyncio
from typing import Any, Callable
class RequestDeduplicator:
def __init__(self):
self._running_tasks = set()
self._lock = asyncio.Lock()
async def execute_unique(self, task_id: str, coro_factory: Callable[[], Any]) -> Any:
"""执行唯一任务，防止重复"""
async with self._lock:
if task_id in self._running_tasks:
raise ValueError(f"任务 '{task_id}' 已在执行中")
self._running_tasks.add(task_id)
try:
coro = coro_factory()
result = await coro
return result
finally:
async with self._lock:
self._running_tasks.discard(task_id)
# 使用示例
async def fetch_data(url: str) -> str:
await asyncio.sleep(1) # 模拟网络请求
return f"数据来自 {url}"
async def main():
deduplicator = RequestDeduplicator()
url = "https://example.com/api"
# 重复请求会抛出异常
task1 = asyncio.create_task(
deduplicator.execute_unique(url, lambda: fetch_data(url))
)
task2 = asyncio.create_task(
deduplicator.execute_unique(url, lambda: fetch_data(url))
)
try:
results = await asyncio.gather(task1, task2, return_exceptions=True)
print(results)
except ValueError as e:
print(f"重复请求被拦截: {e}")
缓存结果去重（智能缓存）
import asyncio
from functools import wraps
from typing import Dict, Any, Optional
import time
class AsyncCacheDeduplicator:
def __init__(self, ttl: int = 60):
self._cache: Dict[str, tuple] = {} # task_id -> (result, timestamp)
self._locks: Dict[str, asyncio.Lock] = {}
self._ttl = ttl
async def get_or_execute(self, task_id: str, coro_factory) -> Any:
"""获取缓存结果或执行新任务"""
# 检查缓存
if task_id in self._cache:
result, timestamp = self._cache[task_id]
if time.time() - timestamp < self._ttl:
return result
# 确保每个任务唯一锁
if task_id not in self._locks:
self._locks[task_id] = asyncio.Lock()
lock = self._locks[task_id]
async with lock:
# 二次检查（另一个协程可能已经完成）
if task_id in self._cache:
result, timestamp = self._cache[task_id]
if time.time() - timestamp < self._ttl:
return result
# 执行任务
coro = coro_factory()
result = await coro
# 缓存结果
self._cache[task_id] = (result, time.time())
return result
def invalidate(self, task_id: str):
"""清除缓存"""
self._cache.pop(task_id, None)
# 使用示例
cache_dedup = AsyncCacheDeduplicator(ttl=30)
async def expensive_operation(task_id: str) -> str:
await asyncio.sleep(2)
return f"计算结果 {task_id}"
async def main():
# 多个协程请求相同任务，只执行一次
tasks = [
cache_dedup.get_or_execute("calc_1", lambda: expensive_operation("calc_1"))
for _ in range(5)
]
results = await asyncio.gather(*tasks)
print(f"结果: {results}")
使用异步信号量控制并发
import asyncio
from typing import Dict, Set
class RateLimitedDeduplicator:
def __init__(self, max_concurrent: int = 10):
self._semaphore = asyncio.Semaphore(max_concurrent)
self._pending: Dict[str, asyncio.Future] = {}
self._lock = asyncio.Lock()
async def execute(self, task_id: str, coro_factory) -> Any:
"""带并发限制的重复请求过滤"""
async with self._lock:
# 如果任务已在执行，共享同一个Future
if task_id in self._pending:
return await self._pending[task_id]
# 创建新Future
future = asyncio.get_event_loop().create_future()
self._pending[task_id] = future
try:
async with self._semaphore:
coro = coro_factory()
result = await coro
future.set_result(result)
except Exception as e:
future.set_exception(e)
finally:
async with self._lock:
self._pending.pop(task_id, None)
return await future
# 使用示例
limiter = RateLimitedDeduplicator(max_concurrent=5)
async def process_request(task_id: str) -> str:
await asyncio.sleep(1)
return f"任务 {task_id} 完成"
async def main():
# 模拟多个重复请求
requests = [
("task_1", lambda: process_request("task_1")),
("task_1", lambda: process_request("task_1")), # 重复
("task_2", lambda: process_request("task_2")),
("task_2", lambda: process_request("task_2")), # 重复
]
tasks = [
limiter.execute(task_id, coro_factory)
for task_id, coro_factory in requests
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
print(f"请求 {i}: {result}")
通用装饰器方案
import asyncio
from functools import wraps
from typing import Dict, Any, Callable
def dedup_async(ttl: int = 60):
"""异步函数去重装饰器"""
def decorator(func: Callable) -> Callable:
cache: Dict[str, asyncio.Future] = {}
lock = asyncio.Lock()
@wraps(func)
async def wrapper(*args, **kwargs):
# 生成任务标识
task_id = f"{func.__name__}:{args}:{kwargs}"
async with lock:
# 如果有正在执行的任务，复用结果
if task_id in cache and not cache[task_id].done():
return await cache[task_id]
# 创建新Future
future = asyncio.get_event_loop().create_future()
cache[task_id] = future
try:
result = await func(*args, **kwargs)
future.set_result(result)
return result
except Exception as e:
future.set_exception(e)
raise e
finally:
# 延迟清理
await asyncio.sleep(ttl)
async with lock:
cache.pop(task_id, None)
return wrapper
return decorator
# 使用示例
@dedup_async(ttl=10)
async def fetch_user_data(user_id: int) -> dict:
await asyncio.sleep(2)
return {"user_id": user_id, "name": f"用户{user_id}"}
async def main():
# 多个协程同时请求同一个用户数据
tasks = [
fetch_user_data(1)
for _ in range(3)
]
results = await asyncio.gather(*tasks)
print(f"所有结果相同: {all(r == results[0] for r in results)}")
选择建议
- 简单去重：使用集合方案（方案1）
- 需要缓存结果：使用缓存方案（方案2）
- 控制并发：使用信号量方案（方案3）
- 装饰器风格：使用装饰器方案（方案4）
根据你的具体场景选择合适的方案，如果需要更复杂的逻辑（如Redis分布式去重）,可以基于这些基础方案扩展。
