缓存 Redis 与 Celery 异步

第 10 章 · 缓存 Redis 与 Celery 异步

本章目标：使用 Flask-Caching 将 Redis 作为缓存后端；掌握 @cache.memoize 与视图缓存；入门 Celery 分布式任务队列（Redis broker）；实现异步发送通知（邮件/站内消息模拟）；在 api-demo 完成本地 Docker 概念验证，理解 Web 进程与 Worker 分工。

学时建议：5 小时（含 2 小时 Docker + Celery 跟练）

前置：本模块 ch05 ORM 查询；ch07 API；ch09 日志（Worker 日志同样重要）。Redis 概念见 ops-deploy 相关章节。

10.1 为什么需要缓存与异步

问题	同步方案痛点	本章方案
商品列表重复查库	每次请求打满数据库	Redis 缓存查询结果
用户注册后发欢迎通知	用户等 SMTP 3～5 秒	Celery 后台投递
日活统计汇总	管理接口超时	定时任务异步聚合

                    ┌─────────────┐
  HTTP 请求 ───────►│ Flask       │
                    │  Gunicorn   │
                    └──────┬──────┘
                           │ cache.get/set
                           ▼
                    ┌─────────────┐
                    │ Redis       │◄──── Celery Worker
                    │ db 0: cache │      读 broker、写结果
                    │ db 1: broker│
                    └──────┬──────┘
                           ▼
                    ┌─────────────┐
                    │ PostgreSQL  │
                    └─────────────┘

api-demo 为教学项目；Redis 地址示例 redis://127.0.0.1:6379，通知使用控制台 backend，不连接真实业务中间件。

10.2 本地 Redis 与 docker-compose

api-demo/docker-compose.dev.yml：

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes

volumes:
  redis_data:

docker compose -f docker-compose.dev.yml up -d redis
redis-cli ping   # PONG

Redis DB	用途
`/0`	Flask-Caching
`/1`	Celery broker
`/2`	Celery result backend（可选）

10.3 Flask-Caching 配置

pip install Flask-Caching redis

api_demo/extensions.py：

from flask_caching import Cache

cache = Cache()

config.py 补充：

class Config:
    CACHE_TYPE = "RedisCache"
    CACHE_REDIS_URL = os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/0")
    CACHE_DEFAULT_TIMEOUT = 300
    CACHE_KEY_PREFIX = "api_demo:"

create_app：

from api_demo.extensions import cache

def create_app(...):
    # ...
    cache.init_app(app)
    return app

验证：

# flask shell
>>> from api_demo.extensions import cache
>>> cache.set("ping", "pong", timeout=60)
>>> cache.get("ping")
'pong'

10.4 视图缓存与 cache.memoize

10.4.1 缓存整个视图响应

from api_demo.extensions import cache
from api_demo.api import api_bp


@api_bp.get("/products/hot")
@cache.cached(timeout=120, query_string=True)
def hot_products():
    products = Product.query.filter_by(is_published=True).order_by(
        Product.stock.desc()
    ).limit(10).all()
    return ok(data=[product_to_dict(p) for p in products])

query_string=True 使不同查询参数拥有独立缓存键。

10.4.2 @cache.memoize 缓存函数结果

适合纯函数或昂贵计算（不直接缓存 Response 对象）：

@cache.memoize(timeout=300)
def get_category_stats(category_id):
    from sqlalchemy import func
    row = db.session.query(
        func.count(Product.id),
        func.coalesce(func.sum(Product.stock), 0),
    ).filter(Product.category_id == category_id).one()
    return {"count": row[0], "total_stock": int(row[1])}

模型更新后失效：

def invalidate_product_caches(product):
    cache.delete_memoized(get_category_stats, product.category_id)
    cache.delete("view//api/v1/products/hot")  # 或 cache.clear() 开发期

方法	说明
`cache.cached`	装饰视图，缓存 HTTP 响应
`cache.memoize`	装饰函数，按参数建键
`cache.delete_memoized(fn, *args)`	精确失效
`cache.clear()`	清空当前前缀（慎用）

10.5 低层 cache API 与缓存穿透

手动 get / set：

def get_product_detail(product_id):
    key = f"product:{product_id}"
    data = cache.get(key)
    if data is not None:
        return data
    product = db.session.get(Product, product_id)
    if product is None:
        return None
    data = product_to_dict(product)
    cache.set(key, data, timeout=600)
    return data

空值缓存（防止缓存穿透）：对不存在的 id 缓存短 TTL 的占位符 {"_miss": True}。

10.6 Celery 应用配置

pip install celery[redis]

api_demo/celery_app.py：