Django中将Redis查询移出视图、实现全视图共享查询结果方案咨询

现有两个方案的缺陷分析 #

• 上下文处理器+sched方案：存在本质缺陷。Django多进程部署时每个worker进程都会独立运行调度器，每秒查询次数等于worker数量，无法完全解决重复查询问题；且sched是阻塞调度，会占用worker进程资源阻塞正常请求，反而加剧性能问题；上下文处理器本身是为模板传递公共变量设计的，用来跑全局定时任务属于误用，后续维护成本极高。
• Celery方案：可以实现异步定时查询，但你的场景需求非常简单，引入Celery需要额外维护worker进程、配置消息代理，反而增加运维复杂度，属于过度设计。

最优方案推荐 #

方案1：轻量缓存方案（满足99%场景，无额外依赖）

通过Django自带的缓存框架控制查询频率，实现每秒最多查询1次Redis，所有视图直接读缓存即可：

1. 配置缓存，在settings.py中添加全局内存缓存：

CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
        'LOCATION': 'redis-df-cache',
    }
}

2. 编写公共工具函数，统一处理Redis查询和缓存逻辑：

import json
import pandas as pd
import redis
from django.core.cache import cache

redisconfig = {} # 替换为你的Redis配置
REDIS_DF_CACHE_KEY = "latest_redis_dataframe"

def get_latest_redis_df():
    # 优先读缓存
    df = cache.get(REDIS_DF_CACHE_KEY)
    if df is not None:
        return df.copy() # 返回副本避免视图修改污染原始数据
    # 缓存过期才查询原始Redis
    r = redis.StrictRedis(**redisconfig)
    keys = sorted(r.keys(pattern="*"))
    vals = r.mget(keys)
    vals = [x for x in vals if x is not None]
    df = pd.DataFrame(map(json.loads, vals))
    # 缓存有效期设为1秒，保证每秒最多查询1次Redis
    cache.set(REDIS_DF_CACHE_KEY, df, timeout=1)
    return df.copy()

3. 视图直接调用即可，无需修改原有业务逻辑：

from your_app.utils import get_latest_redis_df

def your_dashboard_view(request):
    myViewData = get_latest_redis_df()
    # 后续的DataFrame处理、绘图逻辑保持不变
    ...

该方案优势：无额外依赖、逻辑简单易维护，自动兼容多进程部署，并发访问时缓存命中直接读内存，延迟极低。

方案2：定时预更新方案（适合查询耗时超过1秒的场景）

如果Redis查询+生成DataFrame的耗时超过1秒，用户遇到缓存过期时会有等待，可以用APScheduler启动后台定时任务，每秒主动更新缓存，用户永远读取提前生成好的缓存，无任何等待：

1. 安装依赖：pip install apscheduler
2. 在Django的apps.py中添加定时任务启动逻辑：

from django.apps import AppConfig
from apscheduler.schedulers.background import BackgroundScheduler

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app'

    def ready(self):
        from your_app.utils import refresh_redis_df_cache
        scheduler = BackgroundScheduler()
        # 每秒执行一次缓存更新
        scheduler.add_job(refresh_redis_df_cache, 'interval', seconds=1, id='refresh_redis_df', replace_existing=True)
        scheduler.start()

3. 工具函数中新增缓存刷新方法：

def refresh_redis_df_cache():
    r = redis.StrictRedis(**redisconfig)
    keys = sorted(r.keys(pattern="*"))
    vals = r.mget(keys)
    vals = [x for x in vals if x is not None]
    df = pd.DataFrame(map(json.loads, vals))
    # 缓存超时设长一点，避免任务偶发失败时缓存过期
    cache.set(REDIS_DF_CACHE_KEY, df, timeout=5)

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