Redis：使用Canal实现缓存同步

喜欢ヅ旅行 2024-03-24 08:40 21阅读 0赞

#### 引入依赖： ####

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

#### 编写配置： ####

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

#### 修改Item实体类 ####

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

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 javax.persistence.Column;
    
    
    @Data
    @TableName("tb_item")
    public class Item {
        @TableId(type = IdType.AUTO)
        @Id
        private Long id;//id
        @Column(name = "name")
        private String name;//名称
    }

#### 编写监听器 ####

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

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

import com.github.benmanes.caffeine.cache.Cache;
    import top.javatool.canal.client.annotation.CanalTable;
    import top.javatool.canal.client.handler.EntryHandler;
    
    @CanalTable("tb_item")
    @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这个对象中，是我们之前做缓存预热时编写的一个类，内容如下：

import com.fasterxml.jackson.core.JsonProcessingException;
    import com.fasterxml.jackson.databind.ObjectMapper;
    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;
    
        private static final ObjectMapper MAPPER = new ObjectMapper();
    
        // 缓存预热
        @Override
        public void afterPropertiesSet() throws Exception {
            // 初始化缓存
            // 查询item信息
            List<Item> itemList = itemService.list();
            // 放入缓存
            for (Item item : itemList) {
                // item序列化为JSON
                String json = MAPPER.writeValueAsString(item);
                // 存入redis
                redisTemplate.opsForValue().set("item:id:" + item.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);
        }
    }