如何为Django后端服务器请求排队以适配API速率限制

针对你的Django DRF后端+Vue前端架构，结合RapidAPI 10次/秒的速率限制，以下是可落地的解决方案：

一、后端全局请求队列（核心措施） #

方案1：用Celery实现速率控制的任务队列 #

利用Celery的rate_limit参数直接限制任务执行频率，完美匹配10次/秒的要求：

# tasks.py
from celery import shared_task
import requests

@shared_task(rate_limit='10/s')
def call_rapidapi_task(endpoint, params):
    # 替换为你的RapidAPI调用逻辑
    url = f"https://your-rapidapi-endpoint/{endpoint}"
    headers = {"X-RapidAPI-Key": "your-key"}
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    return response.json()

在DRF视图中，将请求转为异步任务，返回任务ID给前端轮询结果：

# views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from .tasks import call_rapidapi_task

class RapidAPIProxy(APIView):
    def post(self, request):
        task = call_rapidapi_task.delay(
            request.data['endpoint'],
            request.data['params']
        )
        return Response({"task_id": task.id})

# 新增任务结果查询视图
class TaskResult(APIView):
    def get(self, request, task_id):
        task = call_rapidapi_task.AsyncResult(task_id)
        if task.ready():
            return Response({"status": "success", "data": task.result})
        return Response({"status": "pending"})

方案2：基于Redis手动实现请求队列 #

如果不想引入Celery，用Redis维护队列，单独启动一个进程按速率消费：

# queue_manager.py
import redis
import time
import requests
from django.conf import settings

r = redis.Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0)
QUEUE_KEY = "rapidapi_queue"
RATE_LIMIT = 10

def enqueue_request(endpoint, params):
    r.rpush(QUEUE_KEY, f"{endpoint}|{str(params)}")

def process_queue():
    while True:
        # 每秒处理10个请求
        for _ in range(RATE_LIMIT):
            item = r.lpop(QUEUE_KEY)
            if not item:
                break
            endpoint, params = item.decode().split("|", 1)
            # 调用RapidAPI逻辑
            call_rapidapi(endpoint, eval(params))
        time.sleep(1)

def call_rapidapi(endpoint, params):
    url = f"https://your-rapidapi-endpoint/{endpoint}"
    headers = {"X-RapidAPI-Key": "your-key"}
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    # 可将结果存入Redis，供前端查询
    r.set(f"result_{endpoint}_{hash(str(params))}", response.json(), ex=300)

启动消费进程：python manage.py shell < queue_manager.py

二、前端侧优化：减少请求量 #

• 合并多端点请求：如果单个Vue组件需要调用3个RapidAPI端点，后端新增一个聚合接口，前端只发一次请求，后端内部处理三个API调用（均走队列）。
• 防抖节流：对用户频繁点击操作做防抖，避免短时间内重复发起请求：

// Vue组件中实现防抖
const debounce = (fn, delay = 300) => {
  let timer = null;
  return (...args) => {
    clearTimeout(timer);
    timer = setTimeout(() => fn.apply(this, args), delay);
  };
};

// 使用示例
const fetchData = debounce(async (endpoint) => {
  const res = await this.$axios.post("/api/rapidapi-proxy", {
    endpoint,
    params: { /* 参数 */ }
  });
  // 处理数据
});

• 本地缓存：用Vuex或localStorage缓存已获取的数据，相同请求短时间内直接读取缓存，不发起后端请求。

三、后端缓存：降低RapidAPI调用频次 #

用Django缓存框架缓存RapidAPI响应结果，减少重复调用：

from django.core.cache import cache

def call_rapidapi(endpoint, params):
    cache_key = f"rapidapi_{endpoint}_{hash(str(params))}"
    cached_data = cache.get(cache_key)
    if cached_data:
        return cached_data
    # 实际调用API
    url = f"https://your-rapidapi-endpoint/{endpoint}"
    headers = {"X-RapidAPI-Key": "your-key"}
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    data = response.json()
    # 缓存5分钟（可根据数据更新频率调整）
    cache.set(cache_key, data, 300)
    return data

四、兜底措施：429错误重试 #

即使做了队列控制，仍可能遇到突发限流，添加重试机制：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 配置重试策略
session = requests.Session()
retry_strategy = Retry(
    total=3,  # 最多重试3次
    backoff_factor=1,  # 重试间隔递增：1s, 2s, 4s
    status_forcelist=[429],  # 只对429错误重试
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)

def call_rapidapi(endpoint, params):
    url = f"https://your-rapidapi-endpoint/{endpoint}"
    headers = {"X-RapidAPI-Key": "your-key"}
    try:
        response = session.get(url, headers=headers, params=params)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        # 记录日志
        import logging
        logger = logging.getLogger(__name__)
        logger.error(f"RapidAPI请求失败: {str(e)}")
        raise

内容的提问来源于stack exchange，提问作者defender777