网站访问统计js代码,百度资讯,手机网站html声明,3m网站源码一、缓存双写一致性 如果redis中有数据#xff1a;需要和数据库中的值相同。如果redis中没有数据#xff1a;数据库中的值要是最新值#xff0c;且准备回写redis。只读缓存。读写缓存#xff1a;①、同步直写策略#xff1a;写数据库后也同步写redis缓存#xff0c;缓存…一、缓存双写一致性 如果redis中有数据需要和数据库中的值相同。如果redis中没有数据数据库中的值要是最新值且准备回写redis。只读缓存。读写缓存①、同步直写策略写数据库后也同步写redis缓存缓存和数据库中的数据一致对于读写缓存来说要想保证缓存和数据库中的数据一致就要采用同步直写策略。②、异步缓写策略正常业务运行中mysql数据变动了但是可以在业务上容许出现一定时间后才作用于redis如仓库、物流等功能。异常情况出现了不得不将失败的动作重新修补有可能需要借助kafka或者rabbitMQ等消息中间件实现重试重写。双检加锁策略多个线程同时去查询数据库的这条数据那么就可以第一个查询数据的请求上使用一个互斥锁来锁住它。其他线程走到这一步拿不到锁就等着等第一个线程查询到了数据然后做缓存。后面的线程进来发现已经有缓存了就直接走缓存。 1.1、数据库和缓存一致性的更新策略 目的给缓存设置过期时间定期清理缓存并回写是保证最终一致性的解决方案。 可以对存入缓存的数据设置过期时间所有的写操作以数据库为准对缓存操作只是尽最大努力即可。就是如果数据库写入成功缓存更新失败那么只要到达过期时间则后面的读请求自然会从数据库中读取新值然后回填缓存达到一致性要以mysql的数据库写入库为准。 1.1.1、在停机的情况下 给出公告服务升级单线程这样重量级的数据操作最好不要多线程。 1.1.2、先更新数据库再更新缓存 1、情况1①、先更新mysql的某商品的库存当前商品的库存是100更新为99。②、先更新mysql修改为99成功然后更新redis。③、出现异常更新redis失败了导致MySQL里面的库存是99而redis里面还是100。所以会导致数据库里的数据和缓存redis里面数据不一致读到redis脏数据。 2、情况2在多线程环境下A,B两个线程有快有慢。①、A更新mysql为100。②、B更新mysql为90。③、B先更新redis为90。④、A再更新redis为100。所以导致redis与mysql更新的数据不一致。 1.1.3、先更新缓存再更新数据库 不推荐业务上一般把mysql作为底单数据库保证最后解释。 1.1.4、先删除缓存再更新数据库 1、请求A进行写操作删除redis缓存后工作正在进行中更新mysql……A还没有彻底更新完mysql还没commit。 2、请求B开工查询查询redis发现缓存不存在(被A从redis中删除了)。 3、请求B继续去数据库查询得到了mysql中的旧值(A还没有更新完)。 4、请求B将旧值写回redis缓存。 5、请求A将新值写入mysql数据库。 这样依然会导致数据不一致的情况发生。 解决方法采用延时双删策略A线程删除redis缓存然后sleep一段时间这期间就是为了让B线程先从数据库读取数据再把缺失的数据写入缓存然后线程A再进行删除。所以线程A sleep的时间就需要大于线程B读取数据再写入缓存的时间。这样其它线程读取数据时会发现缓存缺失所以会从数据库中读取最新值因为这个方案会在第一次删除缓存后延迟一段时间再次进行删除所以叫做延迟双删。 延时双删的不足 这个删除该休眠多久呢 线程A sleep的时间需要大于线程B读取数据再写入缓存的时间。①、在业务程序运行的时候统计下线程读数据和写缓存的操作时间评估出项目的读数据业务逻辑的耗时以此为基础然后写数据的休眠时间则在读数据业务的耗时上加百毫秒就行。这样确保请求结束写请求可以删除读请求造成的缓存脏读。②、新启动一个后台监控程序如watchdog监控程序会加时。 1.1.5、先更新数据库在删除缓存 缺点缓存删除失败或者来不及导致请求再次访问redis时缓存命中读取到的是缓存旧值。 解决方案 可以把要删除的缓存值或者是要更新的数据库值暂存到消息队列中(如使用Kafaka/RabbitMQ)。当程序没有能够成功地删除缓存值或者是更新数据库值时可以从消息队列中重新读取这些值然后再次进行删除或更新。如果能够成功地删除或更新我们就要把这些值从消息队列中去除以免重复操作此时也可以保证数据库和缓存的数据一致了否则还需要再次进行重试。当重试超过一定次数后就需要向业务层发送保错信息了通知运维人员。 总结 1.2、Redis与Mysql数据双写一致性 1.2.1、canal 主要用途用于MySQL数据库增量日志数据的订阅消费和解析是阿里巴巴开发并开源的采用Java语言开发。 主要功能1、数据库镜像2、数据库实时备份。3、索引构建和实时维护(拆分异构索引、倒排索引等)。4、业务cache刷新。5、带业务逻辑的增量数据处理。 工作原理①、canal模拟MySQL  slave的交互协议伪装自己为MySQL master发送dump协议。②、MySQL master收到dump请求开始推送binary log给slave即canal ③、canal解析binary  log对象(原始为byte流)。 下载地址GitHub - alibaba/canal: 阿里巴巴 MySQL binlog 增量订阅消费组件 1.2.2、Redis与Mysql数据双写一致性实现 mysql前置配置 、MySQL 5.7.36、当前主机二进制日志SHOW  MASTER STATUS;、查看SHOW VARIABLES LIKE log_bin;、开启MySQL的binlog写入功能在mysql的ini文件中配置 log-binmysql-bin #开启binlog binlog-formatROW #开启ROW模式 server_id1 #配置MySQL replction需要定义不要和canal的slaveid重复   重启mysql、再次查看SHOW VARIABLES LIKE log_bin; 授权canal连接MySQL #先检查是否有canal SELECT*FROM mysql.user #没有就创建 CREATE USER canal% IDENTIFIED BY canal; GRANT ALL PRIVILEGES ON *.* TO canal% IDENTIFIED BY canal; FLUSH PRIVILEGES;   Canal服务端 、下载linux版本 解压 配置文件 启动 查看日志 Java程序 、sql脚本 CREATE TABLE a_user( id BIGINT(20) NOT NULL AUTO_INCREMENT, userName VARCHAR(100) NOT NULL, PRIMARY  KEY(id) )ENGINEINNODB  AUTO_INCREMENT10 DEFAULT CHARSETutf8mb4   public class RedisCanalClientExample {public static final Integer _60SECONDS 60;public static final String REDIS_IP_ADDR 192.168.200.110;private static void redisInsert(ListColumn columns) {JSONObject jsonObject new JSONObject();for (Column column : columns) {System.out.println(column.getName() : column.getValue() update column.getUpdated());jsonObject.put(column.getName(), column.getValue());}if (columns.size() 0) {try (Jedis jedis RedisUtils.getJedis()) {jedis.set(columns.get(0).getValue(), jsonObject.toJSONString());} catch (Exception e) {e.printStackTrace();}}}private static void redisDelete(ListColumn columns) {JSONObject jsonObject new JSONObject();for (Column column : columns) {jsonObject.put(column.getName(), column.getValue());}if (columns.size() 0) {try (Jedis jedis RedisUtils.getJedis()) {jedis.del(columns.get(0).getValue());} catch (Exception e) {e.printStackTrace();}}}private static void redisUpdate(ListColumn columns) {JSONObject jsonObject new JSONObject();for (Column column : columns) {System.out.println(column.getName() : column.getValue() update column.getUpdated());jsonObject.put(column.getName(), column.getValue());}if (columns.size() 0) {try (Jedis jedis RedisUtils.getJedis()) {jedis.set(columns.get(0).getValue(), jsonObject.toJSONString());System.out.println(---------update after: jedis.get(columns.get(0).getValue()));} catch (Exception e) {e.printStackTrace();}}}public static void printEntry(ListEntry entrys) {for (Entry entry : entrys) {if (entry.getEntryType() EntryType.TRANSACTIONBEGIN || entry.getEntryType() EntryType.TRANSACTIONEND) {continue;}RowChange rowChage null;try {//获取变更的row数据rowChage RowChange.parseFrom(entry.getStoreValue());} catch (Exception e) {throw new RuntimeException(ERROR ## parser of eromanga-event has an error,data: entry.toString(), e);}//获取变动类型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 (RowData rowData : rowChage.getRowDatasList()) {if (eventType EventType.INSERT) {redisInsert(rowData.getAfterColumnsList());} else if (eventType EventType.DELETE) {redisDelete(rowData.getBeforeColumnsList());} else {//EventType.UPDATEredisUpdate(rowData.getAfterColumnsList());}}}}public static void main(String[] args) {System.out.println(---------O(∩_∩)O哈哈~ initCanal() main方法-----------);//// 创建链接canal服务端CanalConnector connector CanalConnectors.newSingleConnector(new InetSocketAddress(REDIS_IP_ADDR, 11111),example,,);int batchSize 1000;//空闲空转计数器int emptyCount 0;System.out.println(---------------------canal init OK开始监听mysql变化------);try {connector.connect();//设置监控的数据库与表//connector.subscribe(.*\\..*);connector.subscribe(test1.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;try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}} else {//计数器重新置零emptyCount 0;printEntry(message.getEntries());}connector.ack(batchId); // 提交确认// connector.rollback(batchId); // 处理失败, 回滚数据}System.out.println(已经监听了 totalEmptyCount 秒无任何消息请重启重试......);} finally {connector.disconnect();}} }