基于Celery在Redis中缓存Flask应用的耗时计算结果

你不需要依赖Celery内部的Redis连接，直接在任务代码里初始化独立的Redis客户端，就能自由读写指定key的缓存数据，完全避开Celery的“黑盒”逻辑。

步骤1：安装Redis Python客户端 #

先确保Celery容器内安装了Redis客户端：

pip install redis

步骤2：初始化Redis客户端 #

在Celery任务文件中，和Celery实例放在一起初始化，使用和Celery一致的Redis服务地址：

from celery import Celery
import redis
import json

# 初始化Celery实例
app = Celery('tasks', broker='redis://redis:6379/0', backend='redis://redis:6379/0')

# 初始化独立的Redis客户端，和Celery共用同一个Redis服务
redis_client = redis.Redis(host='redis', port=6379, db=0, decode_responses=True)
# decode_responses=True：让Redis返回字符串而非字节流，简化JSON处理

# 定义缓存key常量
EXPENSIVE_RESULT_KEY = "app:homepage:expensive_data"

步骤3：补全任务逻辑 #

结合你的需求实现Redis读写，注意处理JSON序列化（Redis仅支持字符串存储）：

@app.task(bind=True, name='expensiveTask')
def expensiveTask(self, args=None):
    self.update_state(state='PENDING')
    returnVal = tH.getDefaultFailureResponse()

    try:
        # 优先读取Redis缓存
        cached_data = redis_client.get(EXPENSIVE_RESULT_KEY)
        if cached_data:
            # 把Redis中的JSON字符串转回Python对象
            returnVal = json.loads(cached_data)
        else:
            # 缓存不存在时执行耗时任务
            returnVal = tH.expensiveFunction()
            # 将结果序列化后存入Redis
            redis_client.set(EXPENSIVE_RESULT_KEY, json.dumps(returnVal))
        self.update_state(state='COMPLETE')

    except Exception as e:
        self.update_state(state='FAILED')
        # 可选：记录异常信息
        print(f"Task error: {str(e)}")

    return returnVal

# 定时更新缓存的任务
@app.task(bind=True, name='scheduledExpensiveTask')
def scheduledExpensiveTask(self, args=None):
    try:
        # 执行耗时任务获取最新数据
        fresh_data = tH.expensiveFunction()
        # 序列化后覆盖Redis中的旧缓存
        redis_client.set(EXPENSIVE_RESULT_KEY, json.dumps(fresh_data))
        return "Cache updated successfully"
    except Exception as e:
        print(f"Scheduled task error: {str(e)}")
        return f"Update failed: {str(e)}"

关键细节说明 #

• decode_responses=True：省去手动将字节流转为字符串的步骤，直接处理可读内容
• JSON序列化：因为expensiveFunction返回的是JSON对象，必须用json.dumps()转为字符串存入Redis，读取时用json.loads()转回Python对象
• 可选过期时间：如果需要缓存自动失效，可给set()方法添加ex参数，比如redis_client.set(EXPENSIVE_RESULT_KEY, json.dumps(returnVal), ex=600)，表示10分钟后缓存过期（和定时任务周期匹配）

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