在线沟通

Write-Behind：Redis + MySQL 异步写入模式实战

甘蔗 330 2026-07-18 20:17:42 0 收藏 保存至本地

Write-Behind：Redis + MySQL 异步写入模式实战

问题背景

我在开发博客平台时，遇到了一个常见的频繁交互性能问题：收藏功能。

用户点击收藏文章的瞬间，如果同步写入 MySQL，整个链路是这样的：

用户点击收藏 → 前端发请求 → Flask 接收 → 写入 MySQL → 返回响应

这条链路在并发量小的个人博客上没有问题，但我希望我的博客从一开始就按生产级标准设计。两个考量：

  1. 收藏操作频率可能很高 — 如果文章火了，可能同时有多人收藏
  2. MySQL 写入是相对昂贵的操作 — 涉及唯一约束检查、索引更新、外键约束等

方案选择

常用的缓存策略有三种：

策略 描述 数据一致性 性能
Cache-Aside 读缓存，未命中读 DB，写同时更新缓存+DB 强 中
Write-Through 写操作同时写入缓存和 DB 强 低
Write-Behind 写操作先写缓存，异步批量同步 DB 最终一致 高

我选择 Write-Behind，因为：

  • 收藏操作不是金融交易，允许短暂延迟
  • 读操作远多于写操作
  • 用户关心的是"我看到我的收藏了"，不关心"DB 里立刻有这条记录"

架构设计

整体流程

用户点击收藏
    ↓
Flask API 层
    ↓
① 立即更新 Redis（同步，毫秒级）
    ├── Redis Set: user:{uid}:favorites → 添加/移除文章 ID
    ├── Redis String: article:{aid}:fav_count → INCR/DECR
    └── Redis Set: dirty_articles:fav_count → 标记"脏数据"
    ↓
② 返回响应给用户（<50ms）
    ↓
    ... 几秒到几分钟后 ...
    ↓
③ Celery 定时任务触发（每 10 分钟）
    ├── 读取 dirty_articles:fav_count 集合
    ├── 批量 MGET 所有文章的 fav_count
    └── 批量 UPDATE MySQL
    ↓
④ 数据一致

核心代码实现

1. 收藏操作（API 层）

# 用户点击收藏时，先写 Redis，再异步写 MySQL
@interact.route('/article/<int:article_id>/favorite', methods=['POST'])
def toggle_favorite(article_id):
    user_id = session['user_id']
    fav_key = f"user:{user_id}:favorites"
    count_key = f"article:{article_id}:fav_count"
    dirty_key = "dirty_articles:fav_count"

    # 判断当前收藏状态
    is_fav = redis_client.sismember(fav_key, article_id)

    if is_fav:
        # 取消收藏
        redis_client.srem(fav_key, article_id)
        redis_client.decr(count_key)
        action = 'remove'
    else:
        # 添加收藏
        redis_client.sadd(fav_key, article_id)
        redis_client.incr(count_key)
        action = 'add'

    # 标记此文章为"脏数据"，等待定时任务同步
    redis_client.sadd(dirty_key, article_id)

    # 异步下发 Celery 任务，写入 MySQL
    async_toggle_favorite.delay(
        user_id, article_id, action,
        timestamp=time.time()
    )

    return jsonify({'status': 'ok'})

2. Celery 异步写入任务

@shared_task(bind=True, max_retries=3)
def async_toggle_favorite(self, user_id, article_id, action, timestamp=None):
    """异步写库任务"""
    try:
        if action == 'add':
            fav = UserFavoriteArticle(
                user_id=user_id,
                article_id=article_id,
                created_at=datetime.fromtimestamp(timestamp)
            )
            db.session.add(fav)
            # 捕获唯一约束冲突（说明库里已有，直接忽略）
            try:
                db.session.commit()
            except IntegrityError:
                db.session.rollback()

        elif action == 'remove':
            UserFavoriteArticle.query.filter_by(
                user_id=user_id,
                article_id=article_id
            ).delete()
            db.session.commit()

    except Exception as exc:
        db.session.rollback()
        raise self.retry(exc=exc, countdown=5)  # 5 秒后重试

3. 收藏数批量同步（定时任务）

@shared_task
def sync_article_fav_counts_to_db():
    """
    定时任务：将 Redis 中发生变动的文章收藏数，批量同步到 MySQL
    每 10 分钟执行一次
    """
    dirty_set_key = "dirty_articles:fav_count"

    # 1. 弹出所有脏数据的文章 ID
    dirty_article_ids = set()
    while True:
        ids = redis_client.spop(dirty_set_key, 100)
        if not ids:
            break
        dirty_article_ids.update([int(i) for i in ids])

    if not dirty_article_ids:
        return "No dirty articles to sync."

    # 2. 批量从 Redis 获取最新计数
    count_keys = [f"article:{aid}:fav_count" for aid in dirty_article_ids]
    counts = redis_client.mget(count_keys)

    update_mapping = {}
    for aid, count in zip(dirty_article_ids, counts):
        if count is not None:
            update_mapping[aid] = int(count)

    # 3. 批量更新 MySQL（一条 SQL 更新多行）
    if update_mapping:
        try:
            db.session.bulk_update_mappings(
                Article,
                [{"id": aid, "fav_count": count}
                 for aid, count in update_mapping.items()]
            )
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            # 同步失败，将 ID 放回脏集合，下次再试
            redis_client.sadd(dirty_set_key, *dirty_article_ids)

    return f"Synced {len(update_mapping)} articles."

关键设计决策

1. 为什么用 SPOP 而不是 SMEMBERS？

# ❌ 不好的做法
all_ids = redis_client.smembers(dirty_key)  # 全量读取
redis_client.delete(dirty_key)              # 然后删除

# ✅ 好的做法
while True:
    ids = redis_client.spop(dirty_key, 100)  # 每次弹出 100 个
    if not ids:
        break
    # 处理这批 ID...

SPOP 的优势：

  • 原子操作：弹出和删除一步完成，无竞态条件
  • 分批处理：每次 100 个，内存可控
  • 避免丢失更新：如果在处理过程中有新的收藏，SMEMBERS + DEL 会丢数据

2. 失败回滚机制

sync_article_fav_counts_to_db() 失败
    ↓
except 捕获异常
    ↓
db.session.rollback()  ← 回滚 MySQL 事务
    ↓
redis_client.sadd(dirty_key, *dirty_article_ids)  ← 把 ID 放回脏集合
    ↓
下次定时任务重新处理

Celery Beat 定时配置

CELERY_BEAT_SCHEDULE = {
    'sync_article_fav_counts_to_db': {
        'task': 'blog_sp.celery_task.favorite_celery.sync_article_fav_counts_to_db',
        'schedule': 600  # 每 10 分钟
    },
}

效果

指标 同步写 MySQL Write-Behind
收藏操作响应时间 ~80ms ~30ms
MySQL 写入次数 N 次（每次收藏） 1 次批量 UPDATE
数据一致性 强一致 最多 10 分钟延迟

适用场景

Write-Behind 适合：

  • ✅ 写操作频繁，但一致性要求不极端的场景（点赞、收藏、浏览量）
  • ✅ 有自然聚合窗口的场景（计数、排行）
  • ✅ 用户能容忍短暂延迟的场景

Write-Behind 不适合：

  • ❌ 金融交易、库存扣减
  • ❌ 写入后立刻需要读取最新结果的场景
  • ❌ 无法接受任何数据丢失的场景

本文基于我独立开发的博客项目 甘蔗知行阁 的实际架构编写。