缓存一致性

读缓存

双检加锁策略

采用双检加锁策略

• 多个线程同时去查询数据库的这条数据，那么我们可以在第一个查询数据的请求上使用一个 互斥锁来锁住它。
• 其他的线程走到这一步拿不到锁就等着，等第一个线程查询到了数据，然后做缓存。
• 后面的线程进来发现已经有缓存了，就直接走缓存。

package com.atguigu.redis.service;import com.atguigu.redis.entities.User;
import com.atguigu.redis.mapper.UserMapper;
import io.swagger.models.auth.In;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.PathVariable;import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;/*** @auther zzyy* @create 2021-05-01 14:58*/
@Service
@Slf4j
public class UserService {public static final String CACHE_KEY_USER = "user:";@Resourceprivate UserMapper userMapper;@Resourceprivate RedisTemplate redisTemplate;/*** 业务逻辑没有写错，对于小厂中厂(QPS《=1000)可以使用，但是大厂不行* @param id* @return*/public User findUserById(Integer id){User user = null;String key = CACHE_KEY_USER+id;//1 先从redis里面查询，如果有直接返回结果，如果没有再去查询mysqluser = (User) redisTemplate.opsForValue().get(key);if(user == null){//2 redis里面无，继续查询mysqluser = userMapper.selectByPrimaryKey(id);if(user == null){//3.1 redis+mysql 都无数据//你具体细化，防止多次穿透，我们业务规定，记录下导致穿透的这个key回写redisreturn user;}else{//3.2 mysql有，需要将数据写回redis，保证下一次的缓存命中率redisTemplate.opsForValue().set(key,user);}}return user;}/*** 加强补充，避免突然key失效了，打爆mysql，做一下预防，尽量不出现击穿的情况。* @param id* @return*/public User findUserById2(Integer id){User user = null;String key = CACHE_KEY_USER+id;//1 先从redis里面查询，如果有直接返回结果，如果没有再去查询mysql，// 第1次查询redis，加锁前user = (User) redisTemplate.opsForValue().get(key);if(user == null) {//2 大厂用，对于高QPS的优化，进来就先加锁，保证一个请求操作，让外面的redis等待一下，避免击穿mysqlsynchronized (UserService.class){//第2次查询redis，加锁后user = (User) redisTemplate.opsForValue().get(key);//3 二次查redis还是null，可以去查mysql了(mysql默认有数据)if (user == null) {//4 查询mysql拿数据(mysql默认有数据)user = userMapper.selectByPrimaryKey(id);if (user == null) {return null;}else{//5 mysql里面有数据的，需要回写redis，完成数据一致性的同步工作redisTemplate.opsForValue().setIfAbsent(key,user,7L,TimeUnit.DAYS);}}}}return user;}
}

写缓存

保障最终数据一致性解决方案

• 给缓存设置过期时间
• 定期清理缓存并回写

先更新数据库，再更新缓存

• 案例演示1->更新缓存异常

  • 先更新mysql的某商品的库存，当前商品的库存是100，更新为99个。
  • 先更新mysql修改为99成功，然后更新redis。
  • 此时假设异常出现，更新redis失败了，这导致mysql里面的库存是99而redis里面的还是100 。
  • 上述发生，会让数据库里面和缓存redis里面数据不一致，读到redis脏数据

• 案例演示2->并发导致

  • A、B两个线程发起调用；A写，B写

    1 A update mysql 1003 B update mysql 804 B update redis 802 A update redis 100
    

  • 最终结果：mysql80，redis100->数据不一致

先更新缓存，再更新数据库

不推荐，业务上一般把mysql作为底单数据库，保证最后解释

• 案例演示->并发导致

  • A、B两个线程发起调用；A写，B写

    A update redis  100B update redis  80B update mysql 80A update mysql 100
    

  • 最终结果：mysql100，redis80->数据不一致

先删除缓存，再更新数据库

• 案例演示->并发导致

  • A、B两个线程发起调用；A写，B读

