R2M分布式锁原理及实践

作者：京东科技 张石磊
1 案例引入

名词简介：
资源：可以理解为一条内容，或者图+文字+链接的载体。
档位ID: 资源的分类组，资源必须归属于档位。
问题描述：当同一个档位下2条资源同时审批通过时，收到擎天审批系统2条消息，消费者应用部署了2台机器，此时正好由2台机器分别消费，在并发消费时，先更新资源状态，然后写缓存，每次取前100条资源，类似select * from resource where gear_id=xxx limit 100 order by id desc;
在写档位缓存，此时事务未提交，并发查询时根据档位Id查询时查询不到对方的数据，全量写缓存时导致后写的缓存覆盖了先写的缓存，即缓存被覆盖，导致投放资源缺失。
方案思考 ：
方案1：一台机器消费mq–单点问题
方案2：将同档位ID的资源路由到同一个queue,需要审批系统配合根据档位Id做路由，审批系统发的消息不只是cms审批数据，此方案不适用。
方案3：在档位级别加分布式锁。
经比较，最终采用方案3是合适的方案.
2 锁说明和分布式锁选择

synchronized锁的粒度是JVM进程维度，集群模式下，不能对共享资源加锁，此时需要跨JVM进程的分布式锁。
分布式锁方式核心实现方式优点缺点分析
1 数据库：
悲观锁，lock
乐观锁，通过版本号实现version
实现简单，不依赖中间件
数据库IO瓶颈，性能差
单实例存在单点问题，主从架构存在数据不一致，主从切换时其他客户端可重复加锁。
2 zookeeper
创建临时节点
CP模型，可靠性高，不存在主从切换不一致问题
频繁创建和销毁临时节点，且
集群模式下，leader数据需要同步到follower才算加锁成功，性能不如redis
主从切换服务不可用
3 redis集群
setnx+expire
性能高
有封装好的框架redission
支持超时自动删除锁
集群支持高可用，AP模型
主从切换时其他客户端可重复加锁。
R2M是基于开源的Redis cluster(Redis 3.0以上版本)构建的高性能分布式缓存系统，我们系统一直在使用，3.2.5版本开始支持分布式锁。
3 r2m分布式锁原理

示例代码：

1. String lockKey = CacheKeyHelper.getGearLockKey(EnvEnum.getEnvFlagEnum(envCode),resource.getGearId());
3. R2mLock lock = (R2mLock) r2mClusterClient.getLock(lockKey);
5. //获取锁，锁的默认有效期30s,获取不到锁就阻塞
7. lock.lock();
9. try {
11.     //业务代码
13.     resourceService.afterApprovedHandle(resource);
15. }  finally {
17.     //释放锁
19.     lock.unlock();
21. }

复制代码

1 加锁核心流程：

加锁流程图：

1）：尝试获取锁，通过执行加锁Lua脚本来做；
2）：若第一步未获取到锁，则去redis订阅解锁消息
3）：一旦持有锁的线程释放了锁，就会广播解锁消息，其他线程自旋重新尝试获取锁。
核心加锁原理：使用lua脚本封装了hset和pexpire命令，保证是一个原子操作， KEYS[1]是加锁的key,argv[2]是加锁的客户端ID(UUID+线程ID),ARGV[1]是锁的有效期，默认30s.

1. private Object acquireInternal(List<String> args) {
3. if (!this.setLocked() && this.getHolderId() != Thread.currentThread().getId()) {
5. return -1L;
7. } else {
9. try {
11. //hash结构，hash的key是加锁的key,键值对的key为客户端的UUID+线程id,value为锁的重入计数器值。
13. return this.lockSha() != null ? this.executor.evalR2mLockSha(this.lockSha(),
15. "if (redis.call('exists', KEYS[1]) == 0) then redis.call('hset', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end;
17. if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end;
19. return -2;", Collections.singletonList(this.lockName), args) : this.executor. == 0) then redis.call('hset', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; return -2;", Collections.singletonList(this.lockName), args);
21. } catch (Exception var3) {
23. this.setUnlocked();
25. throw new R2mLockException("Failed to acquire lock " + this.lockName + ".", var3);
27. }
29. }
31. }

复制代码

args参数

1. private List<String> acquireArgs(long leaseTime) {
3.         List<String> args = new ArrayList();
5.         if (leaseTime != -1L) {
7.             args.add(String.valueOf(leaseTime));
9.         } else {
11.              //默认30s
13.             args.add(String.valueOf(this.internalLockLeaseTime()));
15.         }
17.         //UUID+当前线程id
19.         args.add(this.currentThreadLockId(Thread.currentThread().getId()));
21.         return args;
23.     }
25. 

复制代码

获取锁失败订阅锁的channel

1. //获取锁失败，订阅释放锁的消息
3. private boolean failedAcquire() {
5.             this.subLock();
7.             return false;
9.         }
13. private void subLock() {
15.             CompletableFuture<Void> cf = R2mLock.this.executor.lockSubscribe(R2mLock.this.lockPubSub(), R2mLock.this.getLockChannelName(), R2mLock.this);
17.             if (cf != null) {
19.                 cf.handleAsync(this::reSubIfEx);
21.             }
25.         }

复制代码

锁释放后，订阅者通过自旋尝试获取锁。

