日日操夜夜添-日日操影院-日日草夜夜操-日日干干-精品一区二区三区波多野结衣-精品一区二区三区高清免费不卡

前言

上篇文章簡單介紹canal概念，本文結合常見的緩存業務去講解canal使用。在實際開發過程中，通常都會把數據往redis緩存中保存一份，做下簡單的查詢優化。如果這時候數據庫數據發生變更操作，就不得不在業務代碼中寫一段同步更新redis的代碼，但是這種 數據同步的代碼和業務代碼糅合在一起 看起來不是很優雅，而且還會出現數據不一致問題。那能不能把這部分同步代碼從中抽離出來，形成獨立模塊呢？答案是肯定的，下面通過canal結合Kafka來實現MySQL與redis之間的數據同步。

架構設計

canal是一個偽裝成slave訂閱mysql的binlog，實現數據同步的中間件。上一篇文章 canal入門 中簡單介紹了使用方式，即tcp模式；其實canal也是支持直接發送到MQ中，比如：Kafka、RocketMQ、RabbitMQ。本文采用Kafka講解，實現mysql與redis之間的數據同步。

通過上述結構設計圖可以很清晰的知道用到的組件：MySQL、Canal、Kafka、ZooKeeper、Redis。

Kafka&Zookeeper搭建

首先在官網下載Kafka：

下載后解壓文件夾，可以看到以下幾個文件：

Kafka內部自帶了zookeeper，所以暫不需要去下載搭建zookeeper集群，本文就使用Kafka自帶zookeeper來實現。

通過上述zookeeper啟動命令以及Kafka啟動命令把服務啟動，可以通過以下簡單實現下是否成功：

# 命令常見一個canaltopic 隊列
kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic canaltopic

Canal搭建

canal搭建具體可以參考上文，這里只講解具體的參數配置：

找到/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了。

測試

環境搭建完成后，就可以編寫代碼進行測試。

1、引入pom依賴

<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>

2、封裝Redis工具類

在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);
    }
}

3、創建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

創建一個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進行消費：

@Slf4j
@Component
public class CanalConsumer {


    @Resource
    private RedisClient redisClient;


    @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.hasDdl();
        //獲取類型
        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();
                    log.info("新增數據到redis, id: {}, data: {}", id, JSONObject.toJSONString(tbCommodityInfo));
                    //新增到redis中,過期時間是10分鐘
                    redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);
                    log.info("從redis獲取數據 result: {}", JSONObject.toJSONString(redisClient.getString(id)));
                }
            } else if ("UPDATE".equals(type)) {
                //更新語句
                for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
                    String id = tbCommodityInfo.getId();
                    log.info("修改數據到redis, id: {}, data: {}", id, JSONObject.toJSONString(tbCommodityInfo));
                    //更新到redis中,過期時間是10分鐘
                    redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);
                }
            } else {
                //刪除語句
                for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
                    String id = tbCommodityInfo.getId();
                    log.info("刪除數據從redis, id: {}", id);
                    //從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='商品信息表';

啟動項目后，新增一條數據：

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

可以在控制臺看到以下輸出：

2022-01-02 18:12:51.317  INFO 55608 --- [ntainer#0-0-C-1] c.kingoe.canaldj.mqcanal.CanalConsumer   : 新增數據到redis, id: 3e71a81fd80711eaaed600163e046cc3, data: {"commodity_name":"叉燒包","commodity_price":"3.99","description":"又大又香的叉燒包，老人小孩都喜歡","id":"3e71a81fd80711eaaed600163e046cc3","number":"3"}
2022-01-02 18:12:51.320  INFO 55608 --- [ntainer#0-0-C-1] c.kingoe.canaldj.mqcanal.CanalConsumer   : 從redis獲取數據 result: "{"commodity_name":"叉燒包","commodity_price":"3.99","description":"又大又香的叉燒包，老人小孩都喜歡","id":"3e71a81fd80711eaaed600163e046cc3","number":"3"}"

如果更新呢？試一下Update語句：

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

同樣可以在控制臺看到以下輸出：

2022-01-02 18:14:44.613  INFO 55608 --- [ntainer#0-0-C-1] c.kingoe.canaldj.mqcanal.CanalConsumer   : topic名稱:canaltopic,key:null,分區位置:0,下標:6,value:{"data":[{"id":"3e71a81fd80711eaaed600163e046cc3","commodity_name":"青菜包","commodity_price":"3.99","number":"3","description":"很便宜的青菜包呀，不買也開看看了喂"}],"database":"study","es":1641118484000,"id":7,"isDdl":false,"mysqlType":{"id":"varchar(32)","commodity_name":"varchar(512)","commodity_price":"varchar(36)","number":"int(10)","description":"varchar(2048)"},"old":[{"commodity_name":"叉燒包","description":"又大又香的叉燒包，老人小孩都喜歡"}],"pkNames":["id"],"sql":"","sqlType":{"id":12,"commodity_name":12,"commodity_price":12,"number":4,"description":12},"table":"tb_commodity_info","ts":1641118484602,"type":"UPDATE"}
2022-01-02 18:14:44.616  INFO 55608 --- [ntainer#0-0-C-1] c.kingoe.canaldj.mqcanal.CanalConsumer   : 修改數據到redis, id: 3e71a81fd80711eaaed600163e046cc3, data: {"commodity_name":"青菜包","commodity_price":"3.99","description":"很便宜的青菜包呀，不買也開看看了喂","id":"3e71a81fd80711eaaed600163e046cc3","number":"3"}

經過測試完全么有問題。

總結

既然canal這么強大，難道就沒缺點嘛？答案當然是存在的啦，比如：canal只能同步增量數據、不是實時同步而是準實時同步、MQ順序問題等；盡管有一些缺點，畢竟沒有一樣技術或者產品是完美的，最重要是合適。比如公司目前有個視圖服務提供寬表搜索查詢功能就是通過 同步Mysql數據到Es采用Canal+Kafka的方式來實現的。

如果你覺得這篇文章對你有用，點個贊吧~ 你的點贊是我創作的最大動力~想第一時間看到我更新的文章，可以微信搜索公眾號「CodingCode」。