12月第十九讲:Redis应用Redis相干办理方案 - ToB企服应用市场:ToB评测及商务社交产业平台

//处理MQ消息保证最终数据一致性
@Slf4j
@Component
public class CookbookConsistencyListener implements MessageListenerConcurrently {
@Autowired
private RedisCache redisCache;
//处理MySQL的binlog变化，处理需要清理的缓存key
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
try {
for (MessageExt messageExt : list) {
String msg = new String(messageExt.getBody());
//解析binlog数据模型，并过滤掉查询
BinlogDataDTO binlogData = buildBinlogData(msg);
//获取binlog的模型，获取本次变化的表名称，在本地配置常量类里面匹配对应的缓存key前缀以及缓存标识字段，非配置的表不进行处理
String cacheKey = filterConsistencyTable(binlogData);
//删除该key的缓存
deleteCacheKey(cacheKey);
}
} catch (Exception e) {
log.error("consume error, 缓存清理失败", e);
//本次消费失败，下次重新消费
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
//解析binlog的数据模型，并过滤掉查询的binlog
private BinlogDataDTO buildBinlogData(String msg) {
//先解析binlog的对象，转换为模型
BinlogDataDTO binlogData = BinlogUtils.getBinlogData(msg);
//模型为null，则直接返回
if (Objects.isNull(binlogData)) {
return null;
}
Boolean isOperateType = BinlogType.INSERT.getValue().equals(binlogData.getOperateType())
|| BinlogType.DELETE.getValue().equals(binlogData.getOperateType())
|| BinlogType.UPDATE.getValue().equals(binlogData.getOperateType());
//只保留增删改的binlog对象，如果数据对象为空则也不处理
if (!isOperateType || CollectionUtils.isEmpty(binlogData.getDataMap())) {
return null;
}
//返回解析好的可用模型
return binlogData;
}
//过滤掉目前不需要处理的表的Binlog，并返回组装所需的缓存key
private String filterConsistencyTable(BinlogDataDTO binlogData) {
if (Objects.isNull(binlogData)) {
return null;
}
String tableName = binlogData.getTableName();
List<Map<String, Object>> dataList = binlogData.getDataMap();
//获取配置的常量映射的具体配置
ConsistencyTableEnum consistencyTableEnum = ConsistencyTableEnum.findByEnum(tableName);
if (Objects.isNull(consistencyTableEnum)) {
return null;
}
String cacheValue = "";
if (CollectionUtils.isNotEmpty(dataList)) {
Map<String, Object> dataMap = dataList.get(0);
cacheValue = dataMap.get(consistencyTableEnum.getCacheField()) + "";
}
if (StringUtils.isBlank(cacheValue)) {
return null;
}
//获取配置的缓存前缀key + 当前的标识字段，组装缓存key
return consistencyTableEnum.getCacheKey() + cacheValue;
}
//对缓存进行清理
private void deleteCacheKey(String cacheKey) {
if (StringUtils.isBlank(cacheKey)) {
return;
}
redisCache.delete(cacheKey);
}
}
public enum ConsistencyTableEnum {
//商品表缓存配置
SKU_INFO("sku_info", RedisKeyConstants.GOODS_INFO_PREFIX, "id");
//配置相关的表名称
private final String tableName;
//缓存的前缀key
private final String cacheKey;
//缓存的标识字段
private final String cacheField;
...
}

复制代码