Python脚本如何过滤协程重复请求任务

wen python案例 2026-06-27 244

本文目录导读：

基于集合去重（简单任务）
缓存结果去重（智能缓存）
使用异步信号量控制并发
通用装饰器方案
选择建议

在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)}")