怎么使用canal+Kafka進行數據庫同步操作

這篇文章主要介紹了怎么使用canal+Kafka進行數據庫同步操作的相關知識，內容詳細易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇怎么使用canal+Kafka進行數據庫同步操作文章都會有所收獲，下面我們一起來看看吧。

平時工作中數據庫是我們經常使用的，在微服務拆分的架構中，各服務擁有自己的數據庫，所以常常會遇到服務之間數據通信的問題。比如，B 服務數據庫的數據來源于A服務的數據庫；A 服務的數據有變更操作時，需要同步到 B 服務中。

第一種解決方案:

在代碼邏輯中，有相關 A 服務數據寫操作時，以調用接口的方式，調用 B 服務接口，B 服務再將數據寫到新的數據庫中。這種方式看似簡單，但其實“坑”很多。在 A 服務代碼邏輯中會增加大量這種調用接口同步的代碼，增加了項目代碼的復雜度，以后會越來越難維護。并且，接口調用的方式并不是一個穩定的方式，沒有重試機制，沒有同步位置記錄，接口調用失敗了怎么處理，突然的大量接口調用會產生的問題等，這些都要考慮并且在業務中處理。這里會有不少工作量。想到這里，就將這個方案排除了。

第二種解決方案：

通過數據庫的binlog進行同步。這種解決方案，與 A 服務是獨立的，不會和 A 服務有代碼上的耦合。可以直接 TCP連接進行傳輸數據，優于接口調用的方式。 這是一套成熟的生產解決方案，也有不少binlog同步的中間件工具，所以我們關注的就是哪個工具能夠更好的構建穩定、性能滿足且易于高可用部署的方案。

經過調研，我們選擇了canal。canal是阿里巴巴 MySQL binlog 增量訂閱&消費組件，已經有在生產上實踐的例子，并且方便的支持和其他常用的中間件組件組合，比如kafka，elasticsearch等，也有了canal-go go語言的client庫，滿足我們在go上的需求。

工作流程

1. Canal連接到 A 數據庫，模擬slave

2. canal-client與Canal建立連接，并訂閱對應的數據庫表

3. A 數據庫發生變更寫入到binlog，Canal向數據庫發送dump請求，獲取binlog并解析，發送解析后的數據給canal-client

4. canal-client收到數據，將數據同步到新的數據庫

Protocol Buffer的序列化速度還是很快的。反序列化后得到的數據，是每一行的數據，按照字段名和字段的值的結構，放到一個數組中 代碼簡單示例:

