目錄

一. 前言

二. Canal 簡(jiǎn)介和使用場(chǎng)景

2.1. Canal 簡(jiǎn)介

2.2. Canal 使用場(chǎng)景

三. Canal Server 設(shè)計(jì)

3.1. 整體設(shè)計(jì)

3.2. EventParser 設(shè)計(jì)

3.3.?CanalLogPositionManager 設(shè)計(jì)

3.4.?CanalHAController 類(lèi)圖設(shè)計(jì)

3.5.?EventSink 類(lèi)圖設(shè)計(jì)和擴(kuò)展

3.6.?EventStore 類(lèi)圖設(shè)計(jì)和擴(kuò)展

3.7.?MetaManager 類(lèi)圖設(shè)計(jì)和擴(kuò)展

四. Canal?Client 設(shè)計(jì)

4.1. 整體設(shè)計(jì)

4.2.?Server/Client交互協(xié)議

五. Canal 配置信息

5.1. Canal 配置方式

5.2.?canal.properties

5.3.?instance.properties

5.4.?instance.xml 配置文件

六. Canal 使用

6.1. Canal 下載

6.2.?MySql 配置

6.3.?canal-server 使用

6.4.?canal-adapter 使用

6.5.?canal-admin 使用

6.6.?數(shù)據(jù)同步演示

一. 前言

? ? Canal 是阿里開(kāi)源的一款基于 MySql 數(shù)據(jù)庫(kù) binlog 的增量訂閱和消費(fèi)組件，通過(guò)它可以訂閱數(shù)據(jù)庫(kù)的 binlog 日志，然后進(jìn)行一些數(shù)據(jù)消費(fèi)，如數(shù)據(jù)鏡像、數(shù)據(jù)異構(gòu)、數(shù)據(jù)索引、緩存更新等。相對(duì)于消息隊(duì)列，通過(guò)這種機(jī)制可以實(shí)現(xiàn)數(shù)據(jù)的有序化和一致性。

? ? Canal 主要用途是對(duì) MySql 數(shù)據(jù)庫(kù)增量日志進(jìn)行解析，提供增量數(shù)據(jù)的訂閱和消費(fèi)，簡(jiǎn)單說(shuō)就是可以對(duì) MySql 的增量數(shù)據(jù)進(jìn)行實(shí)時(shí)同步，支持同步到 MySql、ElasticSearch、HBase 等數(shù)據(jù)存儲(chǔ)中去。

二. Canal 簡(jiǎn)介和使用場(chǎng)景

