如何解決數據庫與緩存一致性

緩存一致性 每次逢年過節的時候搶票非常艱難，放票的時候那么多人同時去搶票，如果所有人查詢、購票等都去訪問數據庫，那數據庫的壓力得有多大，這時候很多都會引入緩存， 把車票信息放入緩存，這樣可以減少數據庫壓力。當乘客購買成功之后，數據庫發生了變化，需要及時更新緩存中的數據，以便于其他乘客能從緩存中及時獲取最新車票信息。這就是緩存一致性。

解決數據庫與緩存一致性主要思路：

1、同步雙寫：也就是修改db的時候同時修改一下緩存，這種模式下會出現無法保證數據庫與緩存的原子性。
如果出現多線程同時修改db的情況，網絡延遲導致數據庫修改順序與請求順序錯位
    例如：A 先操作數據庫修改 x=1
         B也修改數據庫  x=2
         但是網絡延遲導致
         B先修改緩存     x=2
         A再修改緩存     x=1
    這樣就導致了數據庫中x=2,而緩存中則是 x=1，導致數據庫與緩存不一致。
  
2、設置有效期：給緩存設置有效期，到期后自動刪除。再次查詢時更新
    優勢：簡單、方便
    缺點：時效性差，緩存過期之前可能不一致
    場景：更新頻率較低，時效性要求低的業務

那我們有沒有什么更加好的解決方案呢？

阿里云的canal就為我們很好的解決了這一問題：

canal: 是Alibaba旗下的一款開源項目,純Java開發.它是基于數據庫增量日志解析,提供 增量數據訂閱&消費 ,目前主要支持mysql。

canal工作原理

mysql的主從復制原理：

MySQL master 將數據變更寫入二進制日志( binary log , 其中記錄叫做二進制日志事件 binary log events ，可以通過 show binlog events 進行查看) 
 MySQL slave 將 master 的 binary log events 拷貝到它的中繼日志( relay log ) 
 MySQL slave 重放 relay log 中事件，將數據變更反映它自己的數據

canal工作原理

canal模擬mysql salve的交互協議,偽裝自己為mysql slave,向mysql master發送dump協議;
 mysql master收到dump請求,開始推送binary log給slave(也就是canal);
 canal解析binary log對象(原始byte流).

canal的安裝配置（以windows為例）

一、登進Mysql后，使用show variables like'log_bin';查詢是否開啟binlog，如果開啟（ON），進行下一步，如果沒開啟(OFF)，在數據庫的my.ini配置文件添加配置

[mysqld]
    # 開啟 binlog
    log-bin=mysql-bin 
    # 選擇 ROW 模式
    binlog-format=ROW
    # 配置 MySQL replaction 需要定義，不要和 canal 的 slaveId 重復
    server_id=1

二、binlog開啟后，創建一個canal用戶并授權，官網配置是@%，表示所有服務器，所以改為localhost就可以，在mysql中，運行如下代碼，設置完成之后重啟：

CREATE USER canal IDENTIFIED BY 'canal'; 
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'localhost' identified by 'canal'; 
   FLUSH PRIVILEGES;

三、安裝canal

1. 下載地址：https://github.com/alibaba/canal/releases/tag/canal-1.1.6-alpha-1

在conf文件夾里找到confcanal.properties

canal.id = 1
 canal.ip =
 canal.port = 11111
 canal.metrics.pull.port = 11112
 canal.zkServers =
 # flush data to zk
 canal.zookeeper.flush.period = 1000
 canal.withoutNetty = false
 # tcp, kafka, RocketMQ
 canal.serverMode = tcp
 # flush meta cursor/parse position to file

• 說明：這個文件是 canal 的基本通用配置，canal 端口號默認就是 11111，修改 canal 的輸出 model，默認 tcp，改為輸出到 kafka
• 重點關注上面的：canal.serverMode = tcp 這個配置，默認情況，如果是使用mysql，可以不做修改，如果需要將數據同步到kafka，或者rocketmq，可以分別修改即可，此處暫不做修改

