canal redis 缓存一致性

雨点打透心脏的1/2处 2022-10-04 14:55 128阅读 0赞

--------------------

## canal redis 缓存一致性 ##

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**缓存一致性**

**为加快数据查询，常对一些热点数据进行缓存，缓存与源数据不可避免存在一致性问题**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70][]

**数据一致性一般发生在更新数据时：**

**如果对数据一致性要求不高，可对缓存数据设置超时时间，**到期后读取最新数据，并更新缓存

**如果对数据一致性要求较高，一般有如下处理**

spring默认规则：更新数据后，再更新缓存
    延时双删：先删除缓存，然后更新数据，延时删除缓存
    延时删除：更新数据后，延时删除缓存

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**更新数据，再更新缓存**

**高并发时可能存在如下情况**

![20210715231231416.png][]

**更新后端数据：**thread 1先更新后端数据、thread 2后更新后端数据；

**修改缓存时：**thread 2先修改缓存，thread 1后修改缓存

**此时，后端数据库为最新数据，缓存为旧的数据**

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**延时双删、延时删除**

**延时双删：先删缓存，再更新后端数据，再延时删除缓存**

![20210716212709643.png][]

**延时删除：先更新后端数据，再延时删除缓存**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 1][]

**如果立即删除缓存，延时双删存在如下可能导致数据不一致**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 2][]

**          **

**如果立即删除缓存，延时删除可能存在如下可能导致数据不一致**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 3][]

**后端数据更新后，延时删除缓存可能会失败，可使用mq重试，确保缓存删除成功**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 4][]

**canal监听 binlog，在后台处理，可以避免对业务造成侵入**

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**示例**

**canal 实现缓存一致性，做如下简单处理**

**新增后端数据：**canal 客户端直接插入缓存

**删除后端数据：**canal 客户端直接删除缓存

**更新后端数据：**canal 客户端直接删除缓存

**查询数据：**缓存中查找，缓存中查找不到，到后端数据库查找，将返回数据存入缓存

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**pojo 层**

**Person**

@Data
    @EqualsAndHashCode(callSuper = false)
    public class Person extends Model<Person> {
    
        private static final long serialVersionUID = 1L;
    
        @TableId(value = "id", type = IdType.AUTO)
        private Integer id;
    
        private String name;
    
        private Integer age;
    
    
        @Override
        protected Serializable pkVal() {
            return this.id;
        }
    
    }

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**myannotation 层**

**CustomCacheable：自定义查询注解**

@Documented
    @Target(ElementType.METHOD)
    @Retention(RetentionPolicy.RUNTIME)
    public @interface CustomCacheable {
     
        String key() default "";
    }

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**aspect 层**

**CustomAspect**