2.1. Canal 簡(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 對(duì)外提供增量數(shù)據(jù)訂閱和消費(fèi)，提供 Kafka、RocketMQ、RabbitMq、Es、Tcp 等組件來(lái)消費(fèi)。

2.2. Canal 使用場(chǎng)景

1. 同步緩存 Redis/全文搜索 ES：Canal 一個(gè)常見(jiàn)應(yīng)用場(chǎng)景是同步緩存/全文搜索，當(dāng)數(shù)據(jù)庫(kù)變更后通過(guò) binlog 進(jìn)行緩存/ES 的增量更新。當(dāng)緩存/ES 更新出現(xiàn)問(wèn)題時(shí)，應(yīng)該回退 binlog 到過(guò)去某個(gè)位置進(jìn)行重新同步，并提供全量刷新緩存/ES 的方法。

2. 下發(fā)任務(wù)：另一種常見(jiàn)應(yīng)用場(chǎng)景是下發(fā)任務(wù)，當(dāng)數(shù)據(jù)變更時(shí)需要通知其他依賴(lài)系統(tǒng)。其原理是任務(wù)系統(tǒng)監(jiān)聽(tīng)數(shù)據(jù)庫(kù)變更，然后將變更的數(shù)據(jù)寫(xiě)入 MQ（比如 Kafka） 進(jìn)行任務(wù)下發(fā)，比如商品數(shù)據(jù)變更后需要通知商品詳情頁(yè)、列表頁(yè)、搜索頁(yè)等相關(guān)系統(tǒng)。這種方式可以保證數(shù)據(jù)下發(fā)的精確性，通過(guò) MQ 發(fā)送消息通知變更緩存是無(wú)法做到這一點(diǎn)的，而且業(yè)務(wù)系統(tǒng)中不會(huì)散落著各種下發(fā) MQ 的代碼，從而實(shí)現(xiàn)了下發(fā)歸集。

3. 數(shù)據(jù)異構(gòu)：在大型網(wǎng)站架構(gòu)中，DB 都會(huì)采用分庫(kù)分表來(lái)解決容量和性能問(wèn)題，但分庫(kù)分表之后帶來(lái)的新問(wèn)題。比如不同維度的查詢或者聚合查詢，此時(shí)就會(huì)非常棘手。一般我們會(huì)通過(guò)數(shù)據(jù)異構(gòu)機(jī)制來(lái)解決此問(wèn)題。所謂的數(shù)據(jù)異構(gòu)，那就是將需要 join 查詢的多表按照某一個(gè)維度又聚合在一個(gè)DB 中，讓你去查詢。Canal 就是實(shí)現(xiàn)數(shù)據(jù)異構(gòu)的手段之一。

三. Canal Server 設(shè)計(jì)

3.1. 整體設(shè)計(jì)

Server 代表一個(gè) Canal 運(yùn)行實(shí)例，對(duì)應(yīng)于一個(gè) JVM。

Instance 對(duì)應(yīng)于一個(gè)數(shù)據(jù)隊(duì)列（1個(gè) Canal Server 對(duì)應(yīng) 1..n 個(gè) Instance），Instance 下的子模塊：

1. EventParser：數(shù)據(jù)源接入，模擬 slave 協(xié)議和 master 進(jìn)行交互，協(xié)議解析；
2. EventSink：Parser 和 Store 鏈接器，進(jìn)行數(shù)據(jù)過(guò)濾，加工，分發(fā)的工作；
3. EventStore：數(shù)據(jù)存儲(chǔ)；
4. MetaManager：增量訂閱 & 消費(fèi)信息管理器。

整體類(lèi)圖設(shè)計(jì)：

• CanalLifeCycle：所有 Canal 模塊的生命周期接口；
• CanalInstance：組合 Parser、Sink、Store 三個(gè)子模塊，三個(gè)子模塊的生命周期統(tǒng)一受 CanalInstance 管理；
• CanalServer：聚合了多個(gè) CanalInstance。

3.2. EventParser 設(shè)計(jì)

每個(gè) EventParser 都會(huì)關(guān)聯(lián)兩個(gè)內(nèi)部組件：

1. CanalLogPositionManager : 記錄binlog 最后一次解析成功位置信息，主要是描述下一次canal啟動(dòng)的位點(diǎn)
2. CanalHAController: 控制 EventParser 的鏈接主機(jī)管理，判斷當(dāng)前該鏈接哪個(gè)mysql數(shù)據(jù)庫(kù)

目前開(kāi)源版本只支持 MySql binlog , 默認(rèn)通過(guò) MySql binlog dump 遠(yuǎn)程獲取 binlog，但也可以使用 LocalBinlog - 類(lèi) relay log 模式，直接消費(fèi)本地文件中的 binlog。

3.3.?CanalLogPositionManager 設(shè)計(jì)

• 如果 CanalEventStore 選擇的是內(nèi)存模式，可不保留解析位置，下一次 Canal 啟動(dòng)時(shí)直接依賴(lài) CanalMetaManager 記錄的最后一次消費(fèi)成功的位點(diǎn)即可（最后一次 ack 提交的數(shù)據(jù)位點(diǎn)）。
• 如果 CanalEventStore 選擇的是持久化模式，可通過(guò) Zookeeper 記錄位點(diǎn)信息，Canal Instance 發(fā)生 failover 切換到另一臺(tái)機(jī)器，可通過(guò)讀取 Zookeeper 獲取位點(diǎn)信息。
• 可通過(guò)實(shí)現(xiàn)自己的 CanalLogPositionManager，比如記錄位點(diǎn)信息到本地文件 /nas 文件實(shí)現(xiàn)簡(jiǎn)單可用的無(wú) HA 模式。

3.4.?CanalHAController 類(lèi)圖設(shè)計(jì)

• 失敗檢測(cè)常見(jiàn)方式可定時(shí)發(fā)送心跳語(yǔ)句到當(dāng)前鏈接的數(shù)據(jù)庫(kù)，超過(guò)一定次數(shù)檢測(cè)失敗時(shí)，嘗試切換到備機(jī)。
• 如果有一套數(shù)據(jù)庫(kù)主備信息管理系統(tǒng)，當(dāng)數(shù)據(jù)庫(kù)主備切換或者機(jī)器下線，推送配置到各個(gè)應(yīng)用節(jié)點(diǎn)，HAController 收到后，控制 EventParser 進(jìn)行鏈接切換。

3.5.?EventSink 類(lèi)圖設(shè)計(jì)和擴(kuò)展

• 數(shù)據(jù)過(guò)濾：支持通配符的過(guò)濾模式，表名，字段內(nèi)容等。
• 數(shù)據(jù)路由/分發(fā)：解決 1:n（1個(gè) Parser 對(duì)應(yīng)多個(gè) Store 的模式）。
• 數(shù)據(jù)歸并：解決 n:1（多個(gè) Parser 對(duì)應(yīng)1個(gè) Store）。
• 數(shù)據(jù)加工：在進(jìn)入 store 之前進(jìn)行額外的處理，比如 join。

數(shù)據(jù) 1:n 業(yè)務(wù)：

? ? 為了合理的利用數(shù)據(jù)庫(kù)資源， 一般常見(jiàn)的業(yè)務(wù)都是按照 schema 進(jìn)行隔離，然后在 MySql 上層或者 dao 這一層面上，進(jìn)行一個(gè)數(shù)據(jù)源路由，屏蔽數(shù)據(jù)庫(kù)物理位置對(duì)開(kāi)發(fā)的影響，阿里系主要是通過(guò) cobar/tddl 來(lái)解決數(shù)據(jù)源路由問(wèn)題。所以，一般一個(gè)數(shù)據(jù)庫(kù)實(shí)例上，會(huì)部署多個(gè) schema，每個(gè) schema 會(huì)有1個(gè)或者多個(gè)業(yè)務(wù)方關(guān)注。

數(shù)據(jù) n:1 業(yè)務(wù)：

? ? 同樣，當(dāng)一個(gè)業(yè)務(wù)的數(shù)據(jù)規(guī)模達(dá)到一定的量級(jí)后，必然會(huì)涉及到水平拆分和垂直拆分的問(wèn)題，針對(duì)這些拆分的數(shù)據(jù)需要處理時(shí)，就需要鏈接多個(gè) Store 進(jìn)行處理，消費(fèi)的位點(diǎn)就會(huì)變成多份，而且數(shù)據(jù)消費(fèi)的進(jìn)度無(wú)法得到盡可能有序的保證。所以，在一定業(yè)務(wù)場(chǎng)景下，需要將拆分后的增量數(shù)據(jù)進(jìn)行歸并處理，比如按照時(shí)間戳/全局Id 進(jìn)行排序歸并。

3.6.?EventStore 類(lèi)圖設(shè)計(jì)和擴(kuò)展

1. 抽象 CanalStoreScavenge ， 解決數(shù)據(jù)的清理，比如定時(shí)清理，滿了之后清理，每次 ack 清理等。

2. CanalEventStore 接口，主要包含 put/get/ack/rollback 的相關(guān)接口。put/get 操作會(huì)組成一個(gè)生產(chǎn)者/消費(fèi)者模式，每個(gè) Store 都會(huì)有存儲(chǔ)大小設(shè)計(jì)，存儲(chǔ)滿了，put 操作會(huì)阻塞等待 get 獲取數(shù)據(jù)，所以不會(huì)無(wú)線占用存儲(chǔ)，比如內(nèi)存大小：

• EventStore 目前實(shí)現(xiàn)了 memory 模式，支持按照內(nèi)存大小和內(nèi)存記錄數(shù)進(jìn)行存儲(chǔ)大小限制。
• 后續(xù)可開(kāi)發(fā)基于本地文件的存儲(chǔ)模式。
• 基于文件存儲(chǔ)和內(nèi)存存儲(chǔ)，開(kāi)發(fā) mixed 模式，做成兩級(jí)隊(duì)列，內(nèi)存 buffer 有空位時(shí)，將文件的數(shù)據(jù)讀入到內(nèi)存 buffer 中（可以通過(guò)配置進(jìn)行配置）。
• mixed 模式實(shí)現(xiàn)可以讓 Canal 落地消費(fèi)/訂閱的模型，取 1 份 binlog 數(shù)據(jù)，提供多個(gè)客戶端消費(fèi)，消費(fèi)有快有慢，各自保留消費(fèi)位點(diǎn)。

3.7.?MetaManager 類(lèi)圖設(shè)計(jì)和擴(kuò)展

• MetaManager 目前支持了多種模式，最頂層 Memory 和 Zookeeper 模式，然后是 mixed 模式-先寫(xiě)內(nèi)存，再寫(xiě) Zookeeper。
• 可通過(guò)實(shí)現(xiàn)自己的 CanalMetaManager，比如記錄位點(diǎn)信息到本地文件 /nas 文件，簡(jiǎn)單可用的無(wú) HA 模式。

四. Canal?Client 設(shè)計(jì)

4.1. 整體設(shè)計(jì)

在了解具體 API 之前，需要提前了解下 Canal Client 的類(lèi)設(shè)計(jì)，這樣才可以正確的使用好 Canal。

大致分為幾部分：

1. ClientIdentity：Canal Client 和 Server 交互之間的身份標(biāo)識(shí)，目前 clientId 寫(xiě)死為1001。（目前 Canal Server 上的一個(gè) Instance 只能有一個(gè) Client 消費(fèi)，ClientId 的設(shè)計(jì)是為1個(gè)Instance 多 Client 消費(fèi)模式而預(yù)留的，暫時(shí)不需要理會(huì)）。
2. CanalConnector：SimpleCanalConnector/ClusterCanalConnector 是兩種 Connector 的實(shí)現(xiàn)，Simple 針對(duì)的是簡(jiǎn)單的 ip 直連模式，Cluster針對(duì)多 ip 的模式，可依賴(lài)CanalNodeAccessStrategy 進(jìn)行 failover 控制。
3. CanalNodeAccessStrategy：SimpleNodeAccessStrategy/ClusterNodeAccessStrategy 是兩種 failover 的實(shí)現(xiàn)，Simple 針對(duì)給定的初始 ip 列表進(jìn)行 failover 選擇，Cluster 基于Zookeeper上的 Cluster 節(jié)點(diǎn)動(dòng)態(tài)選擇正在運(yùn)行的 Canal Server。
4. ClientRunningMonitor/ClientRunningListener/ClientRunningData：Client Running 相關(guān)控制，主要為解決 Client 自身的 failover 機(jī)制。Canal Client 允許同時(shí)啟動(dòng)多個(gè) Canal Client，通過(guò) Running 機(jī)制，可保證只有一個(gè) Client 在工作，其他 Client 做為冷備。當(dāng)運(yùn)行中的Client 掛了，Running 會(huì)控制讓冷備中的 Client 轉(zhuǎn)為工作模式，這樣就可以確保 Canal Client 也不會(huì)是單點(diǎn)，保證整個(gè)系統(tǒng)的高可用性。

4.2.?Server/Client交互協(xié)議

get/ack/rollback 協(xié)議介紹：

1. Message getWithoutAck(int batchSize)，允許指定 batchSize，一次可以獲取多條，每次返回的對(duì)象為 message，包含的內(nèi)容為：
   a. batch id：唯一標(biāo)識(shí)；
   b. entries：具體的數(shù)據(jù)對(duì)象，可參見(jiàn)下面的數(shù)據(jù)介紹。
2. getWithoutAck(int batchSize, Long timeout, TimeUnit unit)，相比于 getWithoutAck(int batchSize)，允許設(shè)定獲取數(shù)據(jù)的 timeout 超時(shí)時(shí)間：
   a. 拿夠 batchSize 條記錄或者超過(guò) timeout 時(shí)間；
   b. timeout=0，阻塞等到足夠的 batchSize。
3. void rollback(long batchId)，顧命思議，回滾上次的 get 請(qǐng)求，重新獲取數(shù)據(jù)?；?get 獲取的 batchId 進(jìn)行提交，避免誤操作。
4. void ack(long batchId)，顧命思議，確認(rèn)已經(jīng)消費(fèi)成功，通知 Server 刪除數(shù)據(jù)?；?get獲取的 batchId 進(jìn)行提交，避免誤操作。

Canal 的 get/ack/rollback 協(xié)議和常規(guī)的 jms 協(xié)議有所不同，允許 get/ack 異步處理，比如可以連續(xù)調(diào)用 get 多次，后續(xù)異步按順序提交 ack/rollback，項(xiàng)目中稱(chēng)之為流式 API。

流式 API 設(shè)計(jì)的好處：

1. get/ack 異步化，減少因 ack 帶來(lái)的網(wǎng)絡(luò)延遲和操作成本（99%的狀態(tài)都是處于正常狀態(tài)，異常的 rollback 屬于個(gè)別情況，沒(méi)必要為個(gè)別的 case 犧牲整個(gè)性能）。
2. get 獲取數(shù)據(jù)后，業(yè)務(wù)消費(fèi)存在瓶頸或者需要多進(jìn)程/多線程消費(fèi)時(shí)，可以不停的輪詢 get 數(shù)據(jù)，不停的往后發(fā)送任務(wù)，提高并行化。（作者在實(shí)際業(yè)務(wù)中的一個(gè) case：業(yè)務(wù)數(shù)據(jù)消費(fèi)需要跨中美網(wǎng)絡(luò)，所以一次操作基本在 200ms 以上，為了減少延遲，所以需要實(shí)施并行化）。

流式 API 設(shè)計(jì)：

• 每次 get 操作都會(huì)在 meta 中產(chǎn)生一個(gè) mark，mark 標(biāo)記會(huì)遞增，保證運(yùn)行過(guò)程中 mark 的唯一性；
• 每次的 get 操作，都會(huì)在上一次的 mark 操作記錄的 cursor 繼續(xù)往后取，如果 mark 不存在，則在 last ack cursor 繼續(xù)往后取；
• 進(jìn)行 ack 時(shí)，需要按照 mark 的順序進(jìn)行數(shù)序 ack，不能跳躍 ack。ack 會(huì)刪除當(dāng)前的 mark 標(biāo)記，并將對(duì)應(yīng)的 mark 位置更新為 last ack cursor；
• 一旦出現(xiàn)異常情況，客戶端可發(fā)起 rollback 情況，重新置位：刪除所有的 mark，清理 get 請(qǐng)求位置，下次請(qǐng)求會(huì)從 last ack cursor 繼續(xù)往后取。

流式 API 帶來(lái)的異步響應(yīng)模型：

五. Canal 配置信息

5.1. Canal 配置方式

Canal 配置方式有兩種：

1. ManagerCanalInstanceGenerator：基于 Manager 管理的配置方式，目前 Alibaba 內(nèi)部配置使用這種方式。大家可以實(shí)現(xiàn) CanalConfigClient，連接各自的管理系統(tǒng)，即可完成接入。
2. SpringCanalInstanceGenerator：基于本地 spring xml 的配置方式，目前開(kāi)源版本已經(jīng)自帶該功能所有代碼，建議使用。

Spring 配置：

Spring 配置的原理是將整個(gè)配置抽象為兩部分：

1. xxxx-instance.xml（Canal 組件的配置定義，可以在多個(gè) Instance 配置中共享）；
2. xxxx.properties（每個(gè) Instance 通道都有各自一份定義，因?yàn)槊總€(gè) MySql 的 ip，帳號(hào)，密碼等信息不會(huì)相同）。

