缓存双写一致性
更新策略探讨
面试题
缓存设计要求
缓存分类：
- 只读缓存：（脚本批量写入，canal 等）
- 读写缓存
- 同步直写：vip数据等即时数据
- 异步缓写：允许延时（仓库，物流），异常出现，有可能需要使用 kafka, rabbitmq 进行弥补，重试重写
双检加锁
如果 qps 过高，会打高 mysql
数据库和缓存更新的几种策略
=》实现最终一致性
可以停机->单线程操作
四种更新策略
先更新数据库，再更新缓存
异常问题1->最后更新redis异常，出现脏数据
异常问题2->多线程更新快慢问题，无法保证第二步写redis的顺序
先更新缓存，再更新数据库
业务上，mysql是底单数据
仍然是多线程更新的问题
自己的理解，只要是写缓存的方式，都会存在线程写入先后导致的数据不一致
先删除缓存，再更新数据库
第一个进来的线程当数据库写未提交或者其他异常情况的时候，仍然可能存在脏数据问题，因为无法保证缓存会被一定及时删除，可能中途被其他线程回写，这段要自己脑子里跑一遍，延迟双删来确保旧的缓存会被删除掉
读缓存会写入旧数据
延时双删
先更新数据库，再删除缓存->主流
存在问题->没来得及更新完缓存，会读取到旧值，但伤害较小
只要涉及到数据库和缓存的双写，百分百存在一致性问题
如何实现最终一致性
只能实现最终一致性
如何取舍&总结
双写一致性落地案例
监听 binlog
mysql -> canal -> redis
Home
canal 选择版本 1.6.1
mysql: 5.7 docker环境安装
1.mysql 准备
docker run -d -p 3306:3306 --privileged=true -v /root/mysql/log:/var/log/mysql -v /root/mysql/data:/var/lib/mysql -v /root/mysql/conf:/etc/mysql/conf.d -e MYSQL_ROOT_PASSWORD=123456 --name mysql mysql:5.7
docker exec -it e8b72b11df2a /bin/bash //e8b72b11df2a 就是容器id
mysql -uroot -p // 密码 123456
# 查看是否开启 binlog
SHOW VARIABLES LIKE 'log_bin';
# 添加 canal 操作账号
DROP USER IF EXISTS 'canal'@'%';
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';
FLUSH PRIVILEGES;
SELECT * FROM mysql.user;
# 创建一张表
CREATE TABLE `t_user`
(
`id` BIGINT ( 20 ) NOT NULL AUTO_INCREMENT,
`userName` VARCHAR ( 100 ) NOT NULL,
PRIMARY KEY ( `id` )
)
ENGINE = INNODB AUTO_INCREMENT = 10 DEFAULT CHARSET = utf8mb4
2.canal 准备
下载 canal.deployer-1.1.6.tar.gz
修改配置文件
canal.instance.master.address=xxx:xxx:xxx:xxx:3306
# username/password (如果配置的是 canal/canal 则默认无需修改)
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
./bin/startup.sh
查看启动日志，检查运行状态
3.编写 canal-client
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.10</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
<!--canal-->
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.0</version>
</dependency>
<!--SpringBoot与Redis整合依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.26</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
配置文件(请按需修改)
server:
port: 5555
spring:
redis:
database: 0
host: 127.0.0.1
port: 6379
# password: 123456
config 文件和 springboot 案例中一致
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(lettuceConnectionFactory);
// 设置key的序列化方式string
redisTemplate.setKeySerializer(new StringRedisSerializer());
// 设置value的序列化方式json
redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
// 设置hash的key的序列化方式string
redisTemplate.setHashKeySerializer(new StringRedisSerializer());
// 设置hash的value的序列化方式json
redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
redisTemplate.afterPropertiesSet();
return redisTemplate;
}
}
biz 操作封装
@Component
public class RedisCanalClient {
@Resource
RedisTemplate redisTemplate;
/**
* 读取 canal 数据写入redis
* @param columns 行
*/
public void redisInsert(List<CanalEntry.Column> columns) {
JSONObject jsonObject = new JSONObject();
for (CanalEntry.Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
jsonObject.put(column.getName(), column.getValue());
}
if (columns.size()>0) {
redisTemplate.opsForValue().set(columns.get(0).getValue(), jsonObject.toJSONString());
}
}
/**
* 读取 canal 数据删除 redis 行
* @param columns 行
*/
public void redisDelete(List<CanalEntry.Column> columns) {
JSONObject jsonObject = new JSONObject();
for (CanalEntry.Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
jsonObject.put(column.getName(), column.getValue());
}
if (columns.size()>0) {
redisTemplate.delete(columns.get(0).getValue());
}
}
/**
* 读取 canal 数据修改 redis 行
* @param columns 行
*/
public void redisUpdate(List<CanalEntry.Column> columns) {
JSONObject jsonObject = new JSONObject();
for (CanalEntry.Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
jsonObject.put(column.getName(), column.getValue());
}
if (columns.size()>0) {
redisTemplate.opsForValue().set(columns.get(0).getValue(), jsonObject.toJSONString());
System.out.println("----------------update after: " + redisTemplate.opsForValue().get(columns.get(0).getValue()));
}
}
/**
* 读取 canal 数据并操作
* @param entrys 行
*/
public void printEntry(List<CanalEntry.Entry> entrys) {
for (CanalEntry.Entry entry : entrys) {
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
continue;
}
CanalEntry.RowChange rowChage = null;
try {
rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
e);
}
CanalEntry.EventType eventType = rowChage.getEventType();
System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
eventType));
for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
if (eventType == CanalEntry.EventType.DELETE) {
redisDelete(rowData.getBeforeColumnsList());
} else if (eventType == CanalEntry.EventType.INSERT) {
redisInsert(rowData.getAfterColumnsList());
} else {
redisUpdate(rowData.getAfterColumnsList());
}
}
}
}
}
test 测试类
@SpringBootTest
public class CanalClientTest {
public static final Integer _60SECONDS = 60;
@Resource
RedisCanalClient redisCanalClient;
/**
* 测试 canal client 监听 server 实现 mysql -> redis
*/
@Test
public void startClient() {
System.out.println("--------------initCanal main()方法------------");
// ======================================================================================
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress("127.0.0.1", 11111), // canal server 地址
"example",
"",
"");
int batchSize = 1000;
int emptyCount = 0;
System.out.println("---------------------canal init OK，开始监听mysql变化------");
try {
connector.connect();
// connector.subscribe(".*\\..*");
// 设置监听的表
connector.subscribe("canal.t_user");
connector.rollback();
int totalEmptyCount = 10 * _60SECONDS;
while (emptyCount < totalEmptyCount) {
System.out.println("我是 canal, 每秒一次正在监听: " + UUID.randomUUID().toString());
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
emptyCount++;
System.out.println("empty count : " + emptyCount);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
} else {
// 计数器置0
emptyCount = 0;
// System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
redisCanalClient.printEntry(message.getEntries());
}
connector.ack(batchId); // 提交确认
// connector.rollback(batchId); // 处理失败, 回滚数据
}
System.out.println("已经监听了"+totalEmptyCount+"秒，无任何消息，请重启重试");
} finally {
connector.disconnect();
}
}
}
启动测试类
4.测试
- 测试新增
手动添加一条记录 id: 7 name:666
查看 redis
- 测试修改
修改mysql id:7 的 name 为 777
- 测试删除
删除 id:7 的这条数据
