Redis-MySQL 数据同步方案

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 添加药品销量
ZADD hot:drug 156 drug:1001
ZADD hot:drug 89 drug:1002
ZADD hot:drug 234 drug:1003

# 销售 +1（原子操作）
ZINCRBY hot:drug 1 drug:1001

# 取 Top 10
ZRANGE hot:drug 0 9 WITHSCORES
# 1) "drug:1003"  # 234
# 2) "drug:1001"  # 156
# 3) "drug:1002"  # 89

1
2
3
4
5
6
7
8

读：
  1. 先读 Redis
  2. miss 则读 MySQL
  3. 回写 Redis

写：
  1. 先写 MySQL
  2. 再删 Redis（不是更新！）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// 读
public Product getProduct(Long id) {
    Product p = redis.get("product:" + id);
    if (p == null) {
        p = mysql.selectById(id);
        redis.setex("product:" + id, 3600, p);  // 1小时过期
    }
    return p;
}

// 写
public void updateProduct(Product p) {
    mysql.updateById(p);
    redis.del("product:" + p.getId());  // 删缓存，下次读 miss 再回填
}

1
2
3

T1 写：删缓存 → 写 DB（未提交）
T2 读：miss → 读 DB（旧值）→ 写缓存（脏数据）
T1 提交

1
2
3
4
5
6

public void updateProduct(Product p) {
    redis.del("product:" + p.getId());  // 第一次删
    mysql.updateById(p);
    Thread.sleep(500);  // 延迟 500ms（经验值，看主从延迟）
    redis.del("product:" + p.getId());  // 第二次删
}

1
2
3

写：
  1. 应用写 DB
  2. Cache 自动同步写（或由 Cache 监听 DB 变更）

1
2
3

写：
  1. 只写 Cache（立即返回）
  2. 异步队列批量 flush 到 DB

1

应用写 MySQL → MySQL 写 binlog → Canal 订阅 → 推送到 Redis

1
2
3

MySQL Master ──binlog──> Canal Server ──Kafka──> 消费服务 ──> Redis
                                          ↑
                                   MQ 解耦 / 重试 / 削峰

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

// 1. 商品查询（Cache-Aside）
public Product getProduct(Long id) {
    String key = "product:" + id;
    Map<Object, Object> cached = redis.opsForHash().entries(key);
    if (cached.isEmpty()) {
        Product p = mysql.selectById(id);
        redis.opsForHash().putAll(key, beanToMap(p));
        redis.expire(key, 3600);
        return p;
    }
    return mapToBean(cached);
}

// 2. 库存扣减（延迟双删 + Lua 原子操作）
public boolean deductStock(Long productId, int qty) {
    String key = "stock:" + productId;
    String lua = """
        local current = tonumber(redis.call('GET', key))
        if current == nil or current < qty then
            return 0
        end
        redis.call('DECRBY', key, qty)
        return 1
    """.replace("qty", String.valueOf(qty));
    Long result = redis.execute(luaScript, Arrays.asList(key));
    if (result == 1) {
        mysql.updateStock(productId, -qty);  // 异步写 DB
    }
    return result == 1;
}

// 3. 销量排行（Write-Behind + ZSet）
@Async
public void onOrderCreated(Order order) {
    // 只写 Redis
    redis.opsForZSet().incrementScore("hot:drug", 
        "drug:" + order.getProductId(), 
        order.getQuantity());
    // 异步队列定时 flush 到 MySQL
    kafkaTemplate.send("sales-stat", order);
}