func Handler(entry protocol.Entry)  { var keys []string rowChange := &protocol.RowChange{} proto.Unmarshal(entry.GetStoreValue(), rowChange) if rowChange != nil { eventType := rowChange.GetEventType() for _, rowData := range rowChange.GetRowDatas() { // 遍歷每一行數據 if eventType == protocol.EventType_DELETE || eventType == protocol.EventType_UPDATE { columns := rowData.GetBeforeColumns() // 得到更改前的所有字段屬性 } else if eventType == protocol.EventType_INSERT { columns := rowData.GetAfterColumns() // 得到更后前的所有字段屬性 } ...... } } }

遇到的問題

為了高可用和更高的性能，我們會創建多個canal-client構成一個集群，來進行解析并同步到新的數據庫。這里就出現了一個比較重要的問題，如何保證canal-client集群解析消費binlog的順序性呢？

我們使用的binlog是row模式。每一個寫操作都會產生一條binlog日志。 舉個簡單的例子：插入了一條 a 記錄，并且立馬修改 a 記錄。這樣會有兩個消息發送給canal-client，如果由于網絡等原因，更新的消息早于插入的消息被處理了，還沒有插入記錄，更新操作的最后效果是失敗的。

怎么辦呢？ canal可以和消息隊列組合呀!而且支持kafka，rabbitmq，rocketmq多種選擇，如此優秀。我們在消息隊列這層來實現消息的順序性。

選擇canal+kafka方案

我們選擇了消息隊列的業界標桿: kafka UCloud提供了kafka和rocketMQ消息隊列產品服務，使用它們能夠快速便捷的搭建起一套消息隊列系統。加速開發，方便運維。

下面就讓我們來一探究竟：

1.選擇kafka消息隊列產品，并申請開通


2.開通完成后，在管理界面，創建kafka集群，根據自身需求，選擇相應的硬件配置


3.一個kafka + ZooKeeper集群就搭建出來了，給力！


并且包含了節點管理、Topic管理、Consumer Group管理，能夠非常方便的直接在控制臺對配置進行修改

監控視圖方面，監控的數據包括kafka生成和消費QPS，集群監控，ZooKeeper的監控。能夠比較完善的提供監控指標。




canal的kafka配置

canal配上kafka也非常的簡單。 vi /usr/local/canal/conf/canal.properties

# 可選項: tcp(默認), kafka, RocketMQ
canal.serverMode = kafka
# ...
# kafka/rocketmq 集群配置: 192.168.1.117:9092,192.168.1.118:9092,192.168.1.119:9092
canal.mq.servers = 127.0.0.1:9002
canal.mq.retries = 0
# flagMessage模式下可以調大該值, 但不要超過MQ消息體大小上限
canal.mq.batchSize = 16384
canal.mq.maxRequestSize = 1048576
# flatMessage模式下請將該值改大, 建議50-200
canal.mq.lingerMs = 1
canal.mq.bufferMemory = 33554432
# Canal的batch size, 默認50K, 由于kafka最大消息體限制請勿超過1M(900K以下)
canal.mq.canalBatchSize = 50
# Canal get數據的超時時間, 單位: 毫秒, 空為不限超時
canal.mq.canalGetTimeout = 100
# 是否為flat json格式對象
canal.mq.flatMessage = false
canal.mq.compressionType = none
canal.mq.acks = all
# kafka消息投遞是否使用事務
canal.mq.transaction = false


# mq config
canal.mq.topic=default
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\\\..*,.*\\\\..*
canal.mq.dynamicTopic=mydatabase.mytable
canal.mq.partition=0
# hash partition config
canal.mq.partitionsNum=3
canal.mq.partitionHash=mydatabase.mytable

解決順序消費問題

看到下面這一行配置

canal.mq.partitionHash=mydatabase.mytable

我們配置了kafka的partitionHash，并且我們一個Topic就是一個表。這樣的效果就是，一個表的數據只會推到一個固定的partition中，然后再推給consumer進行消費處理，同步到新的數據庫。通過這種方式，解決了之前碰到的binlog日志順序處理的問題。這樣即使我們部署了多個kafka consumer端，構成一個集群，這樣consumer從一個partition消費消息，就是消費處理同一個表的數據。這樣對于一個表來說，犧牲掉了并行處理，不過個人覺得，憑借kafka的性能強大的處理架構，我們的業務在kafka這個節點產生瓶頸并不容易。并且我們的業務目的不是實時一致性，在一定延遲下，兩個數據庫保證最終一致性。

(推薦微課：SQL微課)

下圖是最終的同步架構，我們在每一個服務節點都實現了集群化。全都跑在UCloud的UK8s服務上，保證了服務節點的高可用性。

canal也是集群換，但是某一時刻只會有一臺canal在處理binlog，其他都是冗余服務。當這臺canal服務掛了，其中一臺冗余服務就會切換到工作狀態。同樣的，也是因為要保證binlog的順序讀取，所以只能有一臺canal在工作。

并且，我們還用這套架構進行緩存失效的同步。我們使用的緩存模式是:Cache-Aside。同樣的，如果在代碼中數據更改的地方進行緩存失效操作，會將代碼變得復雜。所以，在上述架構的基礎上，將復雜的觸發緩存失效的邏輯放到kafka-client端統一處理，達到一定解耦的目的。

關于“怎么使用canal+Kafka進行數據庫同步操作”這篇文章的內容就介紹到這里，感謝各位的閱讀！相信大家對“怎么使用canal+Kafka進行數據庫同步操作”知識都有一定的了解，大家如果還想學習更多知識，歡迎關注億速云行業資訊頻道。