如何限制Python Flask函数仅单实例运行 避免缓存过期重复调用

你遇到的是典型的缓存击穿问题，核心解决思路是加互斥锁，保证缓存失效时只有1个请求执行数据拉取和计算逻辑，其余请求等待缓存生成后直接返回。以下是两种适配不同部署场景的实现方案：

方案1：单机部署（单进程多线程模式） #

直接用Python内置线程锁实现，代码改动最小：

from flask import Flask, request, abort, render_template
from flask_caching import Cache
import threading
import time

app = Flask(__name__)
cache = Cache(app, config={'CACHE_TYPE': 'SimpleCache'})
# 为慢接口单独创建锁实例
page_lock = threading.Lock()

@app.route('/veryslowpage', methods=['GET'])
def veryslowpage():
    # 先查缓存，存在直接返回
    cache_key = f"veryslowpage_{request.query_string.decode()}"
    cached_result = cache.get(cache_key)
    if cached_result:
        return cached_result
    
    # 缓存不存在，尝试抢锁
    have_lock = page_lock.acquire(blocking=False)
    if have_lock:
        try:
            # 抢到锁后二次检查缓存，避免前面的请求已经生成了缓存
            cached_result = cache.get(cache_key)
            if cached_result:
                return cached_result
            # 执行业务逻辑
            data = callexternalAPIs()
            result = heavydataprocessing(data)
            rendered = render_template("./index.html", content=result)
            # 写入缓存
            cache.set(cache_key, rendered, timeout=3600)
            return rendered
        finally:
            page_lock.release()
    else:
        # 没抢到锁，轮询等待缓存生成
        while not cached_result:
            time.sleep(2)
            cached_result = cache.get(cache_key)
        return cached_result

方案2：分布式部署（多进程/多机器部署） #

用缓存实现分布式锁，适配所有部署场景，通用性最强：

from flask import Flask, request, abort, render_template
from flask_caching import Cache
import time
import uuid

app = Flask(__name__)
# 建议用Redis等共享缓存
cache = Cache(app, config={'CACHE_TYPE': 'RedisCache', 'CACHE_REDIS_URL': 'redis://localhost:6379/0'})

@app.route('/veryslowpage', methods=['GET'])
def veryslowpage():
    query_str = request.query_string.decode()
    cache_key = f"veryslowpage_{query_str}"
    lock_key = f"veryslowpage_lock_{query_str}"
    # 锁的唯一标识，防止误删其他请求加的锁
    lock_identity = str(uuid.uuid4())

    # 先查缓存
    cached_result = cache.get(cache_key)
    if cached_result:
        return cached_result
    
    # 尝试加分布式锁，锁过期时间设为2分钟（大于接口最大执行时间即可）
    lock_acquired = cache.set(lock_key, lock_identity, timeout=120, nx=True)
    if lock_acquired:
        try:
            # 二次检查缓存
            cached_result = cache.get(cache_key)
            if cached_result:
                return cached_result
            # 执行业务逻辑
            data = callexternalAPIs()
            result = heavydataprocessing(data)
            rendered = render_template("./index.html", content=result)
            cache.set(cache_key, rendered, timeout=3600)
            return rendered
        finally:
            # 仅释放自己加的锁
            current_lock = cache.get(lock_key)
            if current_lock == lock_identity:
                cache.delete(lock_key)
    else:
        # 没抢到锁，轮询等待缓存
        while not cached_result:
            time.sleep(2)
            cached_result = cache.get(cache_key)
        return cached_result

优化建议 #

如果允许用户在缓存更新时拿到旧数据，可以选择逻辑过期方案：把缓存的实际过期时间设为7200秒（2小时），同时在缓存值里额外存储1小时的逻辑过期时间。当请求发现逻辑过期时，后台异步启动更新逻辑，当前请求直接返回旧缓存，用户完全无感知，不需要等待。

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