SpringBoot整合canal實(shí)現(xiàn)數(shù)據(jù)緩存一致性解決方案

更新時(shí)間：2024年03月15日 09:19:18 作者：牽著貓散步的鼠鼠

canal主要用途是基于?MySQL?數(shù)據(jù)庫(kù)增量日志解析,提供增量數(shù)據(jù)訂閱和消費(fèi),canal是借助于MySQL主從復(fù)制原理實(shí)現(xiàn),本文將給大家介紹SpringBoot整合canal實(shí)現(xiàn)數(shù)據(jù)緩存一致性解決方案,需要的朋友可以參考下

1.前言

canal [k?'næl]，譯意為水道/管道/溝渠，主要用途是基于 MySQL 數(shù)據(jù)庫(kù)增量日志解析，提供增量數(shù)據(jù)訂閱和消費(fèi)。其誕生的背景是早期阿里巴巴因?yàn)楹贾莺兔绹?guó)雙機(jī)房部署，存在跨機(jī)房同步的業(yè)務(wù)需求，實(shí)現(xiàn)方式主要是基于業(yè)務(wù) trigger 獲取增量變更。從 2010 年開(kāi)始，業(yè)務(wù)逐步嘗試數(shù)據(jù)庫(kù)日志解析獲取增量變更進(jìn)行同步，由此衍生出了大量的數(shù)據(jù)庫(kù)增量訂閱和消費(fèi)業(yè)務(wù)。所以其核心功能如下：

數(shù)據(jù)實(shí)時(shí)備份
異構(gòu)數(shù)據(jù)源(elasticsearch、Hbase)與數(shù)據(jù)庫(kù)數(shù)據(jù)增量同步
業(yè)務(wù)緩存cache 刷新，保證緩存一致性
帶業(yè)務(wù)邏輯的增量數(shù)據(jù)處理，如監(jiān)聽(tīng)某個(gè)數(shù)據(jù)的變化做一定的邏輯處理

原理實(shí)現(xiàn)圖如下所示

canal是借助于MySQL主從復(fù)制原理實(shí)現(xiàn)，所以我們接下來(lái)先來(lái)了解一下主從復(fù)制原理。

大概流程可以理解為如下：

master將改變記錄到二進(jìn)制日志(binary log)中（這些記錄叫做二進(jìn)制日志事件，binary log events，可以通過(guò)show binlog events進(jìn)行查看）；
slave將master的binary log events拷貝到它的中繼日志(relay log)；
slave重做中繼日志中的事件，將改變反映它自己的數(shù)據(jù)。

canal的工作原理：

原理相對(duì)比較簡(jiǎn)單：

canal模擬mysql slave的交互協(xié)議，偽裝自己為mysql slave，向mysql master發(fā)送dump協(xié)議
mysql master收到dump請(qǐng)求，開(kāi)始推送binary log給slave(也就是canal)
canal解析binary log對(duì)象(原始為byte流)

canal組件架構(gòu)實(shí)現(xiàn)

我們可以大概看看canal是怎么實(shí)現(xiàn)，其內(nèi)部的組件抽取、封裝及其對(duì)應(yīng)的功能實(shí)現(xiàn)，以便我們后續(xù)在使用上更加得心應(yīng)手。

說(shuō)明：

server代表一個(gè)canal運(yùn)行實(shí)例，對(duì)應(yīng)于一個(gè)jvm
instance對(duì)應(yīng)于一個(gè)數(shù)據(jù)隊(duì)列（1個(gè)server對(duì)應(yīng)1..n個(gè)instance)

instance模塊：

eventParser (數(shù)據(jù)源接入，模擬slave協(xié)議和master進(jìn)行交互，協(xié)議解析)
eventSink (Parser和Store鏈接器，進(jìn)行數(shù)據(jù)過(guò)濾，加工，分發(fā)的工作)
eventStore (數(shù)據(jù)存儲(chǔ))
metaManager (增量訂閱&消費(fèi)信息管理器)

2.canal部署安裝

上面我們知道canal是通過(guò)把自己偽裝成mysql slave，收集binlog做解析，然后再進(jìn)行后續(xù)同步操作。所以我們的準(zhǔn)備工作必須要求MySQL開(kāi)啟binlog日志：

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

授權(quán) canal 鏈接 MySQL 賬號(hào)具有作為 MySQL slave 的權(quán)限, 如果已有賬戶(hù)可直接 grant

CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

