当前位置：首页 > news >正文

电商项目_性能优化_数据同步

news 2025/8/2 10:13:50

1、缓存同步

使用场景

1. Redis缓存当做数据库

大促等场景，并发请求数量的基数过大，即便只有很小比率的请求直接访问数据库其绝对数量也仍然不小，还是会存在系统雪崩的风险。

构建Redis集群后，由于集群可以水平扩容，因此只要集群足够大，理论上支持海量并发就不是问题。

所以可将将Redis当做数据库使用，将所有数据都缓存到Redis。解决缓存不命中的问题。

2. 缓存一致性

电商项目_性能优化_高并发缓存一致性-CSDN博客

在缓存一致性章节讲到过，为了保证DB和缓存的数据一致性，DB变更时，需要删除Redis缓存。

思考：Redis缓存了这么多的DB数据，如何保证数据一致性呢？

1. 分布式事务保证DB和缓存一致性。 缺点：性能低，另外缓存的异常也不应该影响业务。

2. 写DB时发一条消息，消费端更新Redis。缺点：对代码侵入大，还要实现消息发送接收。

3. 使用开源组件，订阅Binlog日志，更新缓存。 GOOD!

使用Binlog实时更新Redis缓存

Canle组件

官方主页：https://github.com/alibaba/canal

它通过模拟MySOL主从复制的交互协议，把自己伪装成一个MySOL的从节点，向 MySOL主节点发送dump请求。MySOL收到请求后，就会向 Canal开始推送Binlog，Canal解析Binlog字节流之后，将其转换为便于读取的结构化数据，供下游程序订阅使用。实际运用后的运行架构如图：

电商项目实现过程

1. 安装运行Canal服务

2.Mysql打开binlog日志

开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式

3. 给Canal设置一个用来复制数据的MySQL账号

-- 定这个用户名为Canal，密码Canal
CREATE USER canal IDENTIFIED BY 'canal';GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';FLUSH PRIVILEGES;

4. 启动canel服务端

配置中的修改点：要同步的数据表、数据库

代码实现：

public class PromotionData implements IProcessCanalData {   private final static String SMS_HOME_ADVERTISE = "sms_home_advertise";private final static String SMS_HOME_BRAND = "sms_home_brand";private final static String SMS_HOME_NEW_PRODUCT = "sms_home_new_product";private final static String SMS_HOME_RECOMMEND_PRODUCT = "sms_home_recommend_product";@Autowired@Qualifier("promotionConnector")private CanalConnector connector;@PostConstruct@Overridepublic void connect() {tableMapKey.put(SMS_HOME_ADVERTISE,promotionRedisKey.getHomeAdvertiseKey());tableMapKey.put(SMS_HOME_BRAND,promotionRedisKey.getBrandKey());tableMapKey.put(SMS_HOME_NEW_PRODUCT,promotionRedisKey.getNewProductKey());tableMapKey.put(SMS_HOME_RECOMMEND_PRODUCT,promotionRedisKey.getRecProductKey());connector.connect();if("server".equals(subscribe))connector.subscribe(null);elseconnector.subscribe(subscribe);connector.rollback();}@Async@Scheduled(initialDelayString="${canal.promotion.initialDelay:5000}",fixedDelayString = "${canal.promotion.fixedDelay:5000}")@Overridepublic void processData() {try {if(!connector.checkValid()){log.warn("与Canal服务器的连接失效！！！重连，下个周期再检查数据变更");this.connect();}else{Message message = connector.getWithoutAck(batchSize);long batchId = message.getId();int size = message.getEntries().size();if (batchId == -1 || size == 0) {log.info("本次[{}]没有检测到促销数据更新。",batchId);}else{log.info("本次[{}]促销数据本次共有[{}]次更新需要处理",batchId,size);/*一个表在一次周期内可能会被修改多次，而对Redis缓存的处理只需要处理一次即可*/Set<String> factKeys = new HashSet<>();for(CanalEntry.Entry entry : message.getEntries()){if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN|| entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {continue;}CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());String tableName = entry.getHeader().getTableName();if(log.isDebugEnabled()){CanalEntry.EventType eventType = rowChange.getEventType();log.debug("数据变更详情：来自binglog[{}.{}]，数据源{}.{}，变更类型{}",entry.getHeader().getLogfileName(),entry.getHeader().getLogfileOffset(),entry.getHeader().getSchemaName(),tableName,eventType);}factKeys.add(tableMapKey.get(tableName));}for(String key : factKeys){/*此处未做删除缓存失败的补偿操作，可以自行加入*/if(StringUtils.isNotEmpty(key))  redisOpsExtUtil.delete(key);}connector.ack(batchId); // 提交确认log.info("本次[{}]处理促销Canal同步数据完成",batchId);}}} catch (Exception e) {log.error("处理促销Canal同步数据失效，请检查：",e);}}
}