通過(guò) Spring 的 PropertyPlaceholderConfigurer 機(jī)制將其融合，生成一份 Instance 實(shí)例對(duì)象，每個(gè)Instance 對(duì)應(yīng)的組件都是相互獨(dú)立的，互不影響。

properties 配置文件分為兩部分：

1. canal.properties（系統(tǒng)根配置文件），下面詳細(xì)說(shuō)明；
2. instance.properties（Instance 級(jí)別的配置文件，每個(gè) Instance 一份）。

5.2.?canal.properties

Canal 配置主要分為兩部分定義：

1. instance 列表定義，（列出當(dāng)前 Server 上有多少個(gè) Instance，每個(gè) Instance 的加載方式是Spring/Manager 等) ? 以下選一些重要的參數(shù)說(shuō)明一下：

2. common 參數(shù)定義，比如可以將 instance.properties 的公用參數(shù)，抽取放置到這里，這樣每個(gè)Instance 啟動(dòng)的時(shí)候就可以共享?！緄nstance.properties 配置定義優(yōu)先級(jí)高于 canal.properties】以下選一些重要的參數(shù)說(shuō)明一下：

5.3.?instance.properties

在 canal.properties 定義了 canal.destinations 后，需要在 canal.conf.dir 對(duì)應(yīng)的目錄下建立同名的文件。

