奇技 · 指南
Q:什么是分布式锁？
A:当多个进程在同一个系统中，用分布式锁控制多个进程对资源的访问。
分布式锁应用场景
传统的单体应用单机部署情况下，可以使用 java 并发处理相关的 API 进行互斥控制。
分布式系统后由于多线程，多进程分布在不同机器上，使单机部署情况下的并发控制锁策略失效，为了解决跨 JVM 互斥机制来控制共享资源的访问，这就是分布式锁的来源；分布式锁应用场景大都是高并发、大流量场景。
分布式锁实现
1. 加锁机制：根据 hash 节点选择一个客户端执行 lua 脚本
2. 锁互斥机制：再来一个客户端执行同样的 lua 脚本会提示已经存在锁，然后进入循环一直尝试加锁
3. 可重入机制
4. watch dog 自动延期机制
5. 释放锁机制
测试用例
单机
private RedissonClient getClient(){ Config config = new Config(); config.useSingleServer().setAddress("redis://127.0.0.1:6379");//.setPassword("");//.setConnectionMinimumIdleSize(10).setConnectionPoolSize(10);//.setConnectionPoolSize();//172.16.10.164 RedissonClient redissonClient = Redisson.create(config); return redissonClient; } private ExecutorService executorService = Executors.newCachedThreadPool(); @Test public void test() throws Exception { int[] count = {0}; for (int i = 0; i < 10; i++) { RedissonClient client = getClient(); final RedisLock redisLock = new RedisLock(client,"lock_key"); executorService.submit(() -> { try { redisLock.lock(); count[0]++; } catch (Exception e) { e.printStackTrace(); } finally { try { redisLock.unlock(); } catch (Exception e) { e.printStackTrace(); } } }); } executorService.shutdown(); executorService.awaitTermination(1, TimeUnit.HOURS); System.out.println(count[0]); }
复制代码
RedLock
public static RLock create (String url, String key){ Config config = new Config(); config.useSingleServer().setAddress(url); RedissonClient redissonClient = Redisson.create(config); return redissonClient.getLock(key); }
RedissonRedLock redissonRedLock = new RedissonRedLock( create("redis://redis://127.0.0.1:6379","lock_key1"), create("redis://redis://127.0.0.1:6380","lock_key2"), create("redis://redis://127.0.0.1:6381","lock_key3")); RedisRedLock redLock = new RedisRedLock(redissonRedLock);
private ExecutorService executorService = Executors.newCachedThreadPool();
@Test public void test() throws Exception { int[] count = {0}; for (int i = 0; i < 2; i++) { executorService.submit(() -> { try { redLock.lock(); count[0]++; } catch (Exception e) { e.printStackTrace(); } finally { try { redLock.unlock(); } catch (Exception e) { e.printStackTrace(); } } }); } executorService.shutdown(); executorService.awaitTermination(1, TimeUnit.HOURS); System.out.println(count[0]); }
复制代码
基于 ETCD 实现分布式锁分析
ETCD 分布式锁的实现
1. Lease 机制：租约机制（TTL，Time To Live），Etcd 可以为存储的 key-value 对设置租约，当租约到期，key-value 将失效删除；同时也支持续约，通过客户端可以在租约到期之前续约，以避免 key-value 对过期失效。Lease 机制可以保证分布式锁的安全性，为锁对应的 key 配置租约，即使锁的持有者因故障而不能主动释放锁，锁也会因租约到期而自动释放。
2. Revision 机制：每个 key 带有一个 Revision 号，每进行一次事务加一，它是全局唯一的，通过 Revision 的大小就可以知道进行写操作的顺序。在实现分布式锁时，多个客户端同时抢锁，根据 Revision 号大小依次获得锁，可以避免 “羊群效应” ，实现公平锁。
3. Prefix 机制：即前缀机制。例如，一个名为 /etcdlock 的锁，两个争抢它的客户端进行写操作，实际写入的 key 分别为：key1="/etcdlock/UUID1"，key2="/etcdlock/UUID2"，其中，UUID 表示全局唯一的 ID，确保两个 key 的唯一性。写操作都会成功，但返回的 Revision 不一样，那么，如何判断谁获得了锁呢？通过前缀 /etcdlock 查询，返回包含两个 key-value 对的的 KeyValue 列表，同时也包含它们的 Revision，通过 Revision 大小，客户端可以判断自己是否获得锁。
4. Watch 机制：即监听机制，Watch 机制支持 Watch 某个固定的 key，也支持 Watch 一个范围（前缀机制），当被 Watch 的 key 或范围发生变化，客户端将收到通知；在实现分布式锁时，如果抢锁失败，可通过 Prefix 机制返回的 KeyValue 列表获得 Revision 比自己小且相差最小的 key（称为 pre-key），对 pre-key 进行监听，因为只有它释放锁，自己才能获得锁，如果 Watch 到 pre-key 的 DELETE 事件则说明 pre-key 已经释放，自己已经持有锁。
步骤 1：建立连接
客户端连接 Etcd，以 /etcd/lock 为前缀创建全局唯一的 key，假设第一个客户端对应的 key="/etcd/lock/UUID1"，第二个为 key="/etcd/lock/UUID2"；客户端分别为自己的 key 创建租约 - Lease，租约的长度根据业务耗时确定；
步骤 2：创建定时任务作为租约的“心跳”
当一个客户端持有锁期间，其它客户端只能等待，为了避免等待期间租约失效，客户端需创建一个定时任务作为“心跳”进行续约。此外，如果持有锁期间客户端崩溃，心跳停止，key 将因租约到期而被删除，从而锁释放，避免死锁。
步骤 3：客户端将自己全局唯一的 key 写入 Etcd
执行 put 操作，将步骤 1 中创建的 key 绑定租约写入 Etcd，根据 Etcd 的 Revision 机制，假设两个客户端 put 操作返回的 Revision 分别为 1、2，客户端需记录 Revision 用以接下来判断自己是否获得锁
步骤 4：客户端判断是否获得锁
客户端以前缀 /etcd/lock/ 读取 keyValue 列表，判断自己 key 的 Revision 是否为当前列表中最小的，如果是则认为获得锁；否则监听列表中前一个 Revision 比自己小的 key 的删除事件，一旦监听到删除事件或者因租约失效而删除的事件，则自己获得锁。
步骤 5：执行业务
获得锁后，操作共享资源，执行业务代码
步骤 6：释放锁
完成业务流程后，删除对应的 key 释放锁
eg:
public class EtcdDistributeLock extends AbstractLock{
private Client client; private Lock lockClient; private Lease leaseClient; private String lockKey; private String lockPath; /** 锁的次数 */ private AtomicInteger lockCount; /** 租约有效期,防止客户端崩溃，可在租约到期后自动释放锁；另一方面，正常执行过程中，会自动进行续租,单位 ns */ private Long leaseTTL; /** 续约锁租期的定时任务，初次启动延迟，单位默认为 s,默认为1s，可根据业务定制设置*/ private Long initialDelay = 0L; /** 定时任务线程池类 */ ScheduledExecutorService service = null; /** 保存线程与锁对象的映射，锁对象包含重入次数，重入次数的最大限制为Int的最大值 */ private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap();
public EtcdDistributeLock(){}
public EtcdDistributeLock(Client client, String lockKey, long leaseTTL,TimeUnit unit){ this.client = client; lockClient = client.getLockClient(); leaseClient = client.getLeaseClient(); this.lockKey = lockKey; // 转纳秒 this.leaseTTL = unit.toNanos(leaseTTL); service = Executors.newSingleThreadScheduledExecutor(); }
@Override public void lock() { // 检查重入性 Thread currentThread = Thread.currentThread(); LockData oldLockData = threadData.get(currentThread); if (oldLockData != null && oldLockData.isLockSuccess()) { // re-entering int lockCount = oldLockData.lockCount.incrementAndGet(); if(lockCount < 0 ){ throw new Error("超出可重入次数限制"); } return; }
// 记录租约 ID Long leaseId = 0L; try{ leaseId = leaseClient.grant(TimeUnit.NANOSECONDS.toSeconds(leaseTTL)).get().getID(); // 续租心跳周期 long period = leaseTTL - leaseTTL / 5; // 启动定时任务续约 service.scheduleAtFixedRate(new EtcdDistributeLock.KeepAliveRunnable(leaseClient, leaseId), initialDelay,period,TimeUnit.NANOSECONDS); LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), leaseId).get(); if(lockResponse != null){ lockPath = lockResponse.getKey().toString(Charset.forName("utf-8")); log.info("获取锁成功,锁路径:{},线程:{}",lockPath,currentThread.getName()); } }catch (InterruptedException | ExecutionException e){ log.error("获取锁失败",e); return; } // 获取锁成功，锁对象设置 LockData newLockData = new LockData(currentThread, lockKey); newLockData.setLeaseId(leaseId); newLockData.setService(service); threadData.put(currentThread, newLockData); newLockData.setLockSuccess(true); }
@Override public void lockInterruptibly() throws InterruptedException { super.lockInterruptibly(); }
@Override public boolean tryLock() { return super.tryLock(); }
@Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { return super.tryLock(time,unit); }
@Override public void unlock() { Thread currentThread = Thread.currentThread(); LockData lockData = threadData.get(currentThread); if (lockData == null){ throw new IllegalMonitorStateException("You do not own the lock: " + lockKey); } int newLockCount = lockData.lockCount.decrementAndGet(); if ( newLockCount > 0 ) { return; } if ( newLockCount < 0 ) { throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + lockKey); } try { // 释放锁 if(lockPath != null){ lockClient.unlock(ByteSequence.from(lockPath.getBytes())).get(); } if(lockData != null){ // 关闭定时任务 lockData.getService().shutdown(); // 删除租约 if (lockData.getLeaseId() != 0L) { leaseClient.revoke(lockData.getLeaseId()); } } } catch (InterruptedException | ExecutionException e) { log.error("解锁失败",e); }finally { // 移除当前线程资源 threadData.remove(currentThread); } }
@Override public Condition newCondition() { return super.newCondition(); }
/** * 心跳续约线程类 */ public static class KeepAliveRunnable implements Runnable { private Lease leaseClient; private long leaseId;
public KeepAliveRunnable(Lease leaseClient, long leaseId) { this.leaseClient = leaseClient; this.leaseId = leaseId; }
@Override public void run() { // 对该leaseid进行一次续约 leaseClient.keepAliveOnce(leaseId); } }
复制代码
public class EtcdLockTest { private Client client; private String key = "/etcd/lock"; private static final String server = "http://xxxx:xxxx"; private ExecutorService executorService = Executors.newFixedThreadPool(10000);
@Before public void before() throws Exception { initEtcdClient(); }
private void initEtcdClient(){ client = Client.builder().endpoints(server).build(); }
@Test public void testEtcdDistributeLock() throws InterruptedException { int[] count = {0}; for (int i = 0; i < 100; i++) { executorService.submit(() -> { final EtcdDistributeLock lock = new EtcdDistributeLock(client, key,20,TimeUnit.SECONDS); try { lock.lock(); count[0]++; } catch (Exception e) { e.printStackTrace(); } finally { try { lock.unlock(); } catch (Exception e) { e.printStackTrace(); } } }); } executorService.shutdown(); executorService.awaitTermination(1, TimeUnit.HOURS); System.err.println("执行结果: " + count[0]); }}
复制代码
实现原理
1. 启动客户端，确认链接到了服务器
2. 多个客户端并发的在特定路径下创建临时性顺序节点
3. 客户端判断自己的创建的顺序节点是否是最小的，如果是最小的，则获取锁成功
4. 第三步若判定失败，则采用 zk 的 watch 机制监听自己的前一个顺序节点，等待前一个节点的删除（放锁）事件，再开始第三步判定
zookeeper 作为高性能分布式协调框架，可以把其看做一个文件系统，其中有节点的概念，并且分为 4 种：1.持久性节点 2.持久性顺序节点 3.临时性节点 4.临时性顺序节点。
分布式锁的实现主要思路就是：监控其他客户端的状态，来判断自己是否可以获得锁。
采用临时性顺序节点的原因：
1. zk 服务器维护了客户端的会话有效性，当会话失效的时候，其会话所创建的临时性节点都会被删除，通过这一特点，可以通过 watch 临时节点来监控其他客户端的情况，方便自己做出相应动作。
2. 因为 zk 对写操作是顺序性的，所以并发创建的顺序节点会有一个唯一确定的序号，当前锁是公平锁的一种实现，所以依靠这种顺序性可以很好的解释—节点序列小的获取到锁并且可以采用 watch 自己的前一个节点来避免惊群现象（这样 watch 事件的传播是线性的）。
eg:
public class ZKLock extends AbstractLock {
/** * 1.Connect to zk */ private CuratorFramework client;
private InterProcessLock lock ;
public ZKLock(String zkAddress,String lockPath) { // 1.Connect to zk client = CuratorFrameworkFactory.newClient( zkAddress, new RetryNTimes(5, 5000) ); client.start(); if(client.getState() == CuratorFrameworkState.STARTED){ log.info("zk client start successfully!"); log.info("zkAddress:{},lockPath:{}",zkAddress,lockPath); }else{ throw new RuntimeException("客户端启动失败。。。"); } this.lock = defaultLock(lockPath); }
private InterProcessLock defaultLock(String lockPath ){ return new InterProcessMutex(client, lockPath); } @Override public void lock() { try { this.lock.acquire(); } catch (Exception e) { throw new RuntimeException(e); } }
@Override public boolean tryLock() { boolean flag ; try { flag=this.lock.acquire(0,TimeUnit.SECONDS); } catch (Exception e) { throw new RuntimeException(e); } return flag; }
@Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { boolean flag ; try { flag=this.lock.acquire(time,unit); } catch (Exception e) { throw new RuntimeException(e); } return flag; }
@Override public void unlock() { try { this.lock.release(); } catch (Exception e) { throw new RuntimeException(e); } }
}
复制代码
private ExecutorService executorService = Executors.newCachedThreadPool();
@Test public void testLock() throws Exception{ ZKLock zkLock = new ZKLock("xxxx:xxxx","/lockPath"); int[] num = {0}; long start = System.currentTimeMillis(); for(int i=0;i<200;i++){ executorService.submit(()->{ try { zkLock.lock(); num[0]++; } catch (Exception e){ throw new RuntimeException(e); } finally { zkLock.unlock(); } });
} executorService.shutdown(); executorService.awaitTermination(1, TimeUnit.HOURS); log.info("耗时:{}",System.currentTimeMillis()-start); System.out.println(num[0]); }
复制代码
总结
redis 的分布式锁中 redisson 一般为单实例，当单实例不可用时，会阻塞业务流程。主从方式、主从数据异步，会存在锁失效的问题。RedLock 一般要求至少 3 台以上的 redis 主从实例，维护成本相对来说比较高。
2. ZK 锁具备高可用、可重入、阻塞锁特性，可解决失效死锁问题。但是因为需要频繁的创建和删除节点，性能上不如 Redis 方式。
3. ETCD 分布式锁的实现原理与 zk 锁类似，但是 ETCD 分布式锁更加可靠强大。其 Lease 功能保证分布式锁的安全性；watch 功能支持监听某个固定的 key，也支持 watch 一个范围的 key（前缀机制）；revision 功能可通过 Revision 的大小就可以知道进行写操作的顺序。可以避免 “羊群效应” （也称 “惊群效应”），实现公平锁。前缀机制与 watch 功能配合使用解决了死锁问题。总之 ETCD 的灵感来源于 Zookeeper,但实现的时候做了很多的改进，如：高负载下的稳定读写、数据模型的多版本并发控制、稳定的 watch 功能，通知订阅者监听值得变化、可以容忍脑裂现场的发生、客户端的协议使用 gRPC 协议,支持 go、c++、java 等。
原文链接：分布式锁的实现方案
评论