Canal 的 dump 支持串行和并行模式兩種模式，本篇重點(diǎn)梳理 dump 的核心流程，以便對(duì) dump 過(guò)程有一個(gè)充分的了解，更好的理解 Canal 的實(shí)現(xiàn)原理與細(xì)節(jié)，下一篇中將重點(diǎn)關(guān)注Canal是如何引入并行模式來(lái)提高dump的性能，即并行編程相關(guān)的技巧。

從前面的文章我們得知 Canal binlog 日志解析的基本流程如下圖所示：

 

1、 dump 流程分析

在 Canal 中 dump 方法聲明如下：

 

 

• String binlogfilename
    binlog 文件名稱(chēng)，例如  MySQL-bin.000038。

• Long binlogPosition
  在文件中的偏移量。

• SinkFunction func
  每解析出一條binlog日志的處理函數(shù)。

接下來(lái)我們直奔主題，一起來(lái)看一下 MysqlConnection 關(guān)于 dump 的實(shí)現(xiàn)流程。

 

• set wait_timeout=9999999
  連接空閑超時(shí)時(shí)間，默認(rèn)為8消息，用于 Canal Slave 的等待超時(shí)時(shí)間遠(yuǎn)大于默認(rèn)值。

• set net_write_timeout=1800
  網(wǎng)絡(luò)寫(xiě)請(qǐng)求超時(shí)時(shí)間，針對(duì)正在進(jìn)行數(shù)據(jù)讀寫(xiě)的連接，該值默認(rèn)為 60s。

• set net_read_timeout=1800
  網(wǎng)絡(luò)讀請(qǐng)求超時(shí)時(shí)間，針對(duì)正在進(jìn)行數(shù)據(jù)讀寫(xiě)的連接，該值默認(rèn)為 30s。

• set names 'binary'
  設(shè)置服務(wù)端返回結(jié)果時(shí)不做編碼轉(zhuǎn)化，直接按照數(shù)據(jù)庫(kù)的二進(jìn)制編碼進(jìn)行發(fā)送，由客戶(hù)端自己根據(jù)需求進(jìn)行編碼轉(zhuǎn)化。

• set @master_binlog_checksum= @@global.binlog_checksum
  設(shè)置master_binlog_checksum，因?yàn)樵趍ysql5.6之后為binlog引入了checksum機(jī)制，例如crc32，canal作為mysql slave，需要與服務(wù)端相關(guān)參數(shù)保持一致。

• set @slave_uuid=uuid()
  canal相對(duì)與mysql數(shù)據(jù)庫(kù)服務(wù)而言就是一個(gè)從服務(wù)器，這個(gè)指令用于設(shè)置server_id，使用uuid，避免server_id重復(fù)。

• SET @master_heartbeat_period=15
  設(shè)置客戶(hù)端與服務(wù)端心跳發(fā)送間隔，默認(rèn)為15s。

     

  
  Step2：從主庫(kù)查詢(xún)binlog checksum，具體向主庫(kù)發(fā)送 select @@global.binlog_checksum 語(yǔ)句。

     

  
  Step3：向MySQL Master 注冊(cè)從節(jié)點(diǎn)，告知客戶(hù)端的host、port、用戶(hù)名與密碼、serverId，具體實(shí)現(xiàn)是發(fā)送命令CODE為 0x15。

     

  
  Step4：向 MySQL Master 發(fā)送 dump 請(qǐng)求，MySQL是基于請(qǐng)求與應(yīng)答模式，發(fā)送請(qǐng)求命令后，就會(huì)向網(wǎng)絡(luò)通道中寫(xiě)入響應(yīng)請(qǐng)求。（在這里大家不妨先大概思考一下如何讀取 dump 命令的返回值，這部分雖然涉及到網(wǎng)絡(luò)相關(guān)的知識(shí)，我在這邊會(huì)稍微簡(jiǎn)單提一下）。

     

  
  Step5：構(gòu)建 DirectLogFetcher對(duì)象，實(shí)現(xiàn)基于 socket 的日志拉取服務(wù)，并構(gòu)建 LogDecoder 對(duì)象，用于解析 binlog 日志。

     

  
  Step6：使用 while 循環(huán)反復(fù)拉取消息，通過(guò)通過(guò) LogDecoder 對(duì)二進(jìn)制流進(jìn)行解析，提取一條完整的binlog事件，交給 SinkFunction 去處理，并且如果開(kāi)啟了半同步機(jī)制，則需要向master發(fā)送ACK。既然是while循環(huán)，該方法的退出條件還是值得我們關(guān)注的：

• fetch.fetch()方法返回 false

