互联网金融网站建设,网站建设教程纯正苏州久远网络,网站开发通常叫什么部门,重庆网站的制作价格缓存双写一致性
更新策略探讨 面试题 缓存设计要求
缓存分类#xff1a;
只读缓存#xff1a;#xff08;脚本批量写入#xff0c;canal 等#xff09;读写缓存 同步直写#xff1a;vip数据等即时数据异步缓写#xff1a;允许延时#xff08;仓库#xff0c;物流
只读缓存脚本批量写入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 --privilegedtrue -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_PASSWORD123456 --name mysql mysql:5.7docker exec -it e8b72b11df2a /bin/bash //e8b72b11df2a 就是容器idmysql -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 utf8mb42.canal 准备
下载 canal.deployer-1.1.6.tar.gz
修改配置文件 canal.instance.master.addressxxx:xxx:xxx:xxx:3306# username/password (如果配置的是 canal/canal 则默认无需修改)
canal.instance.dbUsernamecanal
canal.instance.dbPasswordcanal./bin/startup.sh查看启动日志检查运行状态 3.编写 canal-client parentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.7.10/versionrelativePath/ !-- lookup parent from repository --/parentdependencies!--canal--dependencygroupIdcom.alibaba.otter/groupIdartifactIdcanal.client/artifactIdversion1.1.0/version/dependency!--SpringBoot与Redis整合依赖--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-redis/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.26/version/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build配置文件(请按需修改)
server:port: 5555spring:redis:database: 0host: 127.0.0.1port: 6379
# password: 123456config 文件和 springboot 案例中一致
Configuration
public class RedisConfig {Beanpublic RedisTemplateString, Object redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {RedisTemplateString, Object redisTemplate new RedisTemplate();redisTemplate.setConnectionFactory(lettuceConnectionFactory);// 设置key的序列化方式stringredisTemplate.setKeySerializer(new StringRedisSerializer());// 设置value的序列化方式jsonredisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());// 设置hash的key的序列化方式stringredisTemplate.setHashKeySerializer(new StringRedisSerializer());// 设置hash的value的序列化方式jsonredisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.afterPropertiesSet();return redisTemplate;}
}biz 操作封装
Component
public class RedisCanalClient {ResourceRedisTemplate redisTemplate;/*** 读取 canal 数据写入redis* param columns 行*/public void redisInsert(ListCanalEntry.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(ListCanalEntry.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(ListCanalEntry.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(ListCanalEntry.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(gt; 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;ResourceRedisCanalClient redisCanalClient;/*** 测试 canal client 监听 server 实现 mysql - redis*/Testpublic 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 {// 计数器置0emptyCount 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 的这条数据
