前言

緩存和數(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");


        UpdateWrapper 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");
    }

由此可見問題還是比較多的，如果這么在項目中使用這種寫法，那最終還是會讀取到臟數(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ù)庫進行頻繁更新，其實基本保證不了瞬間的一致性，只能在最終保證一致性，通過消息異步的方式可以有效的控制緩存更新、刪除的可靠性。

審核編輯：黃飛