如何在Python API服务器中避免重复处理？解决高开销detect_primes函数的并发重复调用问题

你的核心痛点是缓存击穿：当多个并发请求同时请求未缓存的数字时，都会触发高开销的detect_primes调用，而普通缓存无法解决这个问题。我们可以通过请求合并（Request Coalescing）+ 线程安全的任务调度来实现让多个请求共享同一个计算结果，彻底避免重复计算。

以下是基于Flask的具体实现方案，思路同样适用于FastAPI：

核心思路 #

1. 维护一个全局缓存存储已计算的质数结果
2. 维护一个待计算队列和任务注册表，跟踪哪些数字正在等待计算
3. 用一个后台线程批量处理队列中的数字，确保每个数字只被计算一次
4. 并发请求会等待对应数字的计算完成，而不是各自发起计算

完整代码实现 #

from flask import Flask, request
from typing import List, Dict, Tuple
import threading

app = Flask(__name__)

# 全局缓存：存储已计算完成的质数结果
prime_cache: Dict[int, bool] = {}
# 待计算任务注册表：key=数字，value=(等待事件, 占位符)
pending_tasks: Dict[int, Tuple[threading.Event, None]] = {}
# 待计算数字队列（去重前）
calculation_queue: List[int] = []
# 保护队列和缓存的锁
queue_lock = threading.Lock()
cache_lock = threading.Lock()
# 后台线程唤醒事件
queue_event = threading.Event()
# 后台线程运行标志
is_processor_running = True


def detect_primes(nums: List[int]) -> Dict[int, bool]:
    """模拟高开销的质数检测函数，替换为你的实际逻辑"""
    result = {}
    for num in nums:
        if num <= 1:
            result[num] = False
        elif num == 2:
            result[num] = True
        elif num % 2 == 0:
            result[num] = False
        else:
            is_prime = True
            for i in range(3, int(num**0.5) + 1, 2):
                if num % i == 0:
                    is_prime = False
                    break
            result[num] = is_prime
    return result


def batch_calculation_processor():
    """后台线程，批量处理待计算队列中的数字"""
    global is_processor_running
    while is_processor_running:
        # 等待队列中有新数字
        queue_event.wait()
        
        # 取出队列中所有数字并去重
        with queue_lock:
            unique_nums = list(set(calculation_queue))
            calculation_queue.clear()
        queue_event.clear()
        
        if not unique_nums:
            continue
        
        # 执行高开销计算
        calculation_result = detect_primes(unique_nums)
        
        # 更新缓存并通知所有等待的请求
        with cache_lock:
            prime_cache.update(calculation_result)
            for num in unique_nums:
                if num in pending_tasks:
                    event, _ = pending_tasks.pop(num)
                    event.set()  # 通知等待的请求计算完成


# 启动后台批量处理线程
threading.Thread(target=batch_calculation_processor, daemon=True).start()


@app.route('/detect', methods=['GET'])
def search():
    nums_str = request.args.get('nums', '')
    if not nums_str:
        return {}
    
    nums = list(map(int, nums_str.split(',')))
    final_result = {}
    wait_list = []
    
    with cache_lock:
        for num in nums:
            # 优先从缓存取结果
            if num in prime_cache:
                final_result[num] = prime_cache[num]
            # 数字正在计算中，加入等待列表
            elif num in pending_tasks:
                wait_list.append((num, pending_tasks[num][0]))
            # 数字需要计算，加入队列并创建等待事件
            else:
                wait_event = threading.Event()
                pending_tasks[num] = (wait_event, None)
                wait_list.append((num, wait_event))
                with queue_lock:
                    calculation_queue.append(num)
                # 唤醒后台线程开始处理
                queue_event.set()
    
    # 等待所有需要计算的数字完成
    for num, event in wait_list:
        event.wait()
        final_result[num] = prime_cache[num]
    
    # 转换为示例要求的字符串键格式
    return {str(k): v for k, v in final_result.items()}


if __name__ == '__main__':
    app.run(debug=True)

工作流程说明 #

以你提到的两个并发请求为例：

1. 请求1传入13,14,15：这三个数字都不在缓存和待处理任务中，会被加入计算队列，后台线程被唤醒。
2. 请求2传入15,16：15已经在待处理任务中，16被加入队列，后台线程会一次性取出13,14,15,16去重后的列表进行批量计算。
3. 两个请求都会等待各自需要的数字计算完成，最终分别返回对应的结果，整个过程detect_primes只被调用一次。

关键优势 #

• 彻底避免重复计算：无论多少并发请求，同一个数字只会被计算一次
• 批量处理优化：最大化利用detect_primes的批量处理效率（如果你的实际函数支持批量优化）
• 线程安全：所有全局状态操作都通过锁保护，符合Web框架的多线程模型
• 低侵入性：不需要修改detect_primes的核心逻辑，只需要封装调度层

注意事项 #

1. 异常处理：如果detect_primes可能抛出异常，需要在batch_calculation_processor中添加异常捕获，避免后台线程崩溃，同时通知等待的请求（比如设置事件并标记错误）。
2. 缓存内存管理：如果数字范围极大，可以用LRU缓存（比如cachetools.LRUCache）替换普通字典，自动淘汰过期或不常用的结果。
3. 分布式扩展：如果是多进程/多服务器部署，内存中的缓存和队列无法共享，需要改用分布式锁（如Redis Redlock）+ 分布式缓存（如Redis）+ 消息队列（如RabbitMQ）来实现跨节点的请求合并。

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