思維導圖
前言
在很多業務情況下,我們都會在系統中加入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配置文件,修改日誌目錄:
<code>log.dirs=./logs /<code>
首先啟動ZooKeeper,我用的是3.6.1版本:
接著再啟動Kafka,在Kafka的bin目錄下打開cmd,輸入命令:
<code>kafka-server-start.bat ../../config/server.properties/<code>
我們可以看到ZooKeeper上註冊了Kafka相關的配置信息:
然後需要創建一個隊列,用於接收canal傳送過來的數據,使用命令:
<code>kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic canaltopic/<code>
創建的隊列名是canaltopic。
配置Cannal Server
canal官網下載相關安裝包:
找到canal.deployer-1.1.4/conf目錄下的canal.properties配置文件:
<code># 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 /<code>
然後配置instance,找到/conf/example/instance.properties配置文件:
<code>## 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 /<code>
配置完成後,就可以啟動canal了。
測試
這時可以打開kafka的消費者窗口,測試一下kafka是否收到消息。
使用命令進行監聽消費:
<code>kafka-console-consumer.bat --bootstrap-server 127.0.0.1:9092 --from-beginning --topic canaltopic/<code>
有個小坑。我這裡使用的是win10系統的cmd命令行,win10系統默認的編碼是GBK,而Canal Server是UTF-8的編碼,所以控制檯會出現亂碼:
怎麼解決呢?
在cmd命令行執行前切換到UTF-8編碼即可,使用命令行:chcp 65001
然後再執行打開kafka消費端的命令,就不亂碼了:
接下來就是啟動Redis,把數據同步到Redis就完事了。
封裝Redis客戶端
環境搭建完成後,我們可以寫代碼了。
首先引入Kafka和Redis的maven依賴:
<code> org.springframework.kafka spring-kafka org.springframework.boot spring-boot-starter-data-redis /<code>
在application.yml文件增加以下配置:
<code>spring: redis: host: 127.0.0.1 port: 6379 database: 0 password: 123456/<code>
封裝一個操作Redis的工具類:
<code>@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); } }/<code>
創建MQ消費者進行同步
在application.yml配置文件加上kafka的配置信息:
<code>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/<code>
根據上面Kafka消費命令那裡,我們知道了json數據的結構,可以創建一個CanalBean對象進行接收:
<code>public class CanalBean { //數據 private List data; //數據庫名稱 private String database; private long es; //遞增,從1開始 private int id; //是否是DDL語句 private boolean isDdl; //表結構的字段類型 private MysqlType mysqlType; //UPDATE語句,舊數據 private String old; //主鍵名稱 private List pkNames; //sql語句 private String sql; private SqlType sqlType; //表名 private String table; private long ts; //(新增)INSERT、(更新)UPDATE、(刪除)DELETE、(刪除表)ERASE等等 private String type; //getter、setter方法 }/<code>
<code>public class MysqlType { private String id; private String commodity_name; private String commodity_price; private String number; private String description; //getter、setter方法 }/<code>
<code>public class SqlType { private int id; private int commodity_name; private int commodity_price; private int number; private int description; }/<code>
最後就可以創建一個消費者CanalConsumer進行消費:
<code>@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 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); } } } } }/<code>
測試MySQL與Redis同步
mysql對應的表結構如下:
<code>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='商品信息表';/<code>
首先在MySQL創建表。然後啟動項目,接著新增一條數據:
<code>INSERT INTO `canaldb`.`tb_commodity_info` (`id`, `commodity_name`, `commodity_price`, `number`, `description`) VALUES ('3e71a81fd80711eaaed600163e046cc3', '叉燒包', '3.99', '3', '又大又香的叉燒包,老人小孩都喜歡');/<code>
tb_commodity_info表查到新增的數據:
Redis也查到了對應的數據,證明同步成功!
如果更新呢?試一下Update語句:
<code>UPDATE `canaldb`.`tb_commodity_info` SET `commodity_name`='青菜包',`description`='很便宜的青菜包呀,不買也開看看了喂' WHERE `id`='3e71a81fd80711eaaed600163e046cc3';/<code>
沒有問題!
總結
那麼你會說,canal就沒有什麼缺點嗎?
肯定是有的:
- canal只能同步增量數據。
- 不是實時同步,是準實時同步。
- 存在一些bug,不過社區活躍度較高,對於提出的bug能及時修復。
- MQ順序性問題。我這裡把官網的回答列出來,大家參考一下。
儘管有一些缺點,畢竟沒有一樣技術或者產品是完美的,最重要是合適。
我們公司在同步MySQL數據到Elastic Search也是採用Canal+RocketMQ的方式。
參考資料:canal官網
絮叨
上面所有例子的代碼都上傳Github了:
https://github.com/yehongzhi/mall
如果你覺得這篇文章對你有用,點個贊吧
你的點贊是我創作的最大動力。
想第一時間看到我更新的技術文章,就關注我吧。點個關注,不怕迷路。
拒絕做一條鹹魚,我是一個努力讓大家記住的程序員。我們下期再見!!!