怎么實(shí)現(xiàn)MySQL與Redis數(shù)據(jù)同步

思維導(dǎo)圖

 

前言

在很多業(yè)務(wù)情況下，我們都會(huì)在系統(tǒng)中加入redis緩存做查詢優(yōu)化。

如果數(shù)據(jù)庫(kù)數(shù)據(jù)發(fā)生更新，這時(shí)候就需要在業(yè)務(wù)代碼中寫一段同步更新redis的代碼。

這種數(shù)據(jù)同步的代碼跟業(yè)務(wù)代碼糅合在一起會(huì)不太優(yōu)雅，能不能把這些數(shù)據(jù)同步的代碼抽出來(lái)形成一個(gè)獨(dú)立的模塊呢，答案是可以的。

架構(gòu)圖

canal是一個(gè)偽裝成slave訂閱mysql的binlog，實(shí)現(xiàn)數(shù)據(jù)同步的中間件。上一篇文章《canal入門》

我已經(jīng)介紹了最簡(jiǎn)單的使用方法，也就是tcp模式。

實(shí)際上canal是支持直接發(fā)送到MQ的，目前最新版是支持主流的三種MQ：Kafka、RocketMQ、RabbitMQ。而canal的RabbitMQ模式目前是有一定的bug，所以一般使用Kafka或者RocketMQ。

 

本文使用Kafka，實(shí)現(xiàn)Redis與MySQL的數(shù)據(jù)同步。架構(gòu)圖如下：

 

通過(guò)架構(gòu)圖，我們很清晰就知道要用到的組件：MySQL、Canal、Kafka、ZooKeeper、Redis。

下面演示Kafka的搭建，MySQL搭建大家應(yīng)該都會(huì)，ZooKeeper、Redis這些網(wǎng)上也有很多資料參考。

搭建Kafka

首先在官網(wǎng)下載安裝包：

 

解壓，打開/config/server.properties配置文件，修改日志目錄：

log.dirs=./logs

首先啟動(dòng)ZooKeeper，我用的是3.6.1版本：

 

接著再啟動(dòng)Kafka，在Kafka的bin目錄下打開cmd，輸入命令：

kafka-server-start.bat ../../config/server.properties

我們可以看到ZooKeeper上注冊(cè)了Kafka相關(guān)的配置信息：

 

然后需要?jiǎng)?chuàng)建一個(gè)隊(duì)列，用于接收canal傳送過(guò)來(lái)的數(shù)據(jù)，使用命令：

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic canaltopic

創(chuàng)建的隊(duì)列名是canaltopic。

 

配置Cannal Server

canal官網(wǎng)下載相關(guān)安裝包：

 

找到canal.deployer-1.1.4/conf目錄下的canal.properties配置文件：

# tcp, kafka, RocketMQ 這里選擇kafka模式
canal.serverMode = kafka
# 解析器的線程數(shù)，打開此配置，不打開則會(huì)出現(xiàn)阻塞或者不進(jìn)行解析的情況
canal.instance.parser.parallelThreadSize = 16
# 配置MQ的服務(wù)地址，這里配置的是kafka對(duì)應(yīng)的地址和端口
canal.mq.servers = 127.0.0.1:9092
# 配置instance，在conf目錄下要有example同名的目錄，可以配置多個(gè)
canal.destinations = example

然后配置instance，找到/conf/example/instance.properties配置文件：

## mysql serverId , v1.0.26+ will autoGen(自動(dòng)生成，不需配置)
# canal.instance.mysql.slaveId=0

# position info
canal.instance.master.address=127.0.0.1:3306
# 在Mysql執(zhí)行 SHOW MASTER STATUS;查看當(dāng)前數(shù)據(jù)庫(kù)的binlog
canal.instance.master.journal.name=mysql-bin.000006
canal.instance.master.position=4596
# 賬號(hào)密碼
canal.instance.dbUsername=canal
canal.instance.dbPassword=Canal@****
canal.instance.connectionCharset = UTF-8
#MQ隊(duì)列名稱
canal.mq.topic=canaltopic
#單隊(duì)列模式的分區(qū)下標(biāo)
canal.mq.partition=0

配置完成后，就可以啟動(dòng)canal了。

測(cè)試

這時(shí)可以打開kafka的消費(fèi)者窗口，測(cè)試一下kafka是否收到消息。

使用命令進(jìn)行監(jiān)聽消費(fèi)：

kafka-console-consumer.bat --bootstrap-server 127.0.0.1:9092 --from-beginning --topic canaltopic

有個(gè)小坑。我這里使用的是win10系統(tǒng)的cmd命令行，win10系統(tǒng)默認(rèn)的編碼是GBK，而Canal Server是UTF-8的編碼，所以控制臺(tái)會(huì)出現(xiàn)亂碼：

 

怎么解決呢？

在cmd命令行執(zhí)行前切換到UTF-8編碼即可，使用命令行：chcp 65001

然后再執(zhí)行打開kafka消費(fèi)端的命令，就不亂碼了：

 

接下來(lái)就是啟動(dòng)Redis，把數(shù)據(jù)同步到Redis就完事了。

封裝Redis客戶端

環(huán)境搭建完成后，我們可以寫代碼了。

首先引入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

封裝一個(gè)操作Redis的工具類：

@Component
public class RedisClient {

    /**
     * 獲取redis模版
     */
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 設(shè)置redis的key-value
     */
    public void setString(String key, String value) {
        setString(key, value, null);
    }