2. 解壓到適當位置，解壓后在conf文件夾里找到exampleinstance.properties，

canal.instance.mysql.slaveId=20   #只要和mysql的master的不一樣即可
  # enable gtid use true/false
  canal.instance.gtidon=false
  # position info
  canal.instance.master.address=127.0.0.1:3306

• canal.instance.mysql.slaveId=20 #只要和mysql的master的不一樣即可
• canal.instance.master.address=127.0.0.1:3306 ，監聽的mysql的master節點信息
• 配置連接 MySQL 的用戶名和密碼，默認就是我們前面授權的 canal

3. 修改數據庫配置信息，canal.instance.dbUsername、canal.instance.dbPassword為數據庫賬戶密碼，均為canal，剛剛創建賬號密碼，

4. 到bin目錄下啟動 startup.bat，出現如下界面表示啟動成功

四、spring boot中整合canal maven依賴

< dependency >
     < groupId >com.alibaba.otter< /groupId >
     < artifactId >canal.client< /artifactId >
     < version >1.1.4< /version >
 < /dependency >

java 示例：

public class CanalService {
 public static void main(String[] args) throws Exception{

     //1.獲取 canal 連接對象，我在本機上部署的，所以是127.0.0.1
     CanalConnector canalConnector =
             CanalConnectors.newSingleConnector(new
                     InetSocketAddress("127.0.0.1", 11111), "example", "", "");

     System.out.println("canal啟動并開始監聽數據 ...... ");
     while (true){
         canalConnector.connect();
         //訂閱表 test數據庫下的所有表
         canalConnector.subscribe("test.*");
         //獲取數據
         Message message = canalConnector.get(100);
         //解析message
         List< CanalEntry.Entry > entries = message.getEntries();
         if(entries.size() <=0){
             System.out.println("未檢測到數據");
             Thread.sleep(1000);
         }
         for(CanalEntry.Entry entry : entries){
             //1、獲取表名
             String tableName = entry.getHeader().getTableName();
             //2、獲取類型
             CanalEntry.EntryType entryType = entry.getEntryType();
             //3、獲取序列化后的數據
             ByteString storeValue = entry.getStoreValue();

             //判斷是否rowdata類型數據
             if(CanalEntry.EntryType.ROWDATA.equals(entryType)){
                 //對第三步中的數據進行解析
                 CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                 //獲取當前事件的操作類型
                 CanalEntry.EventType eventType = rowChange.getEventType();
                 //獲取數據集
                 List< CanalEntry.RowData > rowDatasList = rowChange.getRowDatasList();
                 //便利數據
                 for(CanalEntry.RowData rowData : rowDatasList){
                     //數據變更之前的內容
                     JSONObject beforeData = new JSONObject();
                     List< CanalEntry.Column > beforeColumnsList = rowData.getAfterColumnsList();
                     for(CanalEntry.Column column : beforeColumnsList){
                         beforeData.put(column.getName(),column.getValue());
                     }
                     //數據變更之后的內容
                     List< CanalEntry.Column > afterColumnsList = rowData.getAfterColumnsList();
                     JSONObject afterData = new JSONObject();
                     for(CanalEntry.Column column : afterColumnsList){
                         afterData.put(column.getName(),column.getValue());
                     }
                     System.out.println("Table ：" + tableName +
                             ",eventType :" + eventType +
                             ",beforeData :" + beforeData +
                             ",afterData : " + afterData);
                     //操作緩存
                 }
             }else {
                 System.out.println("當前操作類型為：" + entryType);
             }
         }
      }
     }
     }

我手動在book表中操作數據，可以看到程序監控輸出結果

五、 最后我們拿到數據之后可以放入消息隊列，這樣可以加入重試機制，還可以防止冪等問題，最后再寫入緩存。

• 消息隊列保證可靠性：寫到隊列中的消息，成功消費之前不會丟失（重啟項目也不擔心）。
• 消息隊列保證消息成功投遞：下游從隊列拉取消息，成功消費后才會刪除消息，否則還會繼續投遞消息給消費者（符合我們重試的場景）。