Canal如何同步缓存，Canal同步mysql和redis中缓存——多级缓存学习

秒速五厘米 2024-03-24 15:32 103阅读 0赞

### 1、Canal介绍 ###

canal github地址： [GitHub - alibaba/canal: 阿里巴巴 MySQL binlog 增量订阅&消费组件][GitHub - alibaba_canal_ _ MySQL binlog]

**Canal** 是用 Java 开发的基于数据库增量日志解析，提供增量数据订阅&消费的中间件。  
目前，Canal 主要支持了 MySQL 的 Binlog 解析，解析完成后才利用 Canal Client 来处理获得  
的相关数据。canal 就是一个**同步增量数据的一个工具**。  
当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x。

### 2、工作原理 ###

**MySQL主备复制原理**

*  MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
 *  MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
 *  MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

**canal 工作原理**

*  canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
 *  MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
 *  canal 解析 binary log 对象(原始为 byte 流)

![bf035abc98e64711b677df2d12e55785.png][]

(1）MySQL master 将数据变更写入二进制日志( **binary log**），其中记录的数据叫做binary log events  
(2）MySQL slave 将 master 的 binary log events拷贝到它的中继日志(**relay log**)  
(3）MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

而Canal就是把自己伪装成MySQL的一个slave节点，从而监听master的binary log变化。再把得到的变化信息通知给Canal的客户端，进而完成对其它数据库的同步。

![a9cdb3f5f53142baa68d7de61884fb6e.png][]

### 3、操作实践 ###

Canal提供了各种语言的客户端，当Canal监听到binlog变化时，会通知Canal的客户端。

![3661c68125454c8baffb2c4ff7e0b1f1.png][]

我们可以利用Canal提供的**Java客户端，监听**Canal通知消息。当收到变化的消息时，完成对缓存的更新。

这里推荐使用GitHub上的第三方开源的**canal-starter**客户端。地址：[https://github.com/NormanGyllenhaal/canal-client][https_github.com_NormanGyllenhaal_canal-client]

与SpringBoot完美整合，自动装配，比官方客户端要简单好用很多。

#### 3.1.引入依赖： ####

<dependency>
        <groupId>top.javatool</groupId>
        <artifactId>canal-spring-boot-starter</artifactId>
        <version>1.2.1-RELEASE</version>
    </dependency>

#### 3.2.编写配置： ####

canal:
      destination: item# canal的集群名字，要与安装canal时设置的名称一致
      server: 192.168.123.101:11111 # canal服务地址

#### 注意：下面是我学习项目时候的案例，不需要的小伙伴可以不用看~ ####

#### 3.3.修改Item实体类 ####

通过@Id、@Column、等注解完成Item与数据库表字段的映射：

package com.heima.item.pojo;
    
    import com.baomidou.mybatisplus.annotation.IdType;
    import com.baomidou.mybatisplus.annotation.TableField;
    import com.baomidou.mybatisplus.annotation.TableId;
    import com.baomidou.mybatisplus.annotation.TableName;
    import lombok.Data;
    import org.springframework.data.annotation.Id;
    import org.springframework.data.annotation.Transient;
    
    import javax.persistence.Column;
    import java.util.Date;
    
    @Data
    @TableName("tb_item")
    public class Item {
        @TableId(type = IdType.AUTO)
        @Id //这个对应的是表的主键
        private Long id;//商品id
        @Column(name = "name")
        private String name;//商品名称
        private String title;//商品标题
        private Long price;//价格（分）
        private String image;//商品图片
        private String category;//分类名称
        private String brand;//品牌名称
        private String spec;//规格
        private Integer status;//商品状态 1-正常，2-下架
        private Date createTime;//创建时间
        private Date updateTime;//更新时间
        @TableField(exist = false)
        @Transient //这个注解 标识字段不属于数据库 
        private Integer stock;
        @TableField(exist = false)
        @Transient
        private Integer sold;
    }

#### 5.3.4.编写监听器 ####

通过实现`EntryHandler<T>`接口编写监听器，监听Canal消息。注意两点：

*  实现类通过`@CanalTable("tb_item")`指定监听的表信息
 *  EntryHandler的泛型是与表对应的实体类

package com.heima.item.canal;
    
    import com.github.benmanes.caffeine.cache.Cache;
    import com.heima.item.config.RedisHandler;
    import com.heima.item.pojo.Item;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import top.javatool.canal.client.annotation.CanalTable;
    import top.javatool.canal.client.handler.EntryHandler;
    
    @CanalTable("tb_item")  //这个字段标注 我们对应的数据库表 canal使用
    @Component
    public class ItemHandler implements EntryHandler<Item> {
    
        @Autowired
        private RedisHandler redisHandler;
        @Autowired
        private Cache<Long, Item> itemCache;
    
        @Override
        public void insert(Item item) {
            // 写数据到JVM进程缓存
            itemCache.put(item.getId(), item);
            // 写数据到redis
            redisHandler.saveItem(item);
        }
    
        @Override
        public void update(Item before, Item after) {
            // 写数据到JVM进程缓存
            itemCache.put(after.getId(), after);
            // 写数据到redis
            redisHandler.saveItem(after);
        }
    
        @Override
        public void delete(Item item) {
            // 删除数据到JVM进程缓存
            itemCache.invalidate(item.getId());
            // 删除数据到redis
            redisHandler.deleteItemById(item.getId());
        }
    }

在这里对Redis的操作都封装到了RedisHandler这个对象中，是我们之前做缓存预热时编写的一个类，内容如下：

package com.heima.item.config;
    
    import com.fasterxml.jackson.core.JsonProcessingException;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.heima.item.pojo.Item;
    import com.heima.item.pojo.ItemStock;
    import com.heima.item.service.IItemService;
    import com.heima.item.service.IItemStockService;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Component;
    
    import java.util.List;
    
    @Component
    public class RedisHandler implements InitializingBean {
    
        @Autowired
        private StringRedisTemplate redisTemplate;
    
        @Autowired
        private IItemService itemService;
        @Autowired
        private IItemStockService stockService;
    
        private static final ObjectMapper MAPPER = new ObjectMapper();
    
        @Override
        public void afterPropertiesSet() throws Exception {
            // 初始化缓存
            // 1.查询商品信息
            List<Item> itemList = itemService.list();
            // 2.放入缓存
            for (Item item : itemList) {
                // 2.1.item序列化为JSON
                String json = MAPPER.writeValueAsString(item);
                // 2.2.存入redis
                redisTemplate.opsForValue().set("item:id:" + item.getId(), json);
            }
    
            // 3.查询商品库存信息
            List<ItemStock> stockList = stockService.list();
            // 4.放入缓存
            for (ItemStock stock : stockList) {
                // 2.1.item序列化为JSON
                String json = MAPPER.writeValueAsString(stock);
                // 2.2.存入redis
                redisTemplate.opsForValue().set("item:stock:id:" + stock.getId(), json);
            }
        }
    
        public void saveItem(Item item) {
            try {
                String json = MAPPER.writeValueAsString(item);
                redisTemplate.opsForValue().set("item:id:" + item.getId(), json);
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }
        }
    
        public void deleteItemById(Long id) {
            redisTemplate.delete("item:id:" + id);
        }
    }

#### 如果Canal安装中出现什么问题，查看我另一篇博客，或许对你有帮助：[Docker安装canal，出现caching\_sha2\_password Auth failed问题的解决，成功解决canal和mysql的连接问题\_Đến❦หัวใจ的博客-CSDN博客][Docker_canal_caching_sha2_password Auth failed_canal_mysql_n_-CSDN] ####

### 觉得有用给个路过的点赞吧~ ###

[GitHub - alibaba_canal_ _ MySQL binlog]: https://github.com/alibaba/canal
[bf035abc98e64711b677df2d12e55785.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/ae31dd3a1e454985bab6d47dff9433cb.png
[a9cdb3f5f53142baa68d7de61884fb6e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/c94d25da4a1e4622b3131a81a1f4bd02.png
[3661c68125454c8baffb2c4ff7e0b1f1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/1ba3d43fa66e48bfb43a9dc8a5ecb1f4.png
[https_github.com_NormanGyllenhaal_canal-client]: https://github.com/NormanGyllenhaal/canal-client
[Docker_canal_caching_sha2_password Auth failed_canal_mysql_n_-CSDN]: https://blog.csdn.net/qq_52348926/article/details/129992189?spm=1001.2014.3001.5501