高并发场景下解决并发数据不一致
简单的场景:
全量数据更新的情况下, 不在乎同一秒的请求都必须要成功, 只留下最新的更新请求数据
- 方案常用的是
1、数据库增加时间戳标识实现的乐观锁, 请求参数从源头带上微秒或者毫秒时间戳数据库存储, 然后在更新SQL语句上比较 (数据库的时间 < 参数传递的时间)
例如:
ALTER TABLE mytable ADD accessTime DATETIME(6) DEFAULT CURRENT_TIMESTAMP(6) not null comment ‘请求源头时间(精度高),用于更新比较,解决并发修改’;
UPDATE mytable SET updateTime = ?, accessTime=? WHERE
statusId = ?
AND
(IFNULL(accessTime, DATE_SUB(NOW(), INTERVAL 2 MINUTE)) < ?);
复杂场景:
比如, 更新数据时候不是全量数据更新的方式, 可能是增量更新或者是覆盖部分更新;
场景: 比如MongoDB数据库, 文档中的某个object字段下有很多key:value,
[图片]
一种并发情况: 并发A、B请求的时候, 他们的请求参数分别新增加不同的object字段下的Key, 或者是覆盖key, 这种新增Key的情况下需要并发的A、B请求都要成功,
{properties.label":24.700000}
{properties.thermostat.localTemp":24.800000}
另外一种情况: 并发A、B请求的时候, 他们的请求参数分别相同的object字段下的Key, 这种情况下需要并发的A、B请求只要最新请求能成功, 比如是A是先请求, B后请求, 那么就不能存在数据库最终存储的数据是A的;
{properties.thermostat.localTemp":24.700000}
{properties.thermostat.localTemp":24.800000}
- 方案
基于时间戳的乐观锁方式 + 判断keys的hash算法, 具体如下
请求参数源码增加一个时间戳, 透传到数据库存储;
根据请求测试里面的所有的keys , 按规则排序, 后进行hash算法计算出值, 然后插入到表里面;
后面更新判断hash算法是否一致, 不一致则判断为增量更新, 不用判断时间戳让其更新成功;
如果hash算法一致则判定为一致, 需要判断数据库时间戳 < 传递的时间戳;
并且这个判断不能在程序中实现, 比如先查询出来然后再比较, 后更新, 这种方式不是原子性的;
但是这种方法还是存在缺陷, 比如:
A线程先请求 参数是: b=2 {“fanMode”: “auto1”, “speedPercent”: 0} A 由于hash不同也能更新成功, 最终b=2, 但是由于是B后请求是, 其实最终数据要b=1
B线程后请求 参数是: b=1 {“fanMode”: “auto2”} B先执行成功;
上面方法主要是解决相邻的2个间隔短的相同key参数请求 - 伪代码:
// 并发控制, 组装请求参数的TreeSet排序的keys-1
Set propertiesKeys = new TreeSet<>();
for (Map.Entry<String, Object> entry : properties.entrySet()) {
if (entry.getValue() == null) {
continue;
}
// 并发控制, 组装请求参数的keys-2
propertiesKeys.add(entry.getKey());
recursiveCalculationAddKeys(entry.getValue(), propertiesKeys);
}
// 并发控制, 获取keys_Md5-3
String string = propertiesKeys.toString();
String md5DigestAsHex = DigestUtils.md5DigestAsHex(string.getBytes(StandardCharsets.UTF_8));
deviceStatus.setPropertiesKeysMd5(md5DigestAsHex);
request.setCompareAccessTime(Boolean.TRUE);
public static void recursiveCalculationAddKeys(Object value, Set propertiesKeys) {
if (Objects.isNull(value)){
return;
}
String type = value.getClass().getSimpleName();
if (value instanceof Map || type.endsWith(“HashMap”)) {
Map<String, Object> valueMap = (Map<String, Object>) value;
valueMap.forEach((k, v) -> {
propertiesKeys.add(k);
// 递归处理所有是map的值
recursiveCalculationAddKeys(v, propertiesKeys);
});
}
// 处理是List的情况
if (value instanceof List || type.endsWith("ArrayList")){
List<Object> valueList = (List<Object>) value;
valueList.forEach(el->{
// 递归处理所有是map的值
recursiveCalculationAddKeys(el, propertiesKeys);
});
}
}
Criteria criteriaVar2 = Criteria.where(“statusId”).is(oldDs.getStatusId());
Query queryVar2 = new Query(criteriaVar2);
if (compareAccessTime) {
// md5比较存在情况下增加查询条件&走单更新方法
criteriaVar2.orOperator(
Criteria.where(“propertiesKeysMd5”).is(propertiesKeysMd5).and(“accessTime”).lt(accessTime),
Criteria.where(“accessTime”).is(null),
Criteria.where(“propertiesKeysMd5”).is(null),
Criteria.where(“propertiesKeysMd5”).ne(propertiesKeysMd5)
); // 匹配空值
queryVar2 = new Query(criteriaVar2);
}
mongoTemplate.findAndModify(queryVar2, update, Map.class, “…”)
1