如何避免Python API服务器中的重复处理？ [复制]

假设一个函数detect_primes调用起来很昂贵，我想避免使用重复的参数重复调用它。我应该怎么办？

使用缓存没有帮助，因为该函数可能会在不同的请求中同时调用。当两个请求都将缓存视为空值时，两个请求都将继续执行昂贵的函数。

def detect_primes(nums: List[int]) -> Dict[int, bool]:
    """ detect whether a list of numbers are prime """

@app.route('/detect', methods=['GET'])
def search():
    args = request.args
    nums = list(map(int, args.get('nums', '').split(',')))
    return detect_primes(nums)

例如，如果一个用户请求 13,14,15，则另一个用户请求 15,16。 答案是{"13": true, "14": false, "15": false} and {"15": false, "16": false}

我想避免打电话detect_primes with [13, 14, 15] and [15, 16]。理想情况下，两个请求都应该等待调用[13, 14, 15, 16]（或两次调用[13, 14, 15] and [16]），并返回相应的结果。

Web框架的选择对我来说并不重要，你可以假设它是flask或fastapi。

编辑：不确定问题如何重复或如何回答Flask 中的全局变量是线程安全的吗？如何在请求之间共享数据？ https://stackoverflow.com/questions/32815451/are-global-variables-thread-safe-in-flask-how-do-i-share-data-between-requests如上所述，无法使用缓存（无论是内存中的 python 缓存还是外部缓存或数据库）。我很高兴被答案证明是错误的。

As per FastAPI 的文档 https://fastapi.tiangolo.com/async/:

当您使用正常声明路径操作函数时def代替async def，它在外部线程池中运行，然后await编辑， 而不是直接调用（因为它会阻塞服务器）。

因此，当您使用def代替async def服务器同时处理请求。

就你的情况而言——既然你将其描述为“理想情况下，两个请求都应该等待......”——你可以定义search端点与async def https://fastapi.tiangolo.com/async/. Async路由在主线程（事件循环）上运行，服务器按顺序处理请求 -只要没有 await调用某个协程（即async函数）在此类路线内，在这种情况下await将把函数控制传递回事件循环，允许循环运行其他任务/请求（请查看这个答案 https://stackoverflow.com/a/71517830/17865804了解更多详细信息和解决方案）。通过这种方式，您可以使用字典来缓存先前（已经计算出的）数字，并使用它在后续请求中快速查找特定数字。您还可以使用类似的方法来限制字典的大小this https://stackoverflow.com/questions/2437617/how-to-limit-the-size-of-a-dictionary。下面给出示例。您可以通过 OpenAPI 测试以下内容：http://127.0.0.1:8000/docs http://127.0.0.1:8000/docs，或者使用诸如这样的 URL.

from fastapi import FastAPI, Query
from typing import List, Dict

app = FastAPI()
d = {}

def is_prime(n) -> bool:
    # check whether 'n' is prime or not

def detect_primes(nums: List[int]) -> Dict[int, bool]:
    res = {}
    for n in nums:
        if n in d:
            res[n] = d.get(n)
            print(f'{n} found in dict')
        else:
            is_n_Prime = is_prime(n)
            res[n] = is_n_Prime
            d[n] = is_n_Prime
    return res

@app.get("/detect")
async def search(nums: List[int] = Query(...)):
    return detect_primes(nums)

但是，如果您需要使用await在你的async def路由（这将导致请求同时处理），您可以使用，例如Semaphore https://docs.python.org/3/library/threading.html#semaphore-objects对象控制对字典的访问，如所述here https://stackoverflow.com/a/42325186/17865804。但是，如果您打算拥有多名工人 https://fastapi.tiangolo.com/deployment/server-workers/同时活跃，每个工人都有自己的东西和变量——工人不共享相同的内存 https://fastapi.tiangolo.com/deployment/concepts/#memory-per-process—您应该考虑使用数据库存储或键值存储（缓存），例如Redis（看看答案here https://stackoverflow.com/a/71537393/17865804 and here https://stackoverflow.com/a/65699375/17865804）。另外，您可能想尝试使用aioredlock https://github.com/joanvila/aioredlock，它允许“在工作人员（进程）之间创建分布式锁”，如上所述here https://stackoverflow.com/a/64911338/17865804.