网站建设需求调研问卷,网站定制开发加公众号,付费ppt模板网站哪个好,旅游网站建设的目的及功能定位简介#xff1a;本文从钉钉实人认证场景的一例数据重复问题出发#xff0c;分析了其原因是因为并发导致幂等失效#xff0c;引出幂等的概念。针对并发场景下的幂等问题#xff0c;提出了一种实现幂等可行的方法论#xff0c;结合通讯录加人业务场景对数据库幂等问题进行了…简介本文从钉钉实人认证场景的一例数据重复问题出发分析了其原因是因为并发导致幂等失效引出幂等的概念。针对并发场景下的幂等问题提出了一种实现幂等可行的方法论结合通讯录加人业务场景对数据库幂等问题进行了简单分析就分布式锁实现幂等方法展开了详细讨论。分析了锁在分布式场景下存在的问题包括单点故障、网络超时、错误释放他人锁、提前释放锁以及分布式锁单点故障等提出了对应的解决方案介绍了对应方案的具体实现。 作者 | 百书 来源 | 阿里技术公众号
写在前面本文讨论的幂等问题均为并发场景下的幂等问题。即系统本存在幂等设计但是在并发场景下失效了。一 摘要
本文从钉钉实人认证场景的一例数据重复问题出发分析了其原因是因为并发导致幂等失效引出幂等的概念。
针对并发场景下的幂等问题提出了一种实现幂等可行的方法论结合通讯录加人业务场景对数据库幂等问题进行了简单分析就分布式锁实现幂等方法展开了详细讨论。
分析了锁在分布式场景下存在的问题包括单点故障、网络超时、错误释放他人锁、提前释放锁以及分布式锁单点故障等提出了对应的解决方案介绍了对应方案的具体实现。
二 问题
钉钉实人认证业务存在数据重复的问题。
1 问题现象
正常情况下数据库中应该只有一条实人认证成功记录但是实际上某用户有多条。 2 问题原因
并发导致了不幂等。
我们先来回顾一下幂等的概念
幂等idempotent、idempotence是一个数学与计算机学概念常见于抽象代数中。 在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。 --来自百度百科实人认证在业务上有幂等设计其一般流程为
1用户选择实人认证后会在服务端初始化一条记录 2用户在钉钉移动端按照指示完成人脸比对 3比对完成后访问服务端修改数据库状态。
在第3步中在修改数据库状态之前会判断「是否已经初始化」、「是否已经实人认证」以及「智科是否返回认证成功」以保证幂等。仅当请求首次访问服务端尝试修改数据库状态时才能满足幂等的判断条件并修改数据库状态。其余任意次请求将直接返回对数据库状态无影响。请求多次访问服务端所产生的结果和请求首次访问服务端一致。因此在实人认证成功的前提下数据库应当有且仅有一条认证成功的记录。 但是在实际过程中我们发现同一个请求会多次修改数据库状态系统并未按照我们预期的那样实现幂等。究其原因是因为请求并发访问在首次请求完成修改服务端状态前并发的其他请求和首次请求都通过了幂等判断对数据库状态进行了多次修改。
并发导致了原幂等设计失效。
并发导致了不幂等。
三 解决方案
解决并发场景下幂等问题的关键是找到唯一性约束执行唯一性检查相同的数据保存一次相同的请求操作一次。
一次访问服务端的请求可能产生以下几种交互
与数据源交互例如数据库状态变更等与其他业务系统交互例如调用下游服务或发送消息等
一次请求可以只包含一次交互也可以包含多次交互。例如一次请求可以仅仅修改一次数据库状态也可以在修改数据库状态后再发送一条数据库状态修改成功的消息。
于是我们可以得出一个结论并发场景下如果一个系统依赖的组件幂等那么该系统在天然幂等。
以数据库为例如果一个请求对数据造成的影响是新增一条数据那么唯一索引可以是幂等问题的解法。数据库会帮助我们执行唯一性检查相同数据不会重复落库。
钉钉通讯录加人就是通过数据库的唯一索引解决了幂等问题。以钉钉通讯录加人为例在向数据库写数据之前会先判断数据是否已经存在于数据库之中如果不存在加人请求最终会向数据库的员工表插入一条数据。大量相同的并发的通讯录加人请求让系统的幂等设计失效成为可能。在一次加人请求中组织ID工号可以唯一标记一个请求在数据库中也存在组织ID工号的唯一索引。因此我们可以保证多次相同的加人请求只会修改一次数据库状态即添加一条记录。
如果所依赖的组件天然幂等那么问题就简单了但是实际情况往往更加复杂。并发场景下如果系统依赖的组件无法幂等我们就需要使用额外的手段实现幂等。
一个常用的手段就是使用分布式锁。分布式锁的实现方式有很多比较常用的是缓存式分布式锁。
四 分布式锁
在What is a Java distributed lock?中有这样几段话
In computer science, locks are mechanisms in a multithreaded environment to prevent different threads from operating on the same resource. When using locking, a resource is locked for access by a specific thread, and can only be accessed by a different thread once the resource has been released. Locks have several benefits: they stop two threads from doing the same work, and they prevent errors and data corruption when two threads try to use the same resource simultaneously. Distributed locks in Java are locks that can work with not only multiple threads running on the same machine, but also threads running on clients on different machines in a distributed system. The threads on these separate machines must communicate and coordinate to make sure that none of them try to access a resource that has been locked up by another. 这几段话告诉我们锁的本质是共享资源的互斥访问分布式锁解决了分布式系统中共享资源的互斥访问的问题。
java.util.concurrent.locks包提供了丰富的锁实现包括公平锁/非公平锁阻塞锁/非阻塞锁读写锁以及可重入锁等。
我们要如何实现一个分布式锁呢
方案一
分布式系统中常见有两个问题
1单点故障问题即当持有锁的应用发生单点故障时锁将被长期无效占有 2网络超时问题即当客户端发生网络超时但实际上锁成功时我们无法再次正确的
获取锁。
要解决问题1一个简单的方案是引入过期时间lease time对锁的持有将是有时效的当应用发生单点故障时被其持有的锁可以自动释放。
要解决问题2一个简单的方案是支持可重入我们为每个获取锁的客户端都配置一个不会重复的身份标识通常是UUID上锁成功后锁将带有该客户端的身份标识。当实际上锁成功而客户端超时重试时我们可以判断锁已被该客户端持有而返回成功。
综上我们给出了一个lease-based distribute lock方案。出于性能考量使用缓存作为锁的存储介质利用MVCCMultiversion concurrency control机制解决共享资源互斥访问问题具体实现可见附录代码。
分布式锁的一般使用方式如下
● 初始化分布式锁的工厂 ● 利用工厂生成一个分布式锁实例 ● 使用该分布式实例上锁和解锁操作
Test
public void testTryLock() {//初始化工厂MdbDistributeLockFactory mdbDistributeLockFactory new MdbDistributeLockFactory();mdbDistributeLockFactory.setNamespace(603);mdbDistributeLockFactory.setMtairManager(new MultiClusterTairManager());//获得锁DistributeLock lock mdbDistributeLockFactory.getLock(TestLock);//上锁解锁操作boolean locked lock.tryLock();if (!locked) {return;}try {//do something } finally {lock.unlock();}
}
该方案简单易用但是问题也很明显。例如释放锁的时候只是简单的将缓存中的key失效所以存在错误释放他人已持有锁问题。所幸只要锁的租期设置的足够长该问题出现几率就足够小。
我们借用Martin Kleppmann在文章How to do distributed locking中的一张图说明该问题。 设想一种情况当占有锁的Client 1在释放锁之前锁就已经到期了Client 2将获取锁此时锁被Client 2持有但是Client 1可能会错误的将其释放。一个更优秀的方案我们给每个锁都设置一个身份标识在释放锁的时候1首先查询锁是否是自己的2如果是自己的则释放锁。受限于实现方式步骤1和步骤2不是原子操作在步骤1和步骤2之间如果锁到期被其他客户端获取此时也会错误的释放他人的锁。
方案二
借助Redis的Lua脚本可以完美的解决存在错误释放他人已持有锁问题的。在Distributed locks with Redis这篇文章的 Correct implementation with a single instance 这一节中我们可以得到我们想要的答案——如何实现一个分布式锁。
当我们想要获取锁时我们可以执行如下方法
SET resource_name my_random_value NX PX 30000
当我们想要释放锁时我们可以执行如下的Lua脚本
if redis.call(get,KEYS[1]) ARGV[1] thenreturn redis.call(del,KEYS[1])
elsereturn 0
end
方案三
在方案一和方案二的讨论过程中有一个问题被我们反复提及锁的自动释放。
这是一把双刃剑
1一方面它很好的解决了持有锁的客户端单点故障的问题 2另一方面如果锁提前释放就会出现锁的错误持有状态
这个时候我们可以引入Watch Dog自动续租机制我们可以参考以下Redisson是如何实现的。
在上锁成功后Redisson会调用renewExpiration()方法开启一个Watch Dog线程为锁自动续期。每过1/3时间续一次成功则继续下一次续期失败取消续期操作。
我们可以再看看Redisson是如何续期的。renewExpiration()方法的第17行renewExpirationAsync()方法是执行锁续期的关键操作我们进入到方法内部可以看到Redisson也是使用Lua脚本进行锁续租的1判断锁是否存在2如果存在则重置过期时间。
private void renewExpiration() {ExpirationEntry ee EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee null) {return;}Timeout task commandExecutor.getConnectionManager().newTimeout(timeout - {ExpirationEntry ent EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ent null) {return;}Long threadId ent.getFirstThreadId();if (threadId null) {return;}RFutureBoolean future renewExpirationAsync(threadId);future.onComplete((res, e) - {if (e ! null) {log.error(Cant update lock getRawName() expiration, e);EXPIRATION_RENEWAL_MAP.remove(getEntryName());return;}if (res) {// reschedule itselfrenewExpiration();} else {cancelExpirationRenewal(null);}});}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}
protected RFutureBoolean renewExpirationAsync(long threadId) {return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,if (redis.call(hexists, KEYS[1], ARGV[2]) 1) then redis.call(pexpire, KEYS[1], ARGV[1]); return 1; end; return 0;,Collections.singletonList(getRawName()),internalLockLeaseTime, getLockName(threadId));
}
方案四
借助Redisson的自动续期机制我们无需再担心锁的自动释放。但是讨论到这里我还是不得不面对一个问题分布式锁本身不是一个分布式应用。当Redis服务器故障无法正常工作时整个分布式锁也就无法提供服务。
更进一步我们可以看看Distributed locks with Redis这篇文章中提到的Redlock算法及其实现。
Redlock算法不是银弹关于它的好与坏也有很多争论
How to do distributed lockingHow to do distributed locking — Martin Kleppmann’s blog
Is Redlock safe?Is Redlock safe? - antirez
Martin Kleppmann和Antirez关于Redlock的争辩https://news.ycombinator.com/item
参考资料
What is a Java distributed lock?What is a Java distributed lock? | Redisson
Distributed locks and synchronizershttps://github.com/redisson/redisson/wiki/8.-distributed-locks-and-synchronizers
Distributed locks with RedisDistributed locks with Redis – Redis
附录
分布式锁
public class MdbDistributeLock implements DistributeLock {/*** 锁的命名空间*/private final int namespace;/*** 锁对应的缓存key*/private final String lockName;/*** 锁的唯一标识保证可重入以应对put成功但是返回超时的情况*/private final String lockId;/*** 是否持有锁。true是*/private boolean locked;/*** 缓存实例*/private final TairManager tairManager;public MdbDistributeLock(TairManager tairManager, int namespace, String lockCacheKey) {this.tairManager tairManager;this.namespace namespace;this.lockName lockCacheKey;this.lockId UUID.randomUUID().toString();}Overridepublic boolean tryLock() {try {//获取锁状态ResultDataEntry getResult null;ResultCode getResultCode null;for (int cnt 0; cnt DEFAULT_RETRY_TIMES; cnt) {getResult tairManager.get(namespace, lockName);getResultCode getResult null ? null : getResult.getRc();if (noNeedRetry(getResultCode)) {break;}}//重入已持有锁返回成功if (ResultCode.SUCCESS.equals(getResultCode) getResult.getValue() ! null lockId.equals(getResult.getValue().getValue())) {locked true;return true;}//不可获取锁返回失败if (!ResultCode.DATANOTEXSITS.equals(getResultCode)) {log.error(tryLock fail code{} lock{} traceId{}, getResultCode, this, EagleEye.getTraceId());return false;}//尝试获取锁ResultCode putResultCode null;for (int cnt 0; cnt DEFAULT_RETRY_TIMES; cnt) {putResultCode tairManager.put(namespace, lockName, lockId, MDB_CACHE_VERSION,DEFAULT_EXPIRE_TIME_SEC);if (noNeedRetry(putResultCode)) {break;}}if (!ResultCode.SUCCESS.equals(putResultCode)) {log.error(tryLock fail code{} lock{} traceId{}, getResultCode, this, EagleEye.getTraceId());return false;}locked true;return true;} catch (Exception e) {log.error(DistributedLock.tryLock fail lock{}, this, e);}return false;}Overridepublic void unlock() {if (!locked) {return;}ResultCode resultCode tairManager.invalid(namespace, lockName);if (!resultCode.isSuccess()) {log.error(DistributedLock.unlock fail lock{} resultCode{} traceId{}, this, resultCode,EagleEye.getTraceId());}locked false;}/*** 判断是否需要重试** param resultCode 缓存的返回码* return true不用重试*/private boolean noNeedRetry(ResultCode resultCode) {return resultCode ! null !ResultCode.CONNERROR.equals(resultCode) !ResultCode.TIMEOUT.equals(resultCode) !ResultCode.UNKNOW.equals(resultCode);}}
分布式锁工厂
public class MdbDistributeLockFactory implements DistributeLockFactory {/*** 缓存的命名空间*/Setterprivate int namespace;Setterprivate MultiClusterTairManager mtairManager;Overridepublic DistributeLock getLock(String lockName) {return new MdbDistributeLock(mtairManager, namespace, lockName);}
}
原文链接 本文为阿里云原创内容未经允许不得转载。