當(dāng)然也可以不用新增賬號(hào)，直接使用root賬號(hào)，為了方便快捷，我下面的案例就是使用root賬號(hào)的哦，當(dāng)然這不符合開(kāi)發(fā)規(guī)范，root權(quán)限賬號(hào)一般人不能用的~

接下來(lái)就是安裝canal了，安裝方式主要分為直接下載安裝包在服務(wù)器通過(guò)命令運(yùn)行和使用docker容器化方式部署，docker容器部署雖然簡(jiǎn)單快捷，但是考慮到不是人人都了解docker，所以我們這里采用直接使用安裝包命令運(yùn)行。

安裝包命令運(yùn)行其實(shí)很簡(jiǎn)單，官網(wǎng)教程步驟也很詳細(xì)：https://github.com/alibaba/canal/wiki/QuickStart，這里我下載version為1.1.5的，在官網(wǎng)下載安裝包解壓之后文件如下：

bin  canal.deployer-1.1.5.tar.gz  conf  lib  logs  plugin

主要看看conf目錄下配置文件：

canal.properties    example    logback.xml    metrics  spring

canal.properties是啟動(dòng)canal server的配置文件：這里面有很多配置，我粘貼部分來(lái)講講

#################################################
#########               destinations            #############
#################################################
canal.destinations = example
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5
# set this value to 'true' means that when binlog pos not found, skip to latest.
# WARN: pls keep 'false' in production env, or if you know what you want.
canal.auto.reset.latest.pos.mode = false
?
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
?
canal.instance.global.mode = spring
canal.instance.global.lazy = false
canal.instance.global.manager.address = ${canal.admin.manager}
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml
?

canal.destinations = example就是指定instance實(shí)例的查找位置，如果我們一個(gè)canal server需要監(jiān)聽(tīng)多個(gè)instance(平時(shí)各個(gè)業(yè)務(wù)線的數(shù)據(jù)庫(kù)都是獨(dú)立的如商品product，倉(cāng)庫(kù)warehouse)，一個(gè)instance監(jiān)聽(tīng)一個(gè)數(shù)據(jù)庫(kù)，這是最常見(jiàn)的需求了，這時(shí)候我就需要配置多個(gè)instance，可以直接把example文件夾拷貝兩份，分別用數(shù)據(jù)庫(kù)命名新文件夾這樣方便我們快速了解該文件夾對(duì)應(yīng)的instance是哪個(gè)業(yè)務(wù)線的。然后就是調(diào)整canal.properties

canal.destinations = product,warehouse

緊接著就是修改每個(gè)instance文件下的instance.properties，適配監(jiān)聽(tīng)的數(shù)據(jù)庫(kù)配置信息

## mysql serverId , v1.0.26+ will autoGen
## v1.0.26版本后會(huì)自動(dòng)生成slaveId，所以可以不用配置
# canal.instance.mysql.slaveId=0
?
# 數(shù)據(jù)庫(kù)地址
canal.instance.master.address=127.0.0.1:3306
# binlog日志名稱(chēng)
canal.instance.master.journal.name=mysql-bin.000001
# mysql主庫(kù)鏈接時(shí)起始的binlog偏移量
canal.instance.master.position=154
# mysql主庫(kù)鏈接時(shí)起始的binlog的時(shí)間戳
canal.instance.master.timestamp=
canal.instance.master.gtid=
?
# username/password
# 在MySQL服務(wù)器授權(quán)的賬號(hào)密碼
canal.instance.dbUsername=canal
canal.instance.dbPassword=Canal@123456
# 字符集
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
?
# table regex .*\..*表示監(jiān)聽(tīng)所有表 也可以寫(xiě)具體的表名，用，隔開(kāi)
canal.instance.filter.regex=.*\..*
# mysql 數(shù)據(jù)解析表的黑名單，多個(gè)表用，隔開(kāi)
canal.instance.filter.black.regex=
?

最后在安裝包目錄下執(zhí)行以下命令就可以啟動(dòng)了：

sh bin/startup.sh

是不是很簡(jiǎn)單?。。?但是你有沒(méi)有發(fā)現(xiàn)這種方式每新增一個(gè)instance，都需要修改配置文件并重啟，這樣會(huì)導(dǎo)致數(shù)據(jù)同步中斷不太友好，而且也沒(méi)有canal server服務(wù)的狀態(tài)監(jiān)控，著實(shí)覺(jué)得這框架不夠完善。阿里巴巴也考慮到了這些問(wèn)題，所以提供了canal-admin，canal-admin設(shè)計(jì)上是為canal提供整體配置管理、節(jié)點(diǎn)運(yùn)維等面向運(yùn)維的功能，提供相對(duì)友好的WebUI操作界面，方便更多用戶(hù)快速和安全的操作。注意：canal-admin有以下限制要求：

