应对淘宝 API 限流规则的智能调度算法设计与实现

pip install requests pydantic

import time
import queue
import threading
import requests
from typing import List, Dict
from dataclasses import dataclass, field
from enum import IntEnum

# ===================== 1. 定义任务优先级枚举（核心业务优先） =====================
class TaskPriority(IntEnum):
    HIGH = 1    # 高优先级：实时价格、订单同步
    NORMAL = 2  # 中优先级：商品详情查询
    LOW = 3     # 低优先级：历史数据同步、批量爬取

# ===================== 2. 任务数据结构 =====================
@dataclass(order=True)
class APITask:
    priority: TaskPriority  # 优先级
    task_id: str = field(compare=False)    # 任务ID
    params: Dict = field(compare=False)    # 请求参数
    retry_times: int = field(default=0, compare=False)  # 重试次数

# ===================== 3. 淘宝API 智能调度器核心实现 =====================
class TaobaoAPIScheduler:
    def __init__(
        self,
        api_keys: List[str],  # 多API密钥列表（提升并发上限）
        qps_limit: int = 2,   # 淘宝API QPS限制（每秒请求数）
        daily_limit: int = 100000,  # 单日调用上限
        max_retry: int = 3    # 最大重试次数
    ):
        # 基础配置
        self.api_keys = api_keys
        self.qps_limit = qps_limit
        self.daily_limit = daily_limit
        self.max_retry = max_retry

        # 调度核心组件
        self.token_queue = queue.Queue(maxsize=qps_limit)  # 令牌桶
        self.task_queue = queue.PriorityQueue()  # 优先级任务队列
        self.current_key_index = 0  # 多密钥轮询索引

        # 计量统计
        self.total_calls = 0  # 总调用次数
        self.call_lock = threading.Lock()

        # 启动令牌生成线程（核心：控制QPS）
        self.token_thread = threading.Thread(target=self._generate_tokens, daemon=True)
        self.token_thread.start()

        # 启动调度执行线程
        self.scheduler_thread = threading.Thread(target=self._schedule_tasks, daemon=True)
        self.scheduler_thread.start()

    # --------------------- 令牌桶生成：严格控制QPS ---------------------
    def _generate_tokens(self):
        """每秒生成固定数量的令牌，匹配淘宝QPS限制"""
        while True:
            for _ in range(self.qps_limit):
                if self.token_queue.full():
                    break
                self.token_queue.put(1)
            time.sleep(1)  # 每秒重置令牌

    # --------------------- 多API密钥轮询负载均衡 ---------------------
    def _get_next_api_key(self) -> str:
        """轮询获取API密钥，分摊并发压力"""
        key = self.api_keys[self.current_key_index % len(self.api_keys)]
        self.current_key_index += 1
        return key

    # --------------------- 调用次数统计（防单日超量） ---------------------
    def _increment_call(self) -> bool:
        """原子性统计调用次数，超出单日限制则拒绝"""
        with self.call_lock:
            if self.total_calls >= self.daily_limit:
                return False
            self.total_calls += 1
            return True

    # --------------------- 淘宝API请求执行 ---------------------
    def _execute_request(self, task: APITask) -> Dict:
        """执行API请求，处理限流重试"""
        api_key = self._get_next_api_key()
        url = "https://api.taobao.com/openapi/get_item"  # 替换为真实API地址
        params = {**task.params, "app_key": api_key}

        try:
            # 校验单日调用量
            if not self._increment_call():
                return {"code": 403, "msg": "单日调用量已耗尽"}

            # 发送请求
            response = requests.get(url, params=params, timeout=5)
            result = response.json()

            # 触发淘宝限流：429状态码
            if response.status_code == 429 or result.get("code") == "rate_limit":
                raise Exception("API限流触发")
            return result

        except Exception as e:
            # 限流/异常，重试
            if task.retry_times < self.max_retry:
                task.retry_times += 1
                self.add_task(task)
                return {"code": 429, "msg": f"限流重试，第{task.retry_times}次"}
            return {"code": 500, "msg": "重试失败"}

    # --------------------- 核心调度逻辑 ---------------------
    def _schedule_tasks(self):
        """持续从优先级队列取任务，获取令牌后执行"""
        while True:
            try:
                # 1. 获取高优先级任务
                task = self.task_queue.get(timeout=1)
                # 2. 获取令牌（无令牌则阻塞，保证不超QPS）
                self.token_queue.get()
                # 3. 执行请求
                self._execute_request(task)
                self.task_queue.task_done()
            except queue.Empty:
                continue

    # --------------------- 外部接口：添加任务到调度器 ---------------------
    def add_task(self, task: APITask):
        """外部调用：添加API任务到调度队列"""
        self.task_queue.put(task)
        print(f"任务[{task.task_id}]已加入调度，优先级：{task.priority.name}")

# ===================== 4. 测试调度器 =====================
if __name__ == '__main__':
    # 初始化调度器（2个API密钥，QPS=2，单日上限1000）
    scheduler = TaobaoAPIScheduler(
        api_keys=["key_1", "key_2"],
        qps_limit=2,
        daily_limit=1000
    )

    # 模拟10个并发任务（高低优先级混合）
    tasks = [
        APITask(priority=TaskPriority.HIGH, task_id="price_001", params={"num_iid": "123456"}),
        APITask(priority=TaskPriority.LOW, task_id="item_002", params={"num_iid": "789012"}),
        APITask(priority=TaskPriority.HIGH, task_id="price_003", params={"num_iid": "345678"}),
    ]

    # 批量提交任务
    for t in tasks:
        scheduler.add_task(t)

    # 保持程序运行
    while True:
        time.sleep(1)
        print(f"已调用API次数：{scheduler.total_calls}")