如果 canal.properties 未定義 instance 列表，但開(kāi)啟了 canal.auto.scan 時(shí)：

1. Server 第一次啟動(dòng)時(shí)，會(huì)自動(dòng)掃描 conf 目錄下，將文件名做為 instance name，啟動(dòng)對(duì)應(yīng)的instance；
2. Server 運(yùn)行過(guò)程中，會(huì)根據(jù) canal.auto.scan.interval 定義的頻率，進(jìn)行掃描：
   1. 發(fā)現(xiàn)目錄有新增，啟動(dòng)新的 Instance；
   2. 發(fā)現(xiàn)目錄有刪除，關(guān)閉老的 Instance；
   3. 發(fā)現(xiàn)對(duì)應(yīng)目錄的 instance.properties 有變化，重啟 Instance。

instance.properties 參數(shù)列表（部分）：

幾點(diǎn)說(shuō)明：

1. MySql 鏈接時(shí)的起始位置

1. canal.instance.master.journal.name + ?canal.instance.master.position：精確指定一個(gè) binlog位點(diǎn)，進(jìn)行啟動(dòng)。
2. canal.instance.master.timestamp：指定一個(gè)時(shí)間戳，Canal 會(huì)自動(dòng)遍歷 mysql binlog，找到對(duì)應(yīng)時(shí)間戳的 binlog 位點(diǎn)后，進(jìn)行啟動(dòng)。
3. 不指定任何信息：默認(rèn)從當(dāng)前數(shù)據(jù)庫(kù)的位點(diǎn)，進(jìn)行啟動(dòng)。（show master status）