MySQL，用于存儲(chǔ)配置和節(jié)點(diǎn)等相關(guān)數(shù)據(jù) canal版本，要求>=1.1.4 (需要依賴(lài)canal-server提供面向admin的動(dòng)態(tài)運(yùn)維管理接口)

在官網(wǎng)下載canal-admin的安裝包解壓如下：

bin  canal.admin-1.1.5.tar.gz  conf  lib  logs

直接來(lái)看conf下的文件：

application.yml  canal_manager.sql  canal-template.properties  instance-template.properties  logback.xml  public

這里看到的就是一個(gè)spring boot框架開(kāi)發(fā)的web項(xiàng)目啦，anal_manager.sql就是canal-admin服務(wù)所依賴(lài)的數(shù)據(jù)庫(kù)初始化腳本，我們得去MySQL執(zhí)行，然后修改配置文件application.yml

server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
?
spring.datasource:
  address: 10.10.0.10:3306
  database: canal_manager
  username: root
  password: root
  driver-class-name: com.mysql.jdbc.Driver
  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1
?
canal:
  adminUser: admin
  adminPasswd: admin
?

這里就配置一下前面執(zhí)行SQL腳本數(shù)據(jù)庫(kù)的連接信息即可，當(dāng)然如果端口8089被占用了就改成別的，到時(shí)候canal server配置對(duì)應(yīng)的就行。在canal-admin的目錄執(zhí)行下面命令就能啟動(dòng)了：

sh bin/startup.sh

這時(shí)候通過(guò)主機(jī)ip:8089就能在瀏覽器訪問(wèn):

默認(rèn)登錄用戶(hù)名密碼：admin/123456，成功進(jìn)入之后：

我們可以通過(guò)界面管理canal集群、canal server 、server下的instance。這樣無(wú)論是我們修改instance的配置還是新增一個(gè)instance都不需要去服務(wù)器操作并重啟服務(wù)了，是不是很方便，直接通過(guò)界面操作修改、重啟即可。

當(dāng)然還是需要像一開(kāi)始一樣在服務(wù)器啟動(dòng)canal server的，需要把配置canal.properties改成如下：

# register ip
canal.register.ip =
?
# canal admin config
canal.admin.manager = 10.10.0.10:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =
canal.admin.register.name =

這里最主要是綁定關(guān)聯(lián)canal-admin，配置admin的地址信息。這里提一下canal.register.ip這個(gè)配置是和canal集群有關(guān)的，canal集群是依靠zookeeper實(shí)現(xiàn)，這里就不展開(kāi)細(xì)講了。成功啟動(dòng)canal server之后，就可以在admin界面看到了：

然后我們可以基于canal server新增instance：mall和fast-api

這時(shí)候你來(lái)查看canal server 下的配置目錄conf下：

canal.properties example fast-api logback.xml mall metrics spring

發(fā)現(xiàn)多了兩個(gè)目錄mall和fast-api，這就是對(duì)應(yīng)我們前面在界面上創(chuàng)建的兩個(gè)instance，admin通過(guò)關(guān)聯(lián)canal server自動(dòng)幫我們生成，是不是很完美?。?！

3.Spring Boot整合canal

3.1數(shù)據(jù)庫(kù)與緩存一致性問(wèn)題概述

自有了緩存的那一天起，緩存與數(shù)據(jù)庫(kù)數(shù)據(jù)一致性問(wèn)題就一直伴隨著后端開(kāi)發(fā)者，所以如何保證數(shù)據(jù)庫(kù)與緩存雙寫(xiě)數(shù)據(jù)一致性也成為了面試的一個(gè)高頻考點(diǎn)。對(duì)于緩存的更新肯定來(lái)自于業(yè)務(wù)的觸發(fā)，且最終的邏輯處理數(shù)據(jù)是需要落庫(kù)的，只是我們需要考慮的是先更新DB還是先更新緩存？是更新緩存還是刪除緩存？在常規(guī)情況下，怎么操作都可以，但一旦面對(duì)高并發(fā)場(chǎng)景，就值得細(xì)細(xì)思量了。接下來(lái)我們就來(lái)分別看看先寫(xiě)數(shù)據(jù)庫(kù)或者先寫(xiě)緩存有啥問(wèn)題？

