关注「Raymond运维」公众号,并设为「星标」,第一时间获取更多运维等文章,不再错过精彩内容。
前言
缓存和数据库如何保证数据的一致是个很经典的问题,关于先更新缓存,还是先更新数据库,或者先删除缓存,还是先删除数据的先后问题,再读写并发的场景下很难做到数据一致,我认为比较好的两种方案:一种是我们经常说的延迟双删机制,但是这个延迟的时间是无法很准确的把握的,还有如果缓存删除失败了应该如何处理,总体来说还是不保险的;另外一种我认为是比较可行的方法,要引入阿里的canal,通过拉取binlog日志解析推送的MQ实现异步更新缓存,达到最终缓存和数据库的一致性;
延迟双删策略

基本流程就是客户端A请求,先去删除缓存,然后将数据写入数据库,此时客户端B查询先去查询缓存,缓存没有返回,去查数据库,此时还没有完成主从同步,拿到是从库的旧数据,然后将旧数据进行缓存,在客户端A完成主从同步后,再次删除缓存,这时数据才是一致的,但是重点就是在休眠的几秒钟,会造成数据的不一致性;
⚠️注意点:第二次删除缓存如果失败,那么缓存里面大概率还是旧数据;所以第二次缓存删除重试的方法比较关键:
一种:失败记录写表,起定时任务去扫描表进行重试,显然这种方式并不会很好,会对数据库造成很大的压力;
另外一种:异步处理,利用消息队列,将消息放在队列中,缓解数据库压力,但是要增加对消息队列的维护;
简单写个延迟双删的demo
@RestController@RequestMappingpublic class RedisController {@Autowiredprivate RedisTemplate redisTemplate;@Autowiredprivate SysUserMapper sysUserMapper;@GetMappingpublic void duobleCancle() throws InterruptedException {redisTemplate.delete("1");SysUser sysUser = sysUserMapper.selectUserById(Long.valueOf(1));SysUser updateSysUser =new SysUser();updateSysUser.setUserName("Lxlxxx");updateSysUser.setEmail("@163.com");UpdateWrapper<SysUser> updateWrapper = new UpdateWrapper();updateWrapper.eq("userId",1);sysUserMapper.update(updateSysUser,updateWrapper);Thread.sleep(3000);redisTemplate.opsForValue().append(sysUser.getUserId(), JSON.toJSONString(sysUser));redisTemplate.delete("1");}
由此可见问题还是比较多的,如果这么在项目中使用这种写法,那最终还是会读取到脏数据;
基于订阅binlog异步更新缓存
大致的流程是这样的:

具体binlog订阅实现
步骤:先安装canal、然后安装rabbitmq、然后就是mysql
Canal配置,因为canal支持 tcp, kafka, rocketMQ, rabbitMQ这四种异步的方式,这里我们使用 rabbitMQ,所以将serverMode配置成rabbitMQ
canal.ip = 1canal.serverMode = rabbitmqcanal.mq.servers = 127.0.0.1canal.mq.vhost=canalcanal.mq.exchange=exchange.tradecanal.mq.username=guestcanal.mq.password=guest---------------------------------------------------------------------------------canal.instance.dbUsername=rootcanal.instance.dbPassword=123456canal.instance.mysql.slaveId=1234canal.instance.master.address=127.0.0.1:3306canal.instance.defaultDatabaseName=testcanal.mq.topic=example
mysql的my.cnf配置
log-bin=mysql-binbinlog-format=ROWserver_id=1
引入依赖,我分别引入的是redis、rabbitmq、mybatis-plus、fastsjon的包
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.3</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>${mybatis.plus.version}</version></dependency>
application.yml配置文件
spring:rabbitmq:virtual-host: canalhost: 127.0.0.1publisher-confirms: truedatasource:url: jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=UTCusername: rootpassword: rootdriver-class-name: com.mysql.jdbc.Driverredis:host: 127.0.0.1
RabbitmqConfig配置
@Configurationpublic class RabbitMqConfig {@Beanpublic Queue TestDirectQueue() {return new Queue("exchange.canal.queue",true);}@BeanDirectExchange TestDirectExchange() {return new DirectExchange("exchange.canal");}@BeanBinding bindingDirect() {return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("example");}}
RabbitMqListener监听消息异步处理 canal拉取的binlog日志
@Component@Slf4jpublic class RabbitMqListener {@Autowiredprivate StringRedisTemplate redisTemplate;@RabbitListener(queues = "exchange.canal.queue")public void process(Message message) {log.info("canal queue消费的消息" + message.getBody());Map map = JSON.parseObject(message.getBody(), Map.class);JSONArray array = null;String sqlType = (String) map.get("type");if (StringUtils.endsWithIgnoreCase("SELECT", sqlType)) {array = JSONArray.parseArray((String) map.get("data"));}if (null == array) {return;}JSONObject jsonObject = array.getJSONObject(0);if (StringUtils.endsWithIgnoreCase("UPDATE", sqlType)|| StringUtils.endsWithIgnoreCase("INSERT", sqlType)) {redisTemplate.boundValueOps(jsonObject.get("code").toString()).set(jsonObject.toString());} else if (StringUtils.endsWithIgnoreCase("DELETE", sqlType)) {redisTemplate.delete(jsonObject.get("code").toString());}if (StringUtils.endsWithIgnoreCase("SELECT", sqlType)) {redisTemplate.boundValueOps(jsonObject.get("code").toString()).set(jsonObject.toString());} else {redisTemplate.delete(jsonObject.get("code").toString());}}}
总结
在高并发的场景下缓存和数据库的一致性的问题,永远是个比较大的问题,在请求量很大的情况下,我们必须使用缓存来减少数据库的压力,但是我们需要对数据库进行频繁更新,其实基本保证不了瞬间的一致性,只能在最终保证一致性,通过消息异步的方式可以有效的控制缓存更新、删除的可靠性。
微
信
群
WeChat group

为方便大家更好的交流运维等相关技术问题,特创建了微信交流群。需要加群的小伙伴们在关注微信公众号后,点击底部菜单关于→联系我,即可获取加群方式。
博客
客
Blog


CSDN博客

掘金博客
长按识别二维码访问博客网站,查看更多优质原创运维等文章。




