前言
緩存和數(shù)據(jù)庫如何保證數(shù)據(jù)的一致是個很經(jīng)典的問題,關(guān)于先更新緩存,還是先更新數(shù)據(jù)庫,或者先刪除緩存,還是先刪除數(shù)據(jù)的先后問題,再讀寫并發(fā)的場景下很難做到數(shù)據(jù)一致,我認為比較好的兩種方案:一種是我們經(jīng)常說的延遲雙刪機制,但是這個延遲的時間是無法很準確的把握的,還有如果緩存刪除失敗了應(yīng)該如何處理,總體來說還是不保險的;另外一種我認為是比較可行的方法,要引入阿里的canal,通過拉取binlog日志解析推送的MQ實現(xiàn)異步更新緩存,達到最終緩存和數(shù)據(jù)庫的一致性;
延遲雙刪策略
基本流程就是客戶端A請求,先去刪除緩存,然后將數(shù)據(jù)寫入數(shù)據(jù)庫,此時客戶端B查詢先去查詢緩存,緩存沒有返回,去查數(shù)據(jù)庫,此時還沒有完成主從同步,拿到是從庫的舊數(shù)據(jù),然后將舊數(shù)據(jù)進行緩存,在客戶端A完成主從同步后,再次刪除緩存,這時數(shù)據(jù)才是一致的,但是重點就是在休眠的幾秒鐘,會造成數(shù)據(jù)的不一致性;
注意點:第二次刪除緩存如果失敗,那么緩存里面大概率還是舊數(shù)據(jù);所以第二次緩存刪除重試的方法比較關(guān)鍵:
一種:失敗記錄寫表,起定時任務(wù)去掃描表進行重試,顯然這種方式并不會很好,會對數(shù)據(jù)庫造成很大的壓力;
另外一種:異步處理,利用消息隊列,將消息放在隊列中,緩解數(shù)據(jù)庫壓力,但是要增加對消息隊列的維護;
簡單寫個延遲雙刪的demo
@RestController @RequestMapping public class RedisController { @Autowired private RedisTemplate redisTemplate; @Autowired private SysUserMapper sysUserMapper; @GetMapping public 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"); UpdateWrapperupdateWrapper = new UpdateWrapper(); updateWrapper.eq("userId",1); sysUserMapper.update(updateSysUser,updateWrapper); Thread.sleep(3000); redisTemplate.opsForValue().append(sysUser.getUserId(), JSON.toJSONString(sysUser)); redisTemplate.delete("1"); }
由此可見問題還是比較多的,如果這么在項目中使用這種寫法,那最終還是會讀取到臟數(shù)據(jù);
基于訂閱binlog異步更新緩存
大致的流程是這樣的:
具體binlog訂閱實現(xiàn)
步驟:先安裝canal、然后安裝rabbitmq、然后就是mysql
Canal配置,因為canal支持 tcp, kafka, rocketMQ, rabbitMQ這四種異步的方式,這里我們使用 rabbitMQ,所以將serverMode配置成rabbitMQ
canal.ip = 1 canal.serverMode = rabbitmq canal.mq.servers = 127.0.0.1 canal.mq.vhost=canal canal.mq.exchange=exchange.trade canal.mq.username=guest canal.mq.password=guest --------------------------------------------------------------------------------- canal.instance.dbUsername=root canal.instance.dbPassword=123456 canal.instance.mysql.slaveId=1234 canal.instance.master.address=127.0.0.1:3306 canal.instance.defaultDatabaseName=test canal.mq.topic=example
mysql的my.cnf配置
log-bin=mysql-bin binlog-format=ROW server_id=1
引入依賴,我分別引入的是redis、rabbitmq、mybatis-plus、fastsjon的包
org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-data-redis org.springframework.boot spring-boot-starter-amqp com.alibaba fastjson 1.2.3 com.baomidou mybatis-plus-boot-starter ${mybatis.plus.version}
application.yml配置文件
spring: rabbitmq: virtual-host: canal host: 127.0.0.1 publisher-confirms: true datasource: url: jdbc//127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=UTC username: root password: root driver-class-name: com.mysql.jdbc.Driver redis: host: 127.0.0.1
RabbitmqConfig配置
@Configuration public class RabbitMqConfig { @Bean public Queue TestDirectQueue() { return new Queue("exchange.canal.queue",true); } @Bean DirectExchange TestDirectExchange() { return new DirectExchange("exchange.canal"); } @Bean Binding bindingDirect() { return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("example"); } }
RabbitMqListener監(jiān)聽消息異步處理 canal拉取的binlog日志
@Component @Slf4j public class RabbitMqListener { @Autowired private 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()); } } }
總結(jié)
在高并發(fā)的場景下緩存和數(shù)據(jù)庫的一致性的問題,永遠是個比較大的問題,在請求量很大的情況下,我們必須使用緩存來減少數(shù)據(jù)庫的壓力,但是我們需要對數(shù)據(jù)庫進行頻繁更新,其實基本保證不了瞬間的一致性,只能在最終保證一致性,通過消息異步的方式可以有效的控制緩存更新、刪除的可靠性。
審核編輯:黃飛
-
數(shù)據(jù)庫
+關(guān)注
關(guān)注
7文章
3900瀏覽量
65774 -
MySQL
+關(guān)注
關(guān)注
1文章
849瀏覽量
27525 -
Redis
+關(guān)注
關(guān)注
0文章
385瀏覽量
11320
原文標題:Redis緩存與Mysql如何保證雙寫一致
文章出處:【微信號:magedu-Linux,微信公眾號:馬哥Linux運維】歡迎添加關(guān)注!文章轉(zhuǎn)載請注明出處。
發(fā)布評論請先 登錄
MySQL與Redis延遲雙刪策略

如何解決數(shù)據(jù)庫與緩存一致性

Redis緩存和MySQL數(shù)據(jù)不一致原因和解決方案
i.MX8M可以調(diào)用哪些刷新/無效緩存函數(shù)來保證緩存一致性?
加速器一致性接口
Cache一致性協(xié)議優(yōu)化研究

自主駕駛系統(tǒng)將使用緩存一致性互連IP和非一致性互連IP
管理基于Cortex?-M7的MCU的高速緩存一致性

Redis緩存更新一致性的方式
如何保障MySQL和Redis的數(shù)據(jù)一致性
redis與mysql如何保持數(shù)據(jù)一致性
redis集群中的hash一致性算法的理解
異構(gòu)計算下緩存一致性的重要性

評論