哈嘍，大家好，我是指北君。不知道大家出去面試有沒有被問到過如何保證數(shù)據(jù)庫(kù)與緩存一致性問題呢？大家又是如何回答的呢？
緩存一致性 每次逢年過節(jié)的時(shí)候搶票非常艱難，放票的時(shí)候那么多人同時(shí)去搶票，如果所有人查詢、購(gòu)票等都去訪問數(shù)據(jù)庫(kù)，那數(shù)據(jù)庫(kù)的壓力得有多大，這時(shí)候很多都會(huì)引入緩存， 把車票信息放入緩存，這樣可以減少數(shù)據(jù)庫(kù)壓力。當(dāng)乘客購(gòu)買成功之后，數(shù)據(jù)庫(kù)發(fā)生了變化，需要及時(shí)更新緩存中的數(shù)據(jù)，以便于其他乘客能從緩存中及時(shí)獲取最新車票信息。這就是緩存一致性。
解決數(shù)據(jù)庫(kù)與緩存一致性主要思路：
1、同步雙寫：也就是修改db的時(shí)候同時(shí)修改一下緩存，這種模式下會(huì)出現(xiàn)無法保證數(shù)據(jù)庫(kù)與緩存的原子性。
如果出現(xiàn)多線程同時(shí)修改db的情況，網(wǎng)絡(luò)延遲導(dǎo)致數(shù)據(jù)庫(kù)修改順序與請(qǐng)求順序錯(cuò)位
例如：A 先操作數(shù)據(jù)庫(kù)修改 x=1
B也修改數(shù)據(jù)庫(kù) x=2
但是網(wǎng)絡(luò)延遲導(dǎo)致
B先修改緩存 x=2
A再修改緩存 x=1
這樣就導(dǎo)致了數(shù)據(jù)庫(kù)中x=2,而緩存中則是 x=1，導(dǎo)致數(shù)據(jù)庫(kù)與緩存不一致。
2、設(shè)置有效期：給緩存設(shè)置有效期，到期后自動(dòng)刪除。再次查詢時(shí)更新
優(yōu)勢(shì)：簡(jiǎn)單、方便
缺點(diǎn)：時(shí)效性差，緩存過期之前可能不一致
場(chǎng)景：更新頻率較低，時(shí)效性要求低的業(yè)務(wù)
那我們有沒有什么更加好的解決方案呢？
阿里云的canal就為我們很好的解決了這一問題：
canal: 是Alibaba旗下的一款開源項(xiàng)目,純Java開發(fā).它是基于數(shù)據(jù)庫(kù)增量日志解析,提供增量數(shù)據(jù)訂閱&消費(fèi),目前主要支持mysql。
canal工作原理
mysql的主從復(fù)制原理：
MySQL master 將數(shù)據(jù)變更寫入二進(jìn)制日志( binary log , 其中記錄叫做二進(jìn)制日志事件 binary log events ，可以通過 show binlog events 進(jìn)行查看)
MySQL slave 將 master 的 binary log events 拷貝到它的中繼日志( relay log )
MySQL slave 重放 relay log 中事件，將數(shù)據(jù)變更反映它自己的數(shù)據(jù)
canal工作原理
canal模擬mysql salve的交互協(xié)議,偽裝自己為mysql slave,向mysql master發(fā)送dump協(xié)議;
mysql master收到dump請(qǐng)求,開始推送binary log給slave(也就是canal);
canal解析binary log對(duì)象(原始byte流).
canal的安裝配置（以windows為例）
一、登進(jìn)Mysql后，使用show variables like'log_bin';查詢是否開啟binlog，如果開啟（ON），進(jìn)行下一步，如果沒開啟(OFF)，在數(shù)據(jù)庫(kù)的my.ini配置文件添加配置
[mysqld]
# 開啟 binlog
log-bin=mysql-bin
# 選擇 ROW 模式
binlog-format=ROW
# 配置 MySQL replaction 需要定義，不要和 canal 的 slaveId 重復(fù)
server_id=1
二、binlog開啟后，創(chuàng)建一個(gè)canal用戶并授權(quán)，官網(wǎng)配置是@%，表示所有服務(wù)器，所以改為localhost就可以，在mysql中，運(yùn)行如下代碼，設(shè)置完成之后重啟：
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'localhost' identified by 'canal';
FLUSH PRIVILEGES;
三、安裝canal
下載地址：https://github.com/alibaba/canal/releases/tag/canal-1.1.6-alpha-1
在conf文件夾里找到\conf\canal.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
說明：這個(gè)文件是 canal 的基本通用配置，canal 端口號(hào)默認(rèn)就是 11111，修改 canal 的輸出 model，默認(rèn) tcp，改為輸出到 kafka
重點(diǎn)關(guān)注上面的：canal.serverMode = tcp 這個(gè)配置，默認(rèn)情況，如果是使用mysql，可以不做修改，如果需要將數(shù)據(jù)同步到kafka，或者rocketmq，可以分別修改即可，此處暫不做修改
解壓到適當(dāng)位置，解壓后在conf文件夾里找到\example\instance.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 ，監(jiān)聽的mysql的master節(jié)點(diǎn)信息
配置連接 MySQL 的用戶名和密碼，默認(rèn)就是我們前面授權(quán)的 canal
修改數(shù)據(jù)庫(kù)配置信息，canal.instance.dbUsername、canal.instance.dbPassword為數(shù)據(jù)庫(kù)賬戶密碼，均為canal，剛剛創(chuàng)建賬號(hào)密碼，
到bin目錄下啟動(dòng) startup.bat，出現(xiàn)如下界面表示啟動(dòng)成功
四、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 連接對(duì)象，我在本機(jī)上部署的，所以是127.0.0.1
CanalConnector canalConnector =
CanalConnectors.newSingleConnector(new
InetSocketAddress("127.0.0.1", 11111), "example", "", "");
System.out.println("canal啟動(dòng)并開始監(jiān)聽數(shù)據(jù) ...... ");
while (true){
canalConnector.connect();
//訂閱表 test數(shù)據(jù)庫(kù)下的所有表
canalConnector.subscribe("test.*");
//獲取數(shù)據(jù)
Message message = canalConnector.get(100);
//解析message
List<CanalEntry.Entry> entries = message.getEntries();
if(entries.size() <=0){
System.out.println("未檢測(cè)到數(shù)據(jù)");
Thread.sleep(1000);
}
for(CanalEntry.Entry entry : entries){
//1、獲取表名
String tableName = entry.getHeader().getTableName();
//2、獲取類型
CanalEntry.EntryType entryType = entry.getEntryType();
//3、獲取序列化后的數(shù)據(jù)
ByteString storeValue = entry.getStoreValue();
//判斷是否rowdata類型數(shù)據(jù)
if(CanalEntry.EntryType.ROWDATA.equals(entryType)){
//對(duì)第三步中的數(shù)據(jù)進(jìn)行解析
CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
//獲取當(dāng)前事件的操作類型
CanalEntry.EventType eventType = rowChange.getEventType();
//獲取數(shù)據(jù)集
List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
//便利數(shù)據(jù)
for(CanalEntry.RowData rowData : rowDatasList){
//數(shù)據(jù)變更之前的內(nèi)容
JSONObject beforeData = new JSONObject();
List<CanalEntry.Column> beforeColumnsList = rowData.getAfterColumnsList();
for(CanalEntry.Column column : beforeColumnsList){
beforeData.put(column.getName(),column.getValue());
}
//數(shù)據(jù)變更之后的內(nèi)容
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("當(dāng)前操作類型為：" + entryType);
}
}
}
}
}我手動(dòng)在book表中操作數(shù)據(jù)，可以看到程序監(jiān)控輸出結(jié)果
五、 最后我們拿到數(shù)據(jù)之后可以放入消息隊(duì)列，這樣可以加入重試機(jī)制，還可以防止冪等問題，最后再寫入緩存。
消息隊(duì)列保證可靠性：寫到隊(duì)列中的消息，成功消費(fèi)之前不會(huì)丟失（重啟項(xiàng)目也不擔(dān)心）。
消息隊(duì)列保證消息成功投遞：下游從隊(duì)列拉取消息，成功消費(fèi)后才會(huì)刪除消息，否則還會(huì)繼續(xù)投遞消息給消費(fèi)者（符合我們重試的場(chǎng)景）。