    /**
     * 設(shè)置redis的key-value，帶過(guò)期時(shí)間
     */
    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對(duì)應(yīng)的值
     */
    public String getString(String key) {
        return stringRedisTemplate.opsForValue().get(key);
    }

    /**
     * 刪除redis中key對(duì)應(yīng)的值
     */
    public Boolean deleteKey(String key) {
        return stringRedisTemplate.delete(key);
    }
}

創(chuàng)建MQ消費(fèi)者進(jìn)行同步

在application.yml配置文件加上kafka的配置信息：

spring:
  kafka:
   # Kafka服務(wù)地址
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      # 指定一個(gè)默認(rèn)的組名
      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

根據(jù)上面Kafka消費(fèi)命令那里，我們知道了json數(shù)據(jù)的結(jié)構(gòu)，可以創(chuàng)建一個(gè)CanalBean對(duì)象進(jìn)行接收：

public class CanalBean {
    //數(shù)據(jù)
    private List<TbCommodityInfo> data;
    //數(shù)據(jù)庫(kù)名稱
    private String database;
    private long es;
    //遞增，從1開始
    private int id;
    //是否是DDL語(yǔ)句
    private boolean isDdl;
    //表結(jié)構(gòu)的字段類型
    private MysqlType mysqlType;
    //UPDATE語(yǔ)句，舊數(shù)據(jù)
    private String old;
    //主鍵名稱
    private List<String> pkNames;
    //sql語(yǔ)句
    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;
}

最后就可以創(chuàng)建一個(gè)消費(fèi)者CanalConsumer進(jìn)行消費(fèi)：

@Component
public class CanalConsumer {
 //日志記錄
    private static Logger log = LoggerFactory.getLogger(CanalConsumer.class);
 //redis操作工具類
    @Resource
    private RedisClient redisClient;
 //監(jiān)聽的隊(duì)列名稱為：canaltopic
    @KafkaListener(topics = "canaltopic")
    public void receive(ConsumerRecord<?, ?> consumer) {
        String value = (String) consumer.value();
        log.info("topic名稱:{},key:{},分區(qū)位置:{},下標(biāo):{},value:{}", consumer.topic(), consumer.key(),consumer.partition(), consumer.offset(), value);
        //轉(zhuǎn)換為javaBean
        CanalBean canalBean = JSONObject.parseObject(value, CanalBean.class);
        //獲取是否是DDL語(yǔ)句
        boolean isDdl = canalBean.getIsDdl();
        //獲取類型
        String type = canalBean.getType();
        //不是DDL語(yǔ)句
        if (!isDdl) {
            List<TbCommodityInfo> tbCommodityInfos = canalBean.getData();
            //過(guò)期時(shí)間
            long TIME_OUT = 600L;
            if ("INSERT".equals(type)) {
                //新增語(yǔ)句
                for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
                    String id = tbCommodityInfo.getId();
                    //新增到redis中,過(guò)期時(shí)間是10分鐘
                    redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);
                }
            } else if ("UPDATE".equals(type)) {
                //更新語(yǔ)句
                for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
                    String id = tbCommodityInfo.getId();
                    //更新到redis中,過(guò)期時(shí)間是10分鐘
                    redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);
                }
            } else {
                //刪除語(yǔ)句
                for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
                    String id = tbCommodityInfo.getId();
                    //從redis中刪除
                    redisClient.deleteKey(id);
                }
            }
        }
    }
}

測(cè)試MySQL與Redis同步

mysql對(duì)應(yīng)的表結(jié)構(gòu)如下：

CREATE TABLE `tb_commodity_info` (
  `id` varchar(32) NOT NULL,
  `commodity_name` varchar(512) DEFAULT NULL COMMENT '商品名稱',
  `commodity_price` varchar(36) DEFAULT '0' COMMENT '商品價(jià)格',
  `number` int(10) DEFAULT '0' COMMENT '商品數(shù)量',
  `description` varchar(2048) DEFAULT '' COMMENT '商品描述',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品信息表';

首先在MySQL創(chuàng)建表。然后啟動(dòng)項(xiàng)目，接著新增一條數(shù)據(jù)：

INSERT INTO `canaldb`.`tb_commodity_info` (`id`, `commodity_name`, `commodity_price`, `number`, `description`) VALUES ('3e71a81fd80711eaaed600163e046cc3', '叉燒包', '3.99', '3', '又大又香的叉燒包，老人小孩都喜歡');

tb_commodity_info表查到新增的數(shù)據(jù)：

 

Redis也查到了對(duì)應(yīng)的數(shù)據(jù)，證明同步成功！

 

如果更新呢？試一下Update語(yǔ)句：

UPDATE `canaldb`.`tb_commodity_info` SET `commodity_name`='青菜包',`description`='很便宜的青菜包呀，不買也開看看了喂' WHERE `id`='3e71a81fd80711eaaed600163e046cc3';

 

 

沒(méi)有問(wèn)題！

總結(jié)

那么你會(huì)說(shuō)，canal就沒(méi)有什么缺點(diǎn)嗎？

肯定是有的：

1. canal只能同步增量數(shù)據(jù)。
2. 不是實(shí)時(shí)同步，是準(zhǔn)實(shí)時(shí)同步。
3. 存在一些bug，不過(guò)社區(qū)活躍度較高，對(duì)于提出的bug能及時(shí)修復(fù)。
4. MQ順序性問(wèn)題。我這里把官網(wǎng)的回答列出來(lái)，大家參考一下。