2. MySql 解析關(guān)注表定義

• 標(biāo)準(zhǔn)的Perl正則，注意轉(zhuǎn)義時(shí)需要雙斜杠：\\。

3. MySql 鏈接的編碼

• 目前 Canal 版本僅支持一個(gè)數(shù)據(jù)庫(kù)只有一種編碼，如果一個(gè)庫(kù)存在多個(gè)編碼，需要通過(guò)filter.regex 配置，將其拆分為多個(gè) canal instance，為每個(gè) Instance 指定不同的編碼。

5.4.?instance.xml 配置文件

目前默認(rèn)支持的 instance.xml 有以下幾種：

1. spring/memory-instance.xml
2. spring/default-instance.xml
3. spring/group-instance.xml

在介紹 instance 配置之前，先了解一下 Canal 如何維護(hù)一份增量訂閱&消費(fèi)的關(guān)系信息：

• 解析位點(diǎn)（Parse 模塊會(huì)記錄，上一次解析 binlog 到了什么位置，對(duì)應(yīng)組件為：CanalLogPositionManager）。
• 消費(fèi)位點(diǎn)（Canal Server 在接收了客戶端的 ack 后，就會(huì)記錄客戶端提交的最后位點(diǎn)，對(duì)應(yīng)的組件為：CanalMetaManager）。

對(duì)應(yīng)的兩個(gè)位點(diǎn)組件，目前都有幾種實(shí)現(xiàn)：

1. Memory（memory-instance.xml 中使用）。
2. Zookeeper。
3. Mixed。
4. Period（default-instance.xml 中使用，集合了 Zookeeper+Memory 模式，先寫(xiě)內(nèi)存，定時(shí)刷新數(shù)據(jù)到 Zookeeper 上）。

memory-instance.xml 介紹：

所有的組件（parser、sink、store）都選擇了內(nèi)存版模式，記錄位點(diǎn)的都選擇了 Memory 模式，重啟后又會(huì)回到初始位點(diǎn)進(jìn)行解析。

特點(diǎn)：速度最快，依賴(lài)最少（不需要 Zookeeper）。

場(chǎng)景：一般應(yīng)用在 quickstart，或者是出現(xiàn)問(wèn)題后，進(jìn)行數(shù)據(jù)分析的場(chǎng)景，不應(yīng)該將其應(yīng)用于生產(chǎn)環(huán)境。

default-instance.xml 介紹：

Store 選擇了內(nèi)存模式，其余的 parser/sink 依賴(lài)的位點(diǎn)管理選擇了持久化模式，目前持久化的方式主要是寫(xiě)入 Zookeeper，保證數(shù)據(jù)集群共享。

特點(diǎn)：支持 HA。

場(chǎng)景：生產(chǎn)環(huán)境，集群化部署。

group-instance.xml 介紹：

主要針對(duì)需要進(jìn)行多庫(kù)合并時(shí)，可以將多個(gè)物理 Instance 合并為一個(gè)邏輯 Instance，提供客戶端訪問(wèn)。

場(chǎng)景：分庫(kù)業(yè)務(wù)。比如產(chǎn)品數(shù)據(jù)拆分了4個(gè)庫(kù)，每個(gè)庫(kù)會(huì)有一個(gè) Instance，如果不用 Group，業(yè)務(wù)上要消費(fèi)數(shù)據(jù)時(shí)，需要啟動(dòng)4個(gè)客戶端，分別鏈接4個(gè) Instance 實(shí)例。使用 Group后，可以在 Canal Server 上合并為一個(gè)邏輯 Instance，只需要啟動(dòng)1個(gè)客戶端，鏈接這個(gè)邏輯 Instance即可。

六. Canal 使用

接下來(lái)我們來(lái)學(xué)習(xí)下 Canal 的使用，以 MySql 實(shí)時(shí)同步數(shù)據(jù)到 ElasticSearch為例。

6.1. Canal 下載

? ? 首先我們需要下載 Canal 的各個(gè)組件 canal-server、canal-adapter、canal-admin。

下載地址：https://github.com/alibaba/canal/releases。

Canal 官方文檔：https://github.com/alibaba/canal/wiki。

Canal 的各個(gè)組件的用途各不相同，下面分別介紹下：

1. canal-server（canal-deploy）：可以直接監(jiān)聽(tīng) MySql 的 binlog，把自己偽裝成 MySql 的從庫(kù)，只負(fù)責(zé)接收數(shù)據(jù)，并不做處理。
2. canal-adapter：相當(dāng)于 Canal 的客戶端，會(huì)從 canal-server 中獲取數(shù)據(jù)，然后對(duì)數(shù)據(jù)進(jìn)行同步，可以同步到 MySql、ElasticSearch 和 HBase 等存儲(chǔ)中去。
3. canal-admin：為 Canal 提供整體配置管理、節(jié)點(diǎn)運(yùn)維等面向運(yùn)維的功能，提供相對(duì)友好的WebUI 操作界面，方便更多用戶快速和安全的操作。