@Aspect
    @Component
    public class CustomAspect {
    
        @Resource(name = "stringRedisTemplate")
        private StringRedisTemplate redisTemplate;
    
        @Pointcut("@annotation(com.example.demo.myannotation.CustomCacheable)")
        public void fun(){
        }
     
        @Around("fun()")
        public Object CustomCacheable(ProceedingJoinPoint joinPoint){
            Object result=null;
    
            String simpleName=joinPoint.getTarget().getClass().getSimpleName();
            String pojoName=simpleName.substring(0,simpleName.length()-"ServiceImpl".length());
            String prefix=pojoName.substring(0,1).toLowerCase()+pojoName.substring(1);
    
            String pojoClass="com.example.demo.pojo."+pojoName;
            Class<?> c=null;
            try {
                c=Class.forName(pojoClass);
            }catch (Exception e){
                e.printStackTrace();
            }
    
            Method method=((MethodSignature)joinPoint.getSignature()).getMethod();
            CustomCacheable cacheable=method.getDeclaredAnnotation(CustomCacheable.class);
            String key=getKeyValue(joinPoint,method,cacheable.key());
            key=prefix+"."+key;
    
            String value=redisTemplate.boundValueOps(key).get();
            if (value!=null){
                return JSON.parseObject(value,c);
            }else {
                try {
                    result=joinPoint.proceed();
    
                    redisTemplate.boundValueOps(key).set(JSON.toJSONString(result),
                            60+new Random().nextLong()/60, TimeUnit.SECONDS);
                }catch (Throwable e){
                    e.printStackTrace();
                }
            }
    
            return result;
        }
     
        private String getKeyValue(ProceedingJoinPoint joinPoint,Method method,String key) {
            if (key==null||!key.startsWith("#")){
                return null;
            }
     
            String result=null;
            if (!key.contains(".")){
                String name=key.substring(1);
     
                Object[] args=joinPoint.getArgs();
                Parameter[] parameters=method.getParameters();
     
                for (int i=0;i<parameters.length;i++){
                    if (parameters[i].getName().equals(name)){
                        result=args[i].toString();
                        break;
                    }
                }
            }else {
                String[] s=key.substring(1).split("\\.");
                Object[] args = joinPoint.getArgs();
                Parameter[] parameters=method.getParameters();
     
                for (int i=0;i<parameters.length;i++){
                    if (parameters[i].getName().equals(s[0])){
                        try {
                            Field field=parameters[i].getType().getDeclaredField(s[1]);
                            field.setAccessible(true);
     
                            result= field.get(args[i]).toString();
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }
                }
            }
     
            return result;
        }
    }

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**service.impl 层**

**PersonServiceImpl**

@Service
    public class PersonServiceImpl extends ServiceImpl<PersonMapper, Person> implements PersonService {
    
        @Resource
        private PersonMapper personMapper;
    
        @Override
        @CustomCacheable(key = "#id")
        public Person getById(Integer id) {
            System.out.println("查询："+id);
    
            return personMapper.selectById(1);
        }
    }

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**service.canal 层**

**CanalRedisService：后端缓存同步**

@Service
    public class CanalRedisService {
    
        @Resource(name = "stringRedisTemplate")
        private StringRedisTemplate redisTemplate;
    
        private final ExecutorService executorService= Executors.newFixedThreadPool(5);
    
        @PostConstruct
        public void sync(){
            executorService.submit(()->{
                CanalConnector connector= CanalConnectors.newSingleConnector(
                        new InetSocketAddress("192.168.57.120",11111),
                        "example","","");
    
                int batchSize=100;
                int emptyCount=0;
                try {
                    connector.connect();
                    connector.subscribe("lihu\\..*");
    
                    int totalEmptyCount=100;
                    while (emptyCount < totalEmptyCount){
                        Message message=connector.getWithoutAck(batchSize);
                        long batchId=message.getId();
                        int size=message.getEntries().size();
    
                        if (batchId==-1 || size==0){
                            emptyCount++;
                            //System.out.println("emptyCount："+emptyCount);
    
                            try {
                                Thread.sleep(10000);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else {
                            emptyCount=0;
    
                            doSync(message.getEntries());
                        }
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    connector.disconnect();
                }
            });
        }
    
        public void doSync(List<CanalEntry.Entry> entries){
            for (CanalEntry.Entry entry:entries){
                if (entry.getEntryType()== CanalEntry.EntryType.TRANSACTIONBEGIN
                        || entry.getEntryType()== CanalEntry.EntryType.TRANSACTIONEND){
                    continue;
                }
    
                CanalEntry.RowChange rowChange=null;
                try {
                    rowChange= CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                }catch (Exception e){
                    throw new RuntimeException(e.getMessage());
                }
    
                CanalEntry.EventType eventType=rowChange.getEventType();
                System.out.println("事件类型："+eventType);
                System.out.println("数据库名："+entry.getHeader().getSchemaName());
                System.out.println("表名："+entry.getHeader().getTableName());
                System.out.println("binlog name："+entry.getHeader().getLogfileName());
                System.out.println("binlog position："+entry.getHeader().getLogfileOffset()+"\n");
    
                String tableName=entry.getHeader().getTableName();
                for (CanalEntry.RowData data:rowChange.getRowDatasList()){
                    if (eventType == CanalEntry.EventType.DELETE){
                        deleteCache(tableName,data);
                    }else if (eventType == CanalEntry.EventType.INSERT
                            || eventType ==CanalEntry.EventType.UPDATE) {
                        insertOrUpdateCache(tableName,data);
                    }
                }
            }
        }
    
        public void deleteCache(String tableName, CanalEntry.RowData rowData){
            if (tableName==null){
                return;
            }
    
            for (CanalEntry.Column column : rowData.getBeforeColumnsList()){
                if (column.getName().equals("id")){
                    String value=column.getValue();
                    String key=tableName+"."+value;
    
                    System.out.println("删除缓存："+key+"  "+value);
                    redisTemplate.delete(key);
                }
            }
        }
    
        public void insertOrUpdateCache(String tableName, CanalEntry.RowData rowData){
            if (tableName==null){
                return;
            }
    
            String baseClassName="com.example.demo.pojo";
            String className=tableName.substring(0,1).toUpperCase()+tableName.substring(1);
    
            try {
                Class<?> c=Class.forName(baseClassName+"."+className);
                Object o=c.getConstructor().newInstance();
    
                String key=tableName+".";
                for (CanalEntry.Column column:rowData.getAfterColumnsList()){
                    if (column.getName().equals("id")){
                        key+=column.getValue();
                    }
    
                    if ("name".equals(column.getName())){
                        Field field=c.getDeclaredField(column.getName());
                        field.setAccessible(true);
                        field.set(o,column.getValue());
                    }else {
                        Field field=c.getDeclaredField(column.getName());
                        field.setAccessible(true);
                        field.set(o,Integer.parseInt(column.getValue()));
                    }
                }
    
                System.out.println("插入或更新缓存："+key+"  "+JSON.toJSONString(o));
                redisTemplate.boundValueOps(key).set(JSON.toJSONString(o),
                        Duration.ofSeconds(60*5+new Random().nextLong()%60));
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**controller 层**

**PersonController**

@RestController
    @RequestMapping("/person")
    public class PersonController {
    
        @Resource
        private PersonService personService;
    
        @RequestMapping("/get")
        public Person getById(Integer id){
            return personService.getById(id);
        }
    
        @RequestMapping("/save")
        public String save(Integer id, String name, Integer age){
            Person person=new Person();
            person.setId(id);
            person.setName(name);
            person.setAge(age);
            personService.save(person);
    
            return "success";
        }
    
        @RequestMapping("/update")
        public String update(Integer id, String name){
            Person person=personService.getById(id);
            person.setName(name);
    
            personService.updateById(person);
            return "success";
        }
    
        @RequestMapping("/delete")
        public String delete(Integer id){
            personService.removeById(id);
    
            return "success";
        }
    }

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**使用测试**

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**插入数据**

**localhost:8080/person/save?id=6&name=瓜田李下&age=20****，控制台输出**

事件类型：INSERT
    数据库名：lihu
    表名：person
    binlog name：binlog.000004
    binlog position：2180
    
    插入或更新缓存：person.6  {"age":20,"id":6,"name":"瓜田李下"}

**redis 缓存：** **插入数据后，redis中新增了缓存数据**

![2021072413151930.png][]

**查询数据： localhost:8080/person/get?id=6**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 5][]

**删除数据：localhost:8080/person/delete?id=6，控制台输出**

事件类型：DELETE
    数据库名：lihu
    表名：person
    binlog name：binlog.000004
    binlog position：2786
    
    删除缓存：person.6  6

**查询数据：localhost:8080/person/get?id=6，****控制台输出 ==> 查询：6**

**说明：**由于缓存数据删除了，需要到后端查询数据

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**更新数据**

**插入数据：**localhost:8080/person/save?id=8&name=瓜田李下&age=20

**更新数据：**localhost:8080/person/update?id=8&name=海贼王，**控制台输出**

事件类型：INSERT
    数据库名：lihu
    表名：person
    binlog name：binlog.000004
    binlog position：3089
    
    插入或更新缓存：person.8  {"age":20,"id":8,"name":"瓜田李下"}
    事件类型：UPDATE
    数据库名：lihu
    表名：person
    binlog name：binlog.000004
    binlog position：3401
    
    插入或更新缓存：person.8  {"age":20,"id":8,"name":"海贼王"}

**redis 缓存数据**

![20210724133107821.png][]

**查询数据：**loclahost:8080/person/get?id=8

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 6][]

**查询时使用redis 缓存，控制台无输出**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70]: /images/20221004/67ba2732ce44411091146dd10584ff95.png
[20210715231231416.png]: /images/20221004/0a026af1ea5e4c50ac1e0f7d565b027f.png
[20210716212709643.png]: /images/20221004/3de6015dbcfd4b5caa79f002f5878ebc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 1]: /images/20221004/19a2716bce6f48e0a7bc46e5e4e39878.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 2]: /images/20221004/995123f3743a40dea642eddf38399bd9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 3]: /images/20221004/5183a6f255bd4b64a556e28ade2ec048.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 4]: /images/20221004/fe1e58f16c7148c78ca74af77c712c4e.png
[2021072413151930.png]: /images/20221004/2071dbef1dba48e6a87911f5033b5906.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 5]: /images/20221004/541bbc27ef154dfd93313c443cbe3f1f.png
[20210724133107821.png]: /images/20221004/9848905749eb4d14a82cb20ba663ff0b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 6]: /images/20221004/735e7afd5b994b1588a4516209c64027.png