1. //tryAcquire获取锁，！tryAcquire就是获取锁失败，锁释放后，通知线程唤醒后返回false,然后通过自旋，尝试获取锁，
3. public final void acquire(long arg) {
5.         if (!tryAcquire(arg) &&
7.             acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
9.             selfInterrupt();
11.     }
15. final boolean acquireQueued(final Node node, long arg) {
17.         boolean failed = true;
19.         try {
21.             boolean interrupted = false;
23.             //内部自旋获取锁
25.             for (;;) {
27.                 final Node p = node.predecessor();
29.                 if (p == head && tryAcquire(arg)) {
31.                     setHead(node);
33.                     p.next = null; // help GC
35.                     failed = false;
37.                     return interrupted;
39.                 }
41.                 if (shouldParkAfterFailedAcquire(p, node) &&
43.                     parkAndCheckInterrupt())
45.                     interrupted = true;
47.             }
49.         } finally {
51.             if (failed)
53.                 cancelAcquire(node);
55.         }
57.     }

复制代码

2 释放锁核心逻辑：

1）删除分布式锁key（如果可重入锁计数为0）

• 发释放锁的广播消息
  

3）取消watchdog

1. private Object unlockInternal(List<String> args) {
3.         logger.debug("{} trying to unlock.", Thread.currentThread().getId());
7.         Object var2;
9.         try {
11.      //判断锁 key 是否存在，如果存在，然后递减hash的value值，当value值为0时再删除锁key，并且广播释放锁的消息
13.             if (this.unlockSha() == null) {
15.                 var2 = this.executor. == 0) then return nil;end; local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); if (counter > 0) then return 0; else redis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]); return 1; end; return nil;", Arrays.asList(this.lockName, this.getLockChannelName()), args);
17.                 return var2;
19.             }
23.             var2 = this.executor.evalR2mLockSha(this.unlockSha(), "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil;end; local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); if (counter > 0) then return 0; else redis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]); return 1; end; return nil;", Arrays.asList(this.lockName, this.getLockChannelName()), args);
25.         } catch (Exception var6) {
27.             throw new R2mLockException("Failed to unlock " + this.lockName + ".", var6);
29.         } finally {
31.             this.finalizeRelease();
33.         }
37.         return var2;
39.     }
43. //取消当前线程的watchdog
45. private void finalizeRelease() {
47.         long threadId = Thread.currentThread().getId();
49.         R2mLock.ExpirableEntry entry = (R2mLock.ExpirableEntry)this.entryCache.get(threadId);
51.         if (entry != null) {
53.             entry.release(threadId);
55.             if (entry.isReleased()) {
57.                 //取消这个线程watchdog定时任务
59.                 entry.getExtTask().cancel();
61.                 this.expEntry.compareAndSet(entry, (Object)null);
63.                 //从缓存watchdog线程的map中删除该线程
65.                 this.entryCache.remove(threadId);
67.             }
69.         }
73.     }

复制代码

3 锁的健壮性思考

1 业务没执行完，锁超时过期怎么办？
客户端加锁默认有效期30s，超过有效期后如果业务没执行完，还需要持有这把锁，r2m客户端提供了续期机制，也就是watchdog机制。
watchdog原理：客户端线程维度（UUID+线程ID,客户端维护一个MAP,key就是UUID+线程ID）的后台定时线程，获取锁成功后，如果客户端还持有当前锁，每隔10s(this.internalLockLeaseTime() / 3L)，去延长锁的有效期（internalLockLeaseTime）

1. //watchdog核心机制 ，internalLockLeaseTime默认30s
3. private void extendLock(long holderId) {
5.         if (this.expEntry.get() != null) {
7.             R2mLock.ExpirableEntry holderEntry = (R2mLock.ExpirableEntry)this.entryCache.get(holderId);
9.             if (holderEntry != null) {
11.                  //每隔10s,如果当前线程持有锁，则续期30s
13.                 if (this.expEntry.compareAndSet(holderEntry, holderEntry)) {
15.                     Timeout task = this.timer().newTimeout((timeout) -> {
17.                         if (this.extendLockInternal(holderId)) {
19.                             this.extendLock(holderId);
21.                         }
25.                     }, this.internalLockLeaseTime() / 3L, TimeUnit.MILLISECONDS);
27.                     if (this.expEntry.get() != null) {
29.                         ((R2mLock.ExpirableEntry)this.expEntry.get()).setExtTask(task);
31.                     }
33.                 }
37.             }
39.         }
41.     }
45. //执行续期lua脚本
47. private boolean extendLockInternal(long threadId) {
49.         Object result;
51.         try {
53.            //只续期
55.             if (this.extendLockSha() != null) {
57.                 result = this.executor.evalR2mLockSha(this.extendLockSha(), "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('pexpire', KEYS[1], ARGV[1]); return 1; end; return 0;", Collections.singletonList(this.lockName), this.extendLockArgs(threadId));
59.             } else {
61.                 result = this.executor. == 1) then redis.call('pexpire', KEYS[1], ARGV[1]); return 1; end; return 0;", Collections.singletonList(this.lockName), this.extendLockArgs(threadId));
63.             }
65.         } catch (Exception var5) {
67.             return false;
69.         }
73.         return Long.parseLong(result.toString()) == 1L;
75.     }

复制代码

2 客户端宕机，锁如何释放？
分布式锁是有效期的，客户端宕机后，watchdog机制失效，锁过期自动失效。
3 redis分布式锁集群模式下缺陷
r2m集群模式，极端情况，master加锁成功，宕机，还未来得及同步到slave,主从切换，slave切换成master,可以继续加锁，对于非及其严格加锁场景，该方案可满足，属于AP；对于严格场景下的分布式锁，可采用基于zookeeper的分布式锁，属于CP，leader宕机，folllower选举时不可用。性能上redis更优。
4 锁的释放问题
注意锁的释放在finally中释放，必须由锁的持有者释放，不能由其他线程释放别人的锁，示例代码中lock放到try的外面。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！