由于不同版本的 MySql、ElasticSearch 和 Canal 會(huì)有兼容性問(wèn)題，所以我們先對(duì)其使用版本做個(gè)約定：

6.2.?MySql 配置

? ? 由于 Canal 是通過(guò)訂閱 MySql 的 binlog 來(lái)實(shí)現(xiàn)數(shù)據(jù)同步的，所以我們需要開(kāi)啟 MySql 的 binlog寫(xiě)入功能，并設(shè)置 binlog-format 為 ROW 模式，我的配置文件為 /mydata/mysql/conf/my.cnf，改為如下內(nèi)容即可：

[mysqld]
## 設(shè)置server_id，同一局域網(wǎng)中需要唯一
server_id=101 
## 指定不需要同步的數(shù)據(jù)庫(kù)名稱(chēng)
binlog-ignore-db=mysql  
## 開(kāi)啟二進(jìn)制日志功能
log-bin=mall-mysql-bin  
## 設(shè)置二進(jìn)制日志使用內(nèi)存大小（事務(wù)）
binlog_cache_size=1M  
## 設(shè)置使用的二進(jìn)制日志格式（mixed,statement,row）
binlog_format=row  
## 二進(jìn)制日志過(guò)期清理時(shí)間。默認(rèn)值為0，表示不自動(dòng)清理。
expire_logs_days=7  
## 跳過(guò)主從復(fù)制中遇到的所有錯(cuò)誤或指定類(lèi)型的錯(cuò)誤，避免slave端復(fù)制中斷。
## 如：1062錯(cuò)誤是指一些主鍵重復(fù)，1032錯(cuò)誤是因?yàn)橹鲝臄?shù)據(jù)庫(kù)數(shù)據(jù)不一致
slave_skip_errors=1062

配置完成后需要重新啟動(dòng) MySql，重啟成功后通過(guò)如下命令查看 binlog 是否啟用：

show variables like '%log_bin%'
+---------------------------------+-------------------------------------+
| Variable_name                   | Value                               |
+---------------------------------+-------------------------------------+
| log_bin                         | ON                                  |
| log_bin_basename                | /var/lib/mysql/mall-mysql-bin       |
| log_bin_index                   | /var/lib/mysql/mall-mysql-bin.index |
| log_bin_trust_function_creators | OFF                                 |
| log_bin_use_v1_row_events       | OFF                                 |
| sql_log_bin                     | ON                                  |
+---------------------------------+-------------------------------------+

再查看下 MySql 的 binlog 模式：

show variables like 'binlog_format%';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW   |
+---------------+-------+

接下來(lái)需要?jiǎng)?chuàng)建一個(gè)擁有從庫(kù)權(quán)限的賬號(hào)，用于訂閱 binlog，這里創(chuàng)建的賬號(hào)為 canal:canal：

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

創(chuàng)建好測(cè)試用的數(shù)據(jù)庫(kù) canal-test，之后創(chuàng)建一張商品表 product，建表語(yǔ)句如下：