這里我們假設(shè)我們的某個(gè)業(yè)務(wù)功能請(qǐng)求就是修改某個(gè)數(shù)據(jù)值

先寫(xiě) MySQL，再寫(xiě) Redis

請(qǐng)求 A、B 都是先寫(xiě) MySQL，然后再寫(xiě) Redis，在高并發(fā)情況下，如果請(qǐng)求 A 在寫(xiě) Redis 時(shí)卡了一會(huì)，請(qǐng)求 B 已經(jīng)依次完成數(shù)據(jù)的更新，就會(huì)出現(xiàn)圖中的問(wèn)題。并發(fā)場(chǎng)景下，這樣的情況是很容易出現(xiàn)的，每個(gè)線程的操作先后順序不同，這樣就導(dǎo)致請(qǐng)求B的緩存值被請(qǐng)求A給覆蓋了，數(shù)據(jù)庫(kù)中是線程B的新值，緩存中是線程A的舊值，并且會(huì)一直這么臟下去直到緩存失效（如果你設(shè)置了過(guò)期時(shí)間的話）。

先寫(xiě)Redis，再寫(xiě)MySQL

和上面一樣只是調(diào)換了寫(xiě)入數(shù)據(jù)庫(kù)與緩存的順序，直接看圖：

高并發(fā)場(chǎng)景下一樣有數(shù)據(jù)一致性問(wèn)題。

還有數(shù)據(jù)更新刪除緩存、延時(shí)雙刪緩存等解決一致性問(wèn)題方式，這里就不一一列舉了，本質(zhì)上上面我只是引出數(shù)據(jù)庫(kù)與緩存雙寫(xiě)一致性問(wèn)題，畢竟我們今天主題是canal，不是雙寫(xiě)一致性問(wèn)題解決方案詳解，有興趣可以自行查閱這個(gè)高頻面試知識(shí)。

3.2 整合canel

canal官方?jīng)]有提供與spring-boot框架快速整合的starter，根據(jù)官網(wǎng)示例直接使用canal client直連canal server操作：

引入依賴(lài)：

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

示例：我們?cè)谏厦娴腸anal admin創(chuàng)建一個(gè)mall的實(shí)例監(jiān)聽(tīng)mall數(shù)據(jù)庫(kù)變化，接下來(lái)我通過(guò)新增、修改、刪除品牌的一條數(shù)據(jù)。

package com.shepherd.common.canal;
?
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
?
import java.net.InetSocketAddress;
import java.util.List;
?
public class CanalClient {
?
    public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {
?
        // 創(chuàng)建canal客戶(hù)端，單鏈接模式
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("10.10.0.10",
                11111), "mall", "", "");
        // 創(chuàng)建連接
        canalConnector.connect();
        while (true) {
            // 訂閱數(shù)據(jù)庫(kù)
            // canalConnector.subscribe("mall");
?
            // 獲取數(shù)據(jù)
            Message message = canalConnector.get(100);
?
            // 獲取Entry集合
            List<CanalEntry.Entry> entries = message.getEntries();
?
            // 判斷集合是否為空,如果為空,則等待一會(huì)繼續(xù)拉取數(shù)據(jù)
            if (entries.size() <= 0) {
//                System.out.println("當(dāng)次抓取沒(méi)有數(shù)據(jù)，休息一會(huì)。。。。。。");
                Thread.sleep(1000);
            } else {
                // 遍歷entries，單條解析
                for (CanalEntry.Entry entry : entries) {
?
                    //1.獲取表名
                    String tableName = entry.getHeader().getTableName();
?
                    //2.獲取類(lèi)型
                    CanalEntry.EntryType entryType = entry.getEntryType();
?
                    //3.獲取序列化后的數(shù)據(jù)
                    ByteString storeValue = entry.getStoreValue();
?
                    //4.判斷當(dāng)前entryType類(lèi)型是否為ROWDATA
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
?
                        //5.反序列化數(shù)據(jù)
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
?
                        //6.獲取當(dāng)前事件的操作類(lèi)型
                        CanalEntry.EventType eventType = rowChange.getEventType();
?
                        //7.獲取數(shù)據(jù)集
                        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
?
                        //8.遍歷rowDataList，并打印數(shù)據(jù)集
                        for (CanalEntry.RowData rowData : rowDataList) {
?
                            JSONObject beforeData = new JSONObject();
                            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                            for (CanalEntry.Column column : beforeColumnsList) {
                                beforeData.put(column.getName(), column.getValue());
                            }
?
                            JSONObject afterData = new JSONObject();
                            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                            for (CanalEntry.Column column : afterColumnsList) {
                                afterData.put(column.getName(), column.getValue());
                            }
?
                            //數(shù)據(jù)打印
                            System.out.println("Table:" + tableName +
                                    ",EventType:" + eventType +
                                    ",Before:" + beforeData +
                                    ",After:" + afterData);
                        }
                    }
                }
            }
        }
    }
}
?
?

