分布式锁的实现方案

 

什么是分布式锁

当多个进程在同一个系统中，用分布式锁控制多个进程对资源的访问

分布式锁应用场景

1. 传统的单体应用单机部署情况下，可以使用java并发处理相关的API进行互斥控制。
2. 分布式系统后由于多线程，多进程分布在不同机器上，使单机部署情况下的并发控制锁策略失效，为了解决跨JVM互斥机制来控制共享资源的访问，这就是分布式锁的来源；分布式锁应用场景大都是高并发、大流量场景。

分布式锁实现

1、基于redis的分布式锁

redis分布式锁的实现

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();
    ----------------------------------------------------------------
      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();
                    }
                }
            });

 

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();
    
    ------------------------------------------------------------------
     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();
                    }
                }
            });
        }

 

redis源码分析

redisson

public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
        //获取当前线程id
        long threadId = Thread.currentThread().getId();
        //尝试获得锁，返回还剩余的锁过期时间
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        //如果ttl为空，代表当前没有锁，获取成功
        if (ttl == null) {
            return;
        }
        
        //如果获取锁失败，则订阅到对应这个锁的channel，一旦其他线程释放锁时，通知线程去获取锁
        RFuture<RedissonLockEntry> future = subscribe(threadId);
        commandExecutor.syncSubscription(future);

        try {
            //循环等待
            while (true) {
                //尝试获得锁
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message //ttl大于0，则等待ttl时间后继续尝试获取锁
                if (ttl >= 0) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().acquire();
                }
            }
        } finally {
            unsubscribe(future, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }

 

2、基于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 已经释放，自己已经持有锁
   

    

    

    

基于ETCD分布式锁

步骤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释放锁

测试用例

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]);
    }
}

 

3、基于Zookeeper分布式锁

实现原理

1. 启动客户端，确认链接到了服务器
2. 多个客户端并发的在特定路径下创建临时性顺序节点
3. 客户端判断自己的创建的顺序节点是否是最小的，如果是最小的，则获取锁成功
4. 第三步若判定失败，则采用zk的watch机制监听自己的前一个顺序节点，等待前一个节点的删除（放锁）事件，再开始第三步判定。
   
   

 

 

zookeeper作为高性能分布式协调框架，可以把其看做一个文件系统，其中有节点的概念，并且分为4种：1.持久性节点2.持久性顺序节点3.临时性节点4.临时性顺序节点。
分布式锁的实现主要思路就是：监控其他客户端的状态，来判断自己是否可以获得锁。
采用临时性顺序节点的原因：

1. zk服务器维护了客户端的会话有效性，当会话失效的时候，其会话所创建的临时性节点都会被删除，通过这一特点，可以通过watch临时节点来监控其他客户端的情况，方便自己做出相应动作。
2. 因为zk对写操作是顺序性的，所以并发创建的顺序节点会有一个唯一确定的序号，当前锁是公平锁的一种实现，所以依靠这种顺序性可以很好的解释—节点序列小的获取到锁并且可以采用watch自己的前一个节点来避免惊群现象（这样watch事件的传播是线性的）。

测试用例

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]);
    }