CREATE TABLE `product`  (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `title` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `sub_title` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `price` decimal(10, 2) NULL DEFAULT NULL,
  `pic` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 2 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

6.3.?canal-server 使用

? ? 將我們下載好的壓縮包 canal.deployer-1.1.5-SNAPSHOT.tar.gz 上傳到 Linux 服務(wù)器，然后解壓到指定目錄 /mydata/canal-server，可使用如下命令解壓：

tar -zxvf canal.deployer-1.1.5-SNAPSHOT.tar.gz

解壓完成后目錄結(jié)構(gòu)如下：

├── bin
│   ├── restart.sh
│   ├── startup.bat
│   ├── startup.sh
│   └── stop.sh
├── conf
│   ├── canal_local.properties
│   ├── canal.properties
│   └── example
│       └── instance.properties
├── lib
├── logs
│   ├── canal
│   │   └── canal.log
│   └── example
│       ├── example.log
│       └── example.log
└── plugin

修改配置文件 conf/example/instance.properties，按如下配置即可，主要是修改數(shù)據(jù)庫(kù)相關(guān)配置：

# 需要同步數(shù)據(jù)的MySQL地址
canal.instance.master.address=127.0.0.1:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=
# 用于同步數(shù)據(jù)的數(shù)據(jù)庫(kù)賬號(hào)
canal.instance.dbUsername=canal
# 用于同步數(shù)據(jù)的數(shù)據(jù)庫(kù)密碼
canal.instance.dbPassword=canal
# 數(shù)據(jù)庫(kù)連接編碼
canal.instance.connectionCharset = UTF-8
# 需要訂閱binlog的表過(guò)濾正則表達(dá)式
canal.instance.filter.regex=.*\\..*

使用 startup.sh 腳本啟動(dòng) canal-server 服務(wù)：

sh bin/startup.sh

啟動(dòng)成功后可使用如下命令查看服務(wù)日志信息：

tail -f logs/canal/canal.log
2020-10-26 16:18:13.354 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[172.17.0.1(172.17.0.1):11111]
2020-10-26 16:18:19.978 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is r

啟動(dòng)成功后可使用如下命令查看 instance 日志信息：

tail -f logs/example/example.log
2020-10-26 16:18:16.056 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2020-10-26 16:18:16.061 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2020-10-26 16:18:18.259 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
2020-10-26 16:18:18.282 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$
2020-10-26 16:18:18.282 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$
2020-10-26 16:18:19.543 [destination = example , address = /127.0.0.1:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
2020-10-26 16:18:19.578 [main] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
2020-10-26 16:18:19.912 [destination = example , address = /127.0.0.1:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just last position
 {"identity":{"slaveId":-1,"sourceAddress":{"address":"localhost","port":3306}},"postion":{"gtid":"","included":false,"journalName":"mall-mysql-bin.000006","position":2271,"serverId":101,"timestamp":1603682664000}}
2020-10-26 16:18:22.435 [destination = example , address = /127.0.0.1:3306 , EventParser] WARN  c.a.o.c.p.inbou

如果想要停止 canal-server 服務(wù)可以使用如下命令：

sh bin/stop.sh

6.4.?canal-adapter 使用

? ? 將我們下載好的壓縮包 canal.adapter-1.1.5-SNAPSHOT.tar.gz 上傳到 Linux 服務(wù)器，然后解壓到指定目錄 /mydata/canal-adpter，解壓完成后目錄結(jié)構(gòu)如下：

├── bin
│   ├── adapter.pid
│   ├── restart.sh
│   ├── startup.bat
│   ├── startup.sh
│   └── stop.sh
├── conf
│   ├── application.yml
│   ├── es6
│   ├── es7
│   │   ├── biz_order.yml
│   │   ├── customer.yml
│   │   └── product.yml
│   ├── hbase
│   ├── kudu
│   ├── logback.xml
│   ├── META-INF
│   │   └── spring.factories
│   └── rdb
├── lib
├── logs
│   └── adapter
│       └── adapter.log
└── plugin

修改配置文件 conf/application.yml，按如下配置即可，主要是修改 canal-server 配置、數(shù)據(jù)源配置和客戶端適配器配置：

canal.conf:
  mode: tcp # 客戶端的模式，可選tcp kafka rocketMQ
  flatMessage: true # 扁平message開(kāi)關(guān), 是否以json字符串形式投遞數(shù)據(jù), 僅在kafka/rocketMQ模式下有效
  zookeeperHosts:    # 對(duì)應(yīng)集群模式下的zk地址
  syncBatchSize: 1000 # 每次同步的批數(shù)量
  retries: 0 # 重試次數(shù), -1為無(wú)限重試
  timeout: # 同步超時(shí)時(shí)間, 單位毫秒
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer
    canal.tcp.server.host: 127.0.0.1:11111 #設(shè)置canal-server的地址
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:

  srcDataSources: # 源數(shù)據(jù)庫(kù)配置
    defaultDS:
      url: jdbc:mysql://127.0.0.1:3306/canal_test?useUnicode=true
      username: canal
      password: canal
  canalAdapters: # 適配器列表
  - instance: example # canal實(shí)例名或者M(jìn)Q topic名
    groups: # 分組列表
    - groupId: g1 # 分組id, 如果是MQ模式將用到該值
      outerAdapters:
      - name: logger # 日志打印適配器
      - name: es7 # ES同步適配器
        hosts: 127.0.0.1:9200 # ES連接地址
        properties:
          mode: rest # 模式可選transport(9300) 或者 rest(9200)
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: elasticsearch # ES集群名稱(chēng)

添加配置文件 canal-adapter/conf/es7/product.yml，用于配置 MySql 中的表與 ElasticSearch 中索引的映射關(guān)系：

dataSourceKey: defaultDS # 源數(shù)據(jù)源的key, 對(duì)應(yīng)上面配置的srcDataSources中的值
destination: example  # canal的instance或者M(jìn)Q的topic
groupId: g1 # 對(duì)應(yīng)MQ模式下的groupId, 只會(huì)同步對(duì)應(yīng)groupId的數(shù)據(jù)
esMapping:
  _index: canal_product # es 的索引名稱(chēng)
  _id: _id  # es 的_id, 如果不配置該項(xiàng)必須配置下面的pk項(xiàng)_id則會(huì)由es自動(dòng)分配
  sql: "SELECT
         p.id AS _id,
         p.title,
         p.sub_title,
         p.price,
         p.pic
        FROM
         product p"        # sql映射
  etlCondition: "where a.c_time>={}"   #etl的條件參數(shù)
  commitBatch: 3000   # 提交批大小

使用 startup.sh 腳本啟動(dòng) canal-adapter 服務(wù)：

sh bin/startup.sh

啟動(dòng)成功后可使用如下命令查看服務(wù)日志信息：

tail -f logs/adapter/adapter.log
20-10-26 16:52:55.148 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: logger succeed
2020-10-26 16:52:57.005 [main] INFO  c.a.o.c.client.adapter.es.core.config.ESSyncConfigLoader - ## Start loading es mapping config ... 
2020-10-26 16:52:57.376 [main] INFO  c.a.o.c.client.adapter.es.core.config.ESSyncConfigLoader - ## ES mapping config loaded
2020-10-26 16:52:58.615 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: es7 succeed
2020-10-26 16:52:58.651 [main] INFO  c.alibaba.otter.canal.connector.core.spi.ExtensionLoader - extension classpath dir: /mydata/canal-adapter/plugin
2020-10-26 16:52:59.043 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Start adapter for canal-client mq topic: example-g1 succeed
2020-10-26 16:52:59.044 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterService - ## the canal client adapters are running now ......
2020-10-26 16:52:59.057 [Thread-4] INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Start to connect destination: example <=============
2020-10-26 16:52:59.100 [main] INFO  org.apache.coyote.http11.Http11NioProtocol - Starting ProtocolHandler ["http-nio-8081"]
2020-10-26 16:52:59.153 [main] INFO  org.apache.tomcat.util.net.NioSelectorPool - Using a shared selector for servlet write/read
2020-10-26 16:52:59.590 [main] INFO  o.s.boot.web.embedded.tomcat.TomcatWebServer - Tomcat started on port(s): 8081 (http) with context path ''
2020-10-26 16:52:59.626 [main] INFO  c.a.otter.canal.adapter.launcher.CanalAdapterApplication - Started CanalAdapterApplication in 31.278 seconds (JVM running for 33.99)
2020-10-26 16:52:59.930 [Thread-4] INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Subscribe destination: example succeed <=============

如果需要停止 canal-adapter 服務(wù)可以使用如下命令：

sh bin/stop.sh

6.5.?canal-admin 使用

? ? 將我們下載好的壓縮包 canal.admin-1.1.5-SNAPSHOT.tar.gz 上傳到 Linux 服務(wù)器，然后解壓到指定目錄 /mydata/canal-admin，解壓完成后目錄結(jié)構(gòu)如下：

├── bin
│   ├── restart.sh
│   ├── startup.bat
│   ├── startup.sh
│   └── stop.sh
├── conf
│   ├── application.yml
│   ├── canal_manager.sql
│   ├── canal-template.properties
│   ├── instance-template.properties
│   ├── logback.xml
│   └── public
│       ├── avatar.gif
│       ├── index.html
│       ├── logo.png
│       └── static
├── lib
└── logs

創(chuàng)建 canal-admin 需要使用的數(shù)據(jù)庫(kù) canal_manager，創(chuàng)建 SQL 腳本為 /mydata/canal-admin/conf/canal_manager.sql，會(huì)創(chuàng)建如下表：

修改配置文件 conf/application.yml，按如下配置即可，主要是修改數(shù)據(jù)源配置和 canal-admin 的管理賬號(hào)配置，注意需要用一個(gè)有讀寫(xiě)權(quán)限的數(shù)據(jù)庫(kù)賬號(hào)，比如管理賬號(hào) root:root：

server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8

spring.datasource:
  address: 127.0.0.1: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

接下來(lái)對(duì)之前搭建的 canal-server 的 conf/canal_local.properties 文件進(jìn)行配置，主要是修改canal-admin 的配置，修改完成后使用 sh bin/startup.sh local 重啟 canal-server：

# register ip
canal.register.ip =

# canal admin config
canal.admin.manager = 127.0.0.1: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 =

使用 startup.sh 腳本啟動(dòng) canal-admin 服務(wù)：

sh bin/startup.sh

啟動(dòng)成功后可使用如下命令查看服務(wù)日志信息：

tail -f logs/admin.log
020-10-27 10:15:04.210 [main] INFO  org.apache.coyote.http11.Http11NioProtocol - Starting ProtocolHandler ["http-nio-8089"]
2020-10-27 10:15:04.308 [main] INFO  org.apache.tomcat.util.net.NioSelectorPool - Using a shared selector for servlet write/read
2020-10-27 10:15:04.534 [main] INFO  o.s.boot.web.embedded.tomcat.TomcatWebServer - Tomcat started on port(s): 8089 (http) with context path ''
2020-10-27 10:15:04.573 [main] INFO  com.alibaba.otter.canal.admin.CanalAdminApplication - Started CanalAdminApplication in 31.203 seconds (JVM running for 34.865)

訪問(wèn) canal-admin 的 Web 界面，輸入賬號(hào)密碼 admin:123456 即可登錄，訪問(wèn)地址：http://192.168.3.101:8089

登錄成功后即可使用 Web 界面操作 canal-server：

6.6.?數(shù)據(jù)同步演示

? ? 經(jīng)過(guò)上面的一系列步驟，Canal 的數(shù)據(jù)同步功能已經(jīng)基本可以使用了，下面我們來(lái)演示下數(shù)據(jù)同步功能。

首先我們需要在 ElasticSearch 中創(chuàng)建索引，和 MySql 中的 product 表相對(duì)應(yīng)，直接在 Kibana 的Dev Tools 中使用如下命令創(chuàng)建即可：

PUT canal_product
{
  "mappings": {
    "properties": {
      "title": {
        "type": "text"
      },
      "sub_title": {
        "type": "text"
      },
      "pic": {
        "type": "text"
      },
      "price": {
        "type": "double"
      }
    }
  }
}

創(chuàng)建完成后可以查看下索引的結(jié)構(gòu)：

之后使用如下 SQL 語(yǔ)句在數(shù)據(jù)庫(kù)中創(chuàng)建一條記錄：

INSERT INTO product ( id, title, sub_title, price, pic ) VALUES ( 5, '小米8', ' 全面屏游戲智能手機(jī) 6GB+64GB', 1999.00, NULL

創(chuàng)建成功后，在 ElasticSearch 中搜索下，發(fā)現(xiàn)數(shù)據(jù)已經(jīng)同步了：

再使用如下 SQL 對(duì)數(shù)據(jù)進(jìn)行修改：

UPDATE product SET title='小米10' WHERE id=5

修改成功后，在 ElasticSearch 中搜索下，發(fā)現(xiàn)數(shù)據(jù)已經(jīng)修改了：

再使用如下 SQL 對(duì)數(shù)據(jù)進(jìn)行刪除操作：

DELETE FROM product WHERE id=5

刪除成功后，在 ElasticSearch 中搜索下，發(fā)現(xiàn)數(shù)據(jù)已經(jīng)刪除了，至此 MySql 同步到 ElasticSearch的功能完成了！

文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-822714.html

到了這里，關(guān)于Canal —— 一款 MySql 實(shí)時(shí)同步到 ES 的阿里開(kāi)源神器的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！