目录
-
- 双写一致性
-
- redis与mysql双写一致性
-
- canal
-
- 配置流程
- 代码案例
- 双写一致性理解
-
- 缓存操作细分
- 缓存一致性多种更新策略
-
- 挂牌报错,凌晨升级
- 先更新数据库,在更新缓存
- 先删除缓存,在更新数据库
- 先更新数据库,在删除缓存
- 延迟双删策略
- 总结
双写一致性
redis与mysql双写一致性
canal
主要是用于mysql数据库增量日志数据的订阅,消费和解析(由阿里开源的java项目),canal是通过伪装成mysql的slave节点来转储master节点的binlog日志的一个中间件,他拿到日志内容以后,就可以把日志的相关数据变更重放到任何地方,可以是其他的mysql,也可以是消息队列,redis甚至是文件中.
配置流程
- 开启mysql的binlog写入功能(需要重启mysql,阿里云的好像默认就开启了)
- 授权canal连接mysql的账号,其实就是新建一个canal专用的账号便于区分(权限可以稍微高一些)
- 去ag真人试玩娱乐官网下载并解压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 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 columns) {
//实现省略,往redis插入数据
}
private static void redisupdate(list columns) {
//实现省略,往redis修改数据
}
private static void redisdelete(list columns) {
//实现省略,往redis删除数据
}
}
双写一致性理解
- redis中有数据,需要和数据库中的值相同
- redis中无数据,需要数据库中的值要是最新值
缓存操作细分
- 只读缓存
- 读写缓存
- 同步直写策略:写数据库时也同步写缓存,缓存和数据库中的数据一致(对于读写缓存来说,要想保证缓存和数据库中的数据一致,就要采用同步直写策略)
缓存一致性多种更新策略
挂牌报错,凌晨升级
让客户稍作等待,然后趁机更新mysql和redis(特别重要级别的数据最好不要多线程)
给缓存设置过期时间,是保证最终一致性的ag真人游戏的解决方案.所有的写操作以数据库为准,对缓存操作只是尽最大的努力即可.也就是说如果数据库写入成功,缓存更新失败,那么只要到达过期时间.后面的请求自然会从数据库中读取新数据然后回填缓存,达到一致性.切记以mysql的数据库写入为准.
先更新数据库,在更新缓存
在高并发的情境下,这个操作是跨两个不同的系统的,就一定会可能发生数据不一致的问题,导致读到脏数据(比如某方更新失败了)
先删除缓存,在更新数据库
容易出现的异常问题:a线程删除了缓存,去更新mysql. b线程过来又要读取,a还在更新中,这时候有可能发生
- 有可能缓存击穿(看你有没有双端检索加锁来初始化缓存)
- b从mysql获得了旧值
- b会把获得的旧值写回到redis缓存(被a删除掉的旧数据,又被b给写会了,缓存的更新就失败了)
- 请求a更新完成,mysql与redis发生了数据不一致的情况
这种方案尽量不要用
先更新数据库,在删除缓存
还是会出现短时间的数据不一致(可能会从缓存中读取到旧数据)
canal就是类似的思想
延迟双删策略
先删除redis的缓存,在更新完数据库之后,再删除一次redis的缓存(延迟删除),这时候能保证数据的最终一致性.
- 这个删除该休眠多久
- 自己根据业务进行一个具体的评估,在此耗时基础上面加个**百毫秒**左右即可
- 如果mysql是主从分离如何
- 从库更可能导致数据不一致问题(还有个主从复制的延迟时间),所以更加需要采用延迟双删的策略了(延迟时间可能需要再加上百毫秒时间)
- 这种同步淘汰策略,吞吐量降低了怎么办
- 可以新起来一个线程去后台做这个事情(用completablefuture等实现)
分布式系统只有最终一致性,很难去做到强一致性
总结
把redis作为只读缓存的话还好,没有一致性的问题,但是如果把redis作为读写缓存来用.建议使用先更新数据库,再删除缓存的方案.理由如下:
- 先删除缓存的值在更新数据库,有可能缓存击穿打满mysql,并且也避免不了数据不一致的问题
- 如果业务应用中读取数据库和写缓存的时间不好估算,那么延迟双删中的等待时间就不好设置