思維導圖
本文章已收錄到個人部落格網站(我愛B站)：me.lovebilibili.com
前言
在很多業務情況下，我們都會在系統中加入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 --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic canaltopic 建立的隊列名是
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-server 127.0.0.1:9092 --from-beginning --topic canaltopic 有個小坑。我這裡使用的是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
public class RedisClient {
/**
* 擷取redis模版
*/
@Resource
private StringRedisTemplate stringRedisTemplate;
/**
* 設定redis的key-value
*/
public void setString(String key, String value) {
setString(key, value, null);
}
/**
* 設定redis的key-value，帶過期時間
*/
public void setString(String key, String value, Long timeOut) {
stringRedisTemplate.opsForValue().set(key, value);
if (timeOut != null) {
stringRedisTemplate.expire(key, timeOut, TimeUnit.SECONDS);
}
}
/**
* 擷取redis中key對應的值
*/
public String getString(String key) {
return stringRedisTemplate.opsForValue().get(key);
}
/**
* 删除redis中key對應的值
*/
public Boolean deleteKey(String key) {
return stringRedisTemplate.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對象進行接收：
public class CanalBean {
//資料
private List<TbCommodityInfo> data;
//資料庫名稱
private String database;
private long es;
//遞增，從1開始
private int id;
//是否是DDL語句
private boolean isDdl;
//表結構的字段類型
private MysqlType mysqlType;
//UPDATE語句，舊資料
private String old;
//主鍵名稱
private List<String> pkNames;
//sql語句
private String sql;
private SqlType sqlType;
//表名
private String table;
private long ts;
//(新增)INSERT、(更新)UPDATE、(删除)DELETE、(删除表)ERASE等等
private String type;
//getter、setter方法
} public class MysqlType {
private String id;
private String commodity_name;
private String commodity_price;
private String number;
private String description;
//getter、setter方法
} public class SqlType {
private int id;
private int commodity_name;
private int commodity_price;
private int number;
private int description;
} 最後就可以建立一個消費者CanalConsumer進行消費：
@Component
public class CanalConsumer {
//日志記錄
private static Logger log = LoggerFactory.getLogger(CanalConsumer.class);
//redis操作工具類
@Resource
private RedisClient redisClient;
//監聽的隊列名稱為：canaltopic
@KafkaListener(topics = "canaltopic")
public void receive(ConsumerRecord<?, ?> consumer) {
String value = (String) consumer.value();
log.info("topic名稱:{},key:{},分區位置:{},下标:{},value:{}", consumer.topic(), consumer.key(),consumer.partition(), consumer.offset(), value);
//轉換為javaBean
CanalBean canalBean = JSONObject.parseObject(value, CanalBean.class);
//擷取是否是DDL語句
boolean isDdl = canalBean.getIsDdl();
//擷取類型
String type = canalBean.getType();
//不是DDL語句
if (!isDdl) {
List<TbCommodityInfo> tbCommodityInfos = canalBean.getData();
//過期時間
long TIME_OUT = 600L;
if ("INSERT".equals(type)) {
//新增語句
for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
String id = tbCommodityInfo.getId();
//新增到redis中,過期時間是10分鐘
redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);
}
} else if ("UPDATE".equals(type)) {
//更新語句
for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
String id = tbCommodityInfo.getId();
//更新到redis中,過期時間是10分鐘
redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);
}
} else {
//删除語句
for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
String id = tbCommodityInfo.getId();
//從redis中删除
redisClient.deleteKey(id);
}
}
}
}
} 測試MySQL與Redis同步
mysql對應的表結構如下：
CREATE TABLE `tb_commodity_info` (
`id` varchar(32) NOT NULL,
`commodity_name` varchar(512) DEFAULT NULL COMMENT '商品名稱',
`commodity_price` varchar(36) DEFAULT '0' COMMENT '商品價格',
`number` int(10) DEFAULT '0' COMMENT '商品數量',
`description` varchar(2048) DEFAULT '' COMMENT '商品描述',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品資訊表'; 首先在MySQL建立表。然後啟動項目，接着新增一條資料：
INSERT INTO `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
如果你覺得這篇文章對你有用，點個贊吧~
你的點贊是我創作的最大動力~
想第一時間看到我更新的文章，可以微信搜尋公衆号「
java技術愛好者
」。
拒絕做一條鹹魚，我是一個努力讓大家記住的程式員。我們下期再見！！！