控制臺(tái)打印如下：

?
Table:brand,EventType:INSERT,Before:{},After:{"image":"","update_time":"","category_id":"1","create_time":"","letter":"H","name":"huawei","description":"世界第一名族企業(yè)","id":"1","is_delete":""}
?
Table:brand,EventType:UPDATE,Before:{"image":"","update_time":"","category_id":"1","create_time":"","letter":"H","name":"huawei","description":"世界第一名族企業(yè)","id":"1","is_delete":""},After:{"image":"http://www.baidu.com/image1.png","update_time":"","category_id":"1","create_time":"","letter":"H","name":"huawei111","description":"世界第一名族企業(yè)","id":"1","is_delete":""}
?
Table:brand,EventType:DELETE,Before:{"image":"http://www.baidu.com/image1.png","update_time":"","category_id":"1","create_time":"","letter":"H","name":"huawei111","description":"世界第一名族企業(yè)","id":"1","is_delete":""},After:{}
?

可以看到canal監(jiān)控到表數(shù)據(jù)的變更方式以及數(shù)據(jù)的前后變化。這樣我們就可以通過(guò)canal監(jiān)聽(tīng)MySQL binlog原理優(yōu)雅實(shí)現(xiàn)緩存與數(shù)據(jù)庫(kù)數(shù)據(jù)一致性解決方案啦，通過(guò)的監(jiān)聽(tīng)得到數(shù)據(jù)信息進(jìn)行緩存同步寫(xiě)操作即可。

4.總結(jié)

canal是一個(gè)增量數(shù)據(jù)同步組件，其好處就是在于對(duì)業(yè)務(wù)邏輯無(wú)侵入，它是通過(guò)把自己偽裝成mysql slave收集binlog實(shí)現(xiàn)數(shù)據(jù)同步的。這里要強(qiáng)調(diào)一下異構(gòu)數(shù)據(jù)源之間要實(shí)現(xiàn)數(shù)據(jù)增量同步，同時(shí)要保證實(shí)時(shí)性、低延時(shí)在大數(shù)據(jù)領(lǐng)域也是一個(gè)令人頭疼的問(wèn)題，不像全量同步簡(jiǎn)單直接全部數(shù)據(jù)寫(xiě)到目的源就可以了。canal就是為了解決增量同步而生，這是其招牌。當(dāng)然canal也是有缺點(diǎn)的，只能監(jiān)聽(tīng)MySQL，其他數(shù)據(jù)庫(kù)oracle就不行了。

同時(shí)也要指明企業(yè)級(jí)的數(shù)據(jù)同步不可能像上面的方式一條條監(jiān)聽(tīng)數(shù)據(jù)變化同步異構(gòu)數(shù)據(jù)源如elasticsearch、Hbase的，因?yàn)橐粭l條處理同步速度之慢不言而喻，當(dāng)然通過(guò)上面方式同步數(shù)據(jù)到緩存redis是可以的，因?yàn)榫彺娴臄?shù)據(jù)一般變化不頻繁且數(shù)據(jù)量不大，但同步其他大數(shù)據(jù)組件就一般都需要批量同步了，這時(shí)候就需要借助消息隊(duì)列中間件如kafka進(jìn)行數(shù)據(jù)堆積從而實(shí)現(xiàn)批量同步，canal+kafka+elasticsearch這個(gè)架構(gòu)是當(dāng)下許多企業(yè)中較常見(jiàn)的一種數(shù)據(jù)同步的方案。

以上就是SpringBoot整合canal實(shí)現(xiàn)數(shù)據(jù)緩存一致性解決方案的詳細(xì)內(nèi)容，更多關(guān)于SpringBoot canal數(shù)據(jù)緩存一致性的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: