Redis缓存双写一致性_redis高级篇之缓存双写一致性工程canle落地案例编码

双写一致性

Redis与Mysql双写一致性

canal

主要是用于MySQL数据库增量日志数据的订阅,消费和解析(由阿里开源的Java项目),canal是通过伪装成MySQL的slave节点来转储master节点的binlog日志的一个中间件,他拿到日志内容以后,就可以把日志的相关数据变更重放到任何地方,可以是其他的MySQL,也可以是消息队列,redis甚至是文件中.

配置流程

• 开启MySQL的binlog写入功能(需要重启MySQL,阿里云的好像默认就开启了)
• 授权canal连接MySQL的账号,其实就是新建一个canal专用的账号便于区分(权限可以稍微高一些)
• 去官网下载并解压canal到自己的目录下,修改instance.properties配置文件
• 换成自己mysql主机所在的ip地址
• 换成自己刚才给MySQL新建的用户及其密码
• 启动canel并查看server和instance实例的日志来确保启动运行成功

代码案例

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class RedisCanalClientExample {

    public static final int _60SECONDS = 60;

    public static void main(String[] args) {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(
                "127.0.0.1", 1111), "example", "", "");
        int batchSize = 1000;
        int emptyCount = 0;
        System.out.println("---------程序启动,开始监听MySQL的变化: ");
        try {
            connector.connect();
            //这个就是你要订阅的变化的那个库表
            connector.subscribe("db_test.t_user");
            connector.rollback();
            int totalEmptyCount = 10 * _60SECONDS;

            while (emptyCount < totalEmptyCount) {
                //获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    emptyCount = 0;
                    printEntry(message.getEntries());
                    System.out.println();
                }
                //提交确认
                connector.ack(batchId);
                //处理失败,回滚数据
                //connector.rollback(batchId);
            }
            System.out.println("empty too many times,exit");
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(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 (InvalidProtocolBufferException e) {
                throw new RuntimeException(e);
            }
            CanalEntry.EventType eventType = rowChange.getEventType();
            System.out.printf("==========binlog[%s:%s],name[%s,%s],eventType : %s%n",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType);
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.INSERT) {
                    redisInsert(rowData.getAfterColumnsList());
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    redisUpdate(rowData.getAfterColumnsList());
                } else {
                    redisDelete(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private static void redisInsert(List<CanalEntry.Column> columns) {
        //实现省略,往redis插入数据
    }

    private static void redisUpdate(List<CanalEntry.Column> columns) {
        //实现省略,往redis修改数据
    }

    private static void redisDelete(List<CanalEntry.Column> columns) {
        //实现省略,往redis删除数据
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

双写一致性理解

• redis中有数据,需要和数据库中的值相同
• redis中无数据,需要数据库中的值要是最新值

缓存操作细分

• 只读缓存
• 读写缓存
• 同步直写策略:写数据库时也同步写缓存,缓存和数据库中的数据一致(对于读写缓存来说,要想保证缓存和数据库中的数据一致,就要采用同步直写策略)

缓存一致性多种更新策略

挂牌报错,凌晨升级

让客户稍作等待,然后趁机更新mysql和redis(特别重要级别的数据最好不要多线程)

给缓存设置过期时间,是保证最终一致性的解决方案.所有的写操作以数据库为准,对缓存操作只是尽最大的努力即可.也就是说如果数据库写入成功,缓存更新失败,那么只要到达过期时间.后面的请求自然会从数据库中读取新数据然后回填缓存,达到一致性.切记以mysql的数据库写入为准.

先更新数据库,在更新缓存

在高并发的情境下,这个操作是跨两个不同的系统的,就一定会可能发生数据不一致的问题,导致读到脏数据(比如某方更新失败了)

先删除缓存,在更新数据库

容易出现的异常问题:A线程删除了缓存,去更新mysql. B线程过来又要读取,A还在更新中,这时候有可能发生

• 有可能缓存击穿(看你有没有双端检索加锁来初始化缓存)
• B从mysql获得了旧值
• B会把获得的旧值写回到Redis缓存(被A删除掉的旧数据,又被B给写会了,缓存的更新就失败了)
• 请求A更新完成,MySQL与Redis发生了数据不一致的情况

这种方案尽量不要用

先更新数据库,在删除缓存

还是会出现短时间的数据不一致(可能会从缓存中读取到旧数据)

canal就是类似的思想

延迟双删策略

先删除Redis的缓存,在更新完数据库之后,再删除一次Redis的缓存(延迟删除),这时候能保证数据的最终一致性.

• 这个删除该休眠多久
• 自己根据业务进行一个具体的评估,在此耗时基础上面加个**百毫秒**左右即可
• 如果MySQL是主从分离如何
• 从库更可能导致数据不一致问题(还有个主从复制的延迟时间),所以更加需要采用延迟双删的策略了(延迟时间可能需要再加上百毫秒时间)
• 这种同步淘汰策略,吞吐量降低了怎么办
• 可以新起来一个线程去后台做这个事情(用CompletableFuture等实现)

分布式系统只有最终一致性,很难去做到强一致性

总结

把Redis作为只读缓存的话还好,没有一致性的问题,但是如果把Redis作为读写缓存来用.建议使用先更新数据库,再删除缓存的方案.理由如下:

• 先删除缓存的值在更新数据库,有可能缓存击穿打满MySQL,并且也避免不了数据不一致的问题
• 如果业务应用中读取数据库和写缓存的时间不好估算,那么延迟双删中的等待时间就不好设置