    • 请求A进行写操作，删除redis缓存后，工作正在进行中，更新mysql…A还么有彻底更新完mysql，还没commit
    • 请求B开工查询，查询redis发现缓存不存在(被A从redis中删除了)
    • 请求B继续，去数据库查询得到了mysql中的旧值(A还没有更新完)
    • 请求B将旧值写回redis缓存
    • 请求A将新值写入mysql数据库

  • 总结

    时间	线程A	线程B	出现的问题
    t1	请求A进行写操作，删除缓存成功后，工作正在mysql进行中…
    t2		1 缓存中读取不到，立刻读mysql，由于A还没有对mysql更新完，读到的是旧值 2 还把从mysql读取的旧值，写回了redis	1 A还没有更新完mysql，导致B读到了旧值 2 线程B遵守回写机制，把旧值写回redis，导致其它请求读取的还是旧值，A白干了。
    t3	A更新完mysql数据库的值，over		redis是被B写回的旧值，mysql是被A更新的新值。出现了，数据不一致问题。

• 解决策略->延时双删

  加上sleep的这段时间，就是为了让线程B能够先从数据库读取数据，再把缺失的数据写入缓存，然后，线程A再进行删除。所以，线程Asleep的时间，就需要大于线程B读取数据再写入缓存的时间。这样一来，其它线程读取数据时，会发现缓存缺失，所以会从数据库中读取最新值。因为这个方案会在第次删除缓存值后，延迟一段时间再次进行删除，所以我们也把它叫做“延迟双删”

  

  • 删除该休眠多久合适？

    • 方式一：

      在业务程序运行的时候，统计下线程读数据和写缓存的操作时间，自行评估自己的项目的读数据业务逻辑的耗时，

      以此为基础来进行估算。然后写数据的休眠时间则在读数据业务逻辑的耗时基础上加百毫秒即可。

    • 方式二

      新启动一个后台监控程序，比如后面要讲解的WatchDog监控程序，会加时

先更新数据库，再删除缓存(推荐~~)

• 案例演示1->更新缓存异常

  • A、B两个线程发起调用；A写，B读

    t3时间上线程A更新Redis缓存失败，会导致Redis缓存与mysql数据不一致的情况发生

    时间	线程A	线程B	出现的问题
    t1	更新数据库中的值…
    t2		缓存中立刻命中，此时B读取的是缓存旧值。	A还没有来得及删除缓存的值，导致B缓存命中读到旧值。
    t3	更新缓存的数据，over

• 解决策略->消息队列重试写Redis缓存

  • 可以把要删除的缓存值或者是要更新的数据库值暂存到消息队列中（例如使用Kafka/RabbitMQ等）。

  • 当程序没有能够成功地删除缓存值或者是更新数据库值时，可以从消息队列中重新读取这些值，然后再次进行删除或更新。

  • 如果能够成功地删除或更新，我们就要把这些值从消息队列中去除，以免重复操作，此时，我们也可以保证数据库和缓存的数据一致了，否则还需要再次进行重试

  • 如果重试超过的一定次数后还是没有成功，我们就需要向业务层发送报错信息了，通知运维人员。

    

如何选方案

优先使用先更新数据库，再删除缓存的方案(先更库→后删存)

• 先删除缓存值再更新数据库，有可能导致请求因缓存缺失而访问数据库，给数据库带来压力导致打满mysql。
• 如果业务应用中读取数据库和写缓存的时间不好估算，那么，延迟双删中的等待时间就不好设置。

策略	高并发多线程条件下	问题	现象	解决方案
先删除redis缓存，再更新mysql	无	缓存删除成功但数据库更新失败	Java程序从数据库中读到旧值	再次更新数据库，重试
	有	缓存删除成功但数据库更新中…有并发读请求	并发请求从数据库读到旧值并回写到redis，导致后续都是从redis读取到旧值	延迟双删
先更新mysql，再删除redis缓存	无	数据库更新成功，但缓存删除失败	Java程序从redis中读到旧值	再次删除缓存，重试
	有	数据库更新成功但缓存删除中…有并发读请求	并发请求从缓存读到旧值	等待redis删除完成，这段时间有数据不一致，短暂存在。

Redis与MySQL数据双写一致性工程落地

阿里巴巴开源的中间件-canal

定义

定义：历史背景是早期阿里巴巴因为杭州和美国双机房部署，存在跨机房数据同步的业务需求，实现方式主要是基于业务 trigger（触发器） 获取增量变更。从2010年开始，阿里巴巴逐步尝试采用解析数据库日志获取增量变更进行同步，由此衍生出了canal项目

官网：https://github.com/alibaba/canal/wiki

作用

• 数据库镜像
• 数据库实时备份
• 索引构建和实时维护(拆分异构索引、倒排索引等)
• 业务 cache 刷新
• 带业务逻辑的增量数据处理

下载

地址：https://github.com/alibaba/canal/releases

工作原理

• MySQL的主从复制

