Canal+Kafka實現MySQL與Redis資料同步
思維導圖
前言
在很多業務情況下，我們都會在系統中加入redis快取做查詢優化。
如果資料庫資料發生更新，這時候就需要在業務程式碼中寫一段同步更新redis的程式碼。
這種資料同步的程式碼跟業務程式碼糅合在一起會不太優雅，能不能把這些資料同步的程式碼抽出來形成一個獨立的模組呢，答案是可以的。
架構圖
canal是一個偽裝成slave訂閱mysql的binlog，實現資料同步的中介軟體。上一篇文章《canal入門》
我已經介紹了最簡單的使用方法，也就是tcp模式。
實際上canal是支援直接傳送到MQ的，目前最新版是支援主流的三種MQ：Kafka、RocketMQ、RabbitMQ。而canal的RabbitMQ模式目前是有一定的bug，所以一般使用Kafka或者RocketMQ。
本文使用Kafka，實現Redis與MySQL的資料同步。架構圖如下：
通過架構圖，我們很清晰就知道要用到的元件：MySQL、Canal、Kafka、ZooKeeper、Redis。
下面演示Kafka的搭建，MySQL搭建大家應該都會，ZooKeeper、Redis這些網上也有很多資料參考。
搭建Kafka
首先在官網下載安裝包：
解壓，開啟/config/server.properties配置檔案，修改日誌目錄：
log.dirs=./logs
首先啟動ZooKeeper，我用的是3.6.1版本：
接著再啟動Kafka，在Kafka的bin目錄下開啟cmd，輸入命令：
kafka-server-start.bat../../config/server.properties
我們可以看到ZooKeeper上註冊了Kafka相關的配置資訊：
然後需要建立一個佇列，用於接收canal傳送過來的資料，使用命令：
kafka-topics.bat--create--zookeeperlocalhost:2181--replication-factor1--partitions1--topiccanaltopic
建立的佇列名是canaltopic。
配置Cannal Server
canal官網下載相關安裝包：
找到canal.deployer-1.1.4/conf目錄下的canal.properties配置檔案：
# tcp, kafka, RocketMQ 這裡選擇kafka模式
canal.serverMode = kafka
# 解析器的執行緒數，開啟此配置，不開啟則會出現阻塞或者不進行解析的情況
canal.instance.parser.parallelThreadSize = 16
# 配置MQ的服務地址，這裡配置的是kafka對應的地址和埠
canal.mq.servers = 127.0.0.1:9092
# 配置instance，在conf目錄下要有example同名的目錄，可以配置多個
canal.destinations = example
然後配置instance，找到/conf/example/instance.properties配置檔案：
## mysql serverId , v1.0.26+ will autoGen(自動生成，不需配置)
# canal.instance.mysql.slaveId=0
# position info
canal.instance.master.address=127.0.0.1:3306
# 在Mysql執行 SHOW MASTER STATUS;檢視當前資料庫的binlog
canal.instance.master.journal.name=mysql-bin.000006
canal.instance.master.position=4596
# 賬號密碼
canal.instance.dbUsername=canal
canal.instance.dbPassword=Canal@****
canal.instance.connectionCharset = UTF-8
#MQ佇列名稱
canal.mq.topic=canaltopic
#單佇列模式的分割槽下標
canal.mq.partition=0
配置完成後，就可以啟動canal了。
測試
這時可以開啟kafka的消費者視窗，測試一下kafka是否收到訊息。
使用命令進行監聽消費：
kafka-console-consumer.bat--bootstrap-server127.0.0.1:9092--from-beginning--topiccanaltopic
有個小坑。我這裡使用的是win10系統的cmd命令列，win10系統預設的編碼是GBK，而Canal Server是UTF-8的編碼，所以控制檯會出現亂碼：
怎麼解決呢？
在cmd命令列執行前切換到UTF-8編碼即可，使用命令列：chcp 65001
然後再執行開啟kafka消費端的命令，就不亂碼了：
接下來就是啟動Redis，把資料同步到Redis就完事了。
封裝Redis客戶端
環境搭建完成後，我們可以寫程式碼了。
首先引入Kafka和Redis的maven依賴：
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
在application.yml檔案增加以下配置：
spring:
redis:
host:127.0.0.1
port:6379
database:0
password:123456
封裝一個操作Redis的工具類：
@Component
publicclassRedisClient{
/**
*獲取redis模版
*/
@Resource
privateStringRedisTemplatestringRedisTemplate;
/**
*設定redis的key-value
*/
publicvoidsetString(Stringkey,Stringvalue){
setString(key,value,null);
}
/**
*設定redis的key-value，帶過期時間
*/
publicvoidsetString(Stringkey,Stringvalue,LongtimeOut){
stringRedisTemplate.opsForValue().set(key,value);
if(timeOut!=null){
stringRedisTemplate.expire(key,timeOut,TimeUnit.SECONDS);
}
}
/**
*獲取redis中key對應的值
*/
publicStringgetString(Stringkey){
returnstringRedisTemplate.opsForValue().get(key);
}
/**
*刪除redis中key對應的值
*/
publicBooleandeleteKey(Stringkey){
returnstringRedisTemplate.delete(key);
}
}
建立MQ消費者進行同步
在application.yml配置檔案加上kafka的配置資訊：
spring:
kafka:
#Kafka服務地址
bootstrap-servers:127.0.0.1:9092
consumer:
#指定一個預設的組名
group-id:consumer-group1
#序列化反序列化
key-deserializer:org.apache.kafka.common.serialization.StringDeserializer
value-deserializer:org.apache.kafka.common.serialization.StringDeserializer
producer:
key-serializer:org.apache.kafka.common.serialization.StringDeserializer
value-serializer:org.apache.kafka.common.serialization.StringDeserializer
#批量抓取
batch-size:65536
#快取容量
buffer-memory:524288
根據上面Kafka消費命令那裡，我們知道了json資料的結構，可以建立一個CanalBean物件進行接收：
publicclassCanalBean{
//資料
privateList<TbCommodityInfo>data;
//資料庫名稱
privateStringdatabase;
privatelonges;
//遞增，從1開始
privateintid;
//是否是DDL語句
privatebooleanisDdl;
//表結構的欄位型別
privateMysqlTypemysqlType;
//UPDATE語句，舊資料
privateStringold;
//主鍵名稱
privateList<String>pkNames;
//sql語句
privateStringsql;
privateSqlTypesqlType;
//表名
privateStringtable;
privatelongts;
//(新增)INSERT、(更新)UPDATE、(刪除)DELETE、(刪除表)ERASE等等
privateStringtype;
//getter、setter方法
}
publicclassMysqlType{
privateStringid;
privateStringcommodity_name;
privateStringcommodity_price;
privateStringnumber;
privateStringdescription;
//getter、setter方法
}
publicclassSqlType{
privateintid;
privateintcommodity_name;
privateintcommodity_price;
privateintnumber;
privateintdescription;
}
最後就可以建立一個消費者CanalConsumer進行消費：
@Component
publicclassCanalConsumer{
//日誌記錄
privatestaticLoggerlog=LoggerFactory.getLogger(CanalConsumer.class);
//redis操作工具類
@Resource
privateRedisClientredisClient;
//監聽的佇列名稱為：canaltopic
@KafkaListener(topics="canaltopic")
publicvoidreceive(ConsumerRecord<?,?>consumer){
Stringvalue=(String)consumer.value();
log.info("topic名稱:{},key:{},分割槽位置:{},下標:{},value:{}",consumer.topic(),consumer.key(),consumer.partition(),consumer.offset(),value);
//轉換為javaBean
CanalBeancanalBean=JSONObject.parseObject(value,CanalBean.class);
//獲取是否是DDL語句
booleanisDdl=canalBean.getIsDdl();
//獲取型別
Stringtype=canalBean.getType();
//不是DDL語句
if(!isDdl){
List<TbCommodityInfo>tbCommodityInfos=canalBean.getData();
//過期時間
longTIME_OUT=600L;
if("INSERT".equals(type)){
//新增語句
for(TbCommodityInfotbCommodityInfo:tbCommodityInfos){
Stringid=tbCommodityInfo.getId();
//新增到redis中,過期時間是10分鐘
redisClient.setString(id,JSONObject.toJSONString(tbCommodityInfo),TIME_OUT);
}
}elseif("UPDATE".equals(type)){
//更新語句
for(TbCommodityInfotbCommodityInfo:tbCommodityInfos){
Stringid=tbCommodityInfo.getId();
//更新到redis中,過期時間是10分鐘
redisClient.setString(id,JSONObject.toJSONString(tbCommodityInfo),TIME_OUT);
}
}else{
//刪除語句
for(TbCommodityInfotbCommodityInfo:tbCommodityInfos){
Stringid=tbCommodityInfo.getId();
//從redis中刪除
redisClient.deleteKey(id);
}
}
}
}
}
測試MySQL與Redis同步
mysql對應的表結構如下：
CREATETABLE`tb_commodity_info`(
`id`varchar(32)NOTNULL,
`commodity_name`varchar(512)DEFAULTNULLCOMMENT'商品名稱',
`commodity_price`varchar(36)DEFAULT'0'COMMENT'商品價格',
`number`int(10)DEFAULT'0'COMMENT'商品數量',
`description`varchar(2048)DEFAULT''COMMENT'商品描述',
PRIMARYKEY(`id`)
)ENGINE=InnoDBDEFAULTCHARSET=utf8mb4COMMENT='商品資訊表';
首先在MySQL建立表。然後啟動專案，接著新增一條資料：
INSERTINTO`canaldb`.`tb_commodity_info`(`id`,`commodity_name`,`commodity_price`,`number`,`description`)VALUES('3e71a81fd80711eaaed600163e046cc3','叉燒包','3.99','3','又大又香的叉燒包，老人小孩都喜歡');
tb_commodity_info表查到新增的資料：
Redis也查到了對應的資料，證明同步成功！
如果更新呢？試一下Update語句：
UPDATE`canaldb`.`tb_commodity_info`SET`commodity_name`='青菜包',`description`='很便宜的青菜包呀，不買也開看看了喂'WHERE`id`='3e71a81fd80711eaaed600163e046cc3';
沒有問題！
總結
那麼你會說，canal就沒有什麼缺點嗎？
肯定是有的：
- canal只能同步增量資料。
- 不是實時同步，是準實時同步。
- 存在一些bug，不過社群活躍度較高，對於提出的bug能及時修復。
- MQ順序性問題。我這裡把官網的回答列出來，大家參考一下
儘管有一些缺點，畢竟沒有一樣技術或者產品是完美的，最重要是合適。
我們公司在同步MySQL資料到Elastic Search也是採用Canal+RocketMQ的方式。
參考資料：canal官網
絮叨
上面所有例子的程式碼都上傳Github了：
https://github.com/yehongzhi/mall