• SinkFunction 的 sink 方法 false，SinkFunction的詳細(xì)處理流程將在下文介紹，這里先告知返回false的情況是 binlog 日志解析線程已停止運(yùn)行。

上面粗略的介紹了 dump 命令的幾個(gè)核心關(guān)鍵步驟，要想詳細(xì)掌握其實(shí)現(xiàn)細(xì)節(jié)，我們必須繼續(xù)深入探討如下幾個(gè)問(wèn)題：

• DirectLogFetcher 內(nèi)部工作機(jī)制

• LogDecoder binlog 日志解析

• 發(fā)送Dump底層網(wǎng)實(shí)現(xiàn)思路

2、DirectLogFetcher 內(nèi)部工作機(jī)制

2.1 DirectLogFetcher 類(lèi)圖

 

• LogBuffer
  日志buffer，主要定義如下屬性：

• byte[] buffer
  緩存區(qū)中數(shù)據(jù)容器。

• int origin
  當(dāng)前buffer中的讀指針

• int limit
  當(dāng)前buffer的最大可讀可寫(xiě)指針

• int position
  當(dāng)前buffer的寫(xiě)指針。

• int semival
  是否需要發(fā)送ACK(用于半同步)。
  LogBuffer封裝了字節(jié)相關(guān)的操作，不僅定義了上面的屬性，也定義了字節(jié)讀取相關(guān)眾多API，其截圖如下：

     

• LogFetcher binlog日志抓取抽象類(lèi)，定義了如下關(guān)鍵屬性與抽象方法。

• int DEFAULT_INITIAL_CAPACITY
  LogBuffer中的初始容量，默認(rèn)為8K。

• float DEFAULT_GROWTH_FACTOR
  容量增長(zhǎng)因子，默認(rèn)為 2.0。

• int   BIN_LOG_HEADER_SIZE
  binlog日志條目 header 的長(zhǎng)度，固定為4字節(jié)。

• float  factor
  增長(zhǎng)因子。

• public abstract boolean fetch()
  抓取binlog日志。

• public abstract void close()
       關(guān)閉 Fetch。

• DirectLogFetcher Canal LogFetcher模式實(shí)現(xiàn)類(lèi)，其核心屬性如下：

• SocketChannel channel
  網(wǎng)絡(luò)通道，用于發(fā)送dump請(qǐng)求的網(wǎng)絡(luò)通道。

• boolean issemi = false
  是否開(kāi)啟半同步。

2.2 fetch流程詳解

接下來(lái)我們重點(diǎn)剖析 DirectLogFetcher 的 fetch 方法，來(lái)探究其實(shí)現(xiàn)原理。
在研究DirectLogFetcher的fetch方法之前，我們先重點(diǎn)跟蹤一下其內(nèi)部網(wǎng)絡(luò)讀寫(xiě)方法fetch0方法，該方法是具體與網(wǎng)絡(luò)讀寫(xiě)相關(guān)的實(shí)現(xiàn)。

 

• int off
  從通道中讀取到的內(nèi)容放入到buffer中的起始位置

• int len
  期望從通道中讀取的字節(jié)長(zhǎng)度。

該方法的實(shí)現(xiàn)關(guān)鍵點(diǎn)如下：

• 首先先確保接收緩存區(qū)有足夠的剩余空間，如果空間不足，則進(jìn)行擴(kuò)容。

• 然后從通道中讀取指定長(zhǎng)度的字節(jié)。

接下來(lái)我們來(lái)重點(diǎn)看一下DirectLogFetcher的fetch的實(shí)現(xiàn)流程。

 

 

 

 

 

從上面的流程來(lái)看，DirectLogFetcher#fetch 方法結(jié)束后，就將進(jìn)入到LogDecoder中。經(jīng)過(guò)一次DirectLogFetcher#fetch方法后，即取回一條binlog日志，即二進(jìn)制流，接下來(lái)就根據(jù)binlog協(xié)議對(duì)其解析。本文暫不深入該方法，如果大家想深入數(shù)據(jù)庫(kù)中間件方面，可以作為一個(gè)很好的示例，面向MySQL通信協(xié)議進(jìn)行編程。

３、SinkFunction

通過(guò) LogDecoder從中解析一個(gè)事件后，會(huì)調(diào)用SinkFunction的sink方法，如果該方法返回 false，一次dump請(qǐng)求將介紹，接下來(lái)我們看一下其sink方法。

 

 

 

• 首先需要調(diào)用EventSink組件將解析出來(lái)的數(shù)據(jù)傳入EventSink。

• EventSink組件處理成功后，會(huì)提交解析位點(diǎn)。