  • 当 master 主服务器上的数据发生改变时，则将其改变写入二进制事件日志文件binlog中；

  • salve 从服务器会在一定时间间隔内对 master 主服务器上的二进制日志进行探测，探测其是否发生过改变，如果探测到 master 主服务器的二进制事件日志发生了改变，则开始一个 I/O Thread 请求 master 二进制事件日志；

  • 同时 master 主服务器为每个 I/O Thread 启动一个dump Thread，用于向其发送二进制事件日志

  • slave 从服务器将接收到的二进制事件日志保存至自己本地的中继日志文件中；

  • salve 从服务器将启动 SQL Thread 从中继日志中读取二进制日志，在本地重放，使得其数据和主服务器保持一致；

  • 最后 I/O Thread 和 SQL Thread 将进入睡眠状态，等待下一次被唤醒；

    

• canal工作原理

  • canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave，向 MySQL master 发送dump 协议

  • MySQL master 收到 dump 请求，开始推送 binary log 给 slave ( canal )

  • canal解析 binary log 对象(原始为 byte 流)

    

一致性工程案例

MySQL

• 查看当前主机二进制日志 show master status;

  

• 查看log_bin日志状态 show variables like 'log_bin';

  

• 开启MySQL的binlog写入功能(Windows下的my.ini配置文件，Linux下的my.cnf配置文件)

  

  log-bin=mysql-bin #开启 binlog
  binlog-format=ROW #选择 ROW 模式
  server_id=1    #配置MySQL replaction需要定义，不要和canal的 slaveId重复ROW模式 除了记录sql语句之外，还会记录每个字段的变化情况，能够清楚的记录每行数据的变化历史，但会占用较多的空间。
  STATEMENT模式只记录了sql语句，但是没有记录上下文信息，在进行数据恢复的时候可能会导致数据的丢失情况；
  MIX模式比较灵活的记录，理论上说当遇到了表结构变更的时候，就会记录为statement模式。当遇到了数据更新或者删除情况下就会变为row模式；
  

• 重启MySQL

• 验证开启binlog是否成功 show variables like 'log_bin';

• 授权canal连接mysql的账号

  • 查看当前目录下的账号 SELECT * FROM mysql.user;

    

  • 创建账号并授权(此次mysql版本为5.7)

    [删除canal用户，可忽略]
    DROP USER IF EXISTS 'canal'@'%';CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';  
    GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';  
    FLUSH PRIVILEGES;[查看是否添加成功]
    SELECT * FROM mysql.user;
    

canal服务端

• 下载

  canal.deployer-1.1.6.tar.gz：https://github.com/alibaba/canal/releases/tag/canal-1.1.6

  

• 解压

  

• 配置

  修改/mycanal/conf/example路径下instance.properties文件

  • 换成自己的mysql主机master的IP地址

    

  • 换成自己的在mysql新建的canal账户

    

• 启动

  /opt/mycanal/bin路径下执行

  • ./startup.sh

• 查看验证

  • 查看server日志

    

  • 查看样例example的日志

    

canal客户端

https://github.com/bithaolee/canal-python