分布式锁之zookeeper

zookeeper实现分布式锁

zookeeper案例实现

1. 加入jar包依赖（使用curator框架）

<dependency>
     <groupId>org.apache.curator</groupId>
     <artifactId>curator-framework</artifactId>
     <version>2.12.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
     <artifactId>curator-recipes</artifactId>
     <version>2.12.0</version>
</dependency>

2. 配置curator

public class ZKCuratorManager {
    private static InterProcessMutex lock;
    private static CuratorFramework cf;
    private static String zkAddr = "127.0.0.1:2181";
    private static String lockPath = "/distribute-lock";
    static {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        cf = CuratorFrameworkFactory.builder()
                .connectString(zkAddr)
                .sessionTimeoutMs(2000)
                .retryPolicy(retryPolicy)
                .build();
        cf.start();
    }

    public static InterProcessMutex getLock(){
        lock = new InterProcessMutex(cf, lockPath);
        return lock;
    }
}

3. 锁的获取和释放

public class ZKCuratorLock {
    //从配置类中获取分布式锁对象
    private static InterProcessMutex lock =  ZKCuratorManager.getLock();
    //加锁
    public static boolean acquire(){
        try {
            lock.acquire();
            System.out.println(Thread.currentThread().getName() + " acquire success");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }
    //锁的释放
    public static void release(){
        try {
            lock.release();
            System.out.println(Thread.currentThread().getName() + " release success");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

4. 使用CyclicBarrier模拟并发获取分布式锁

/**
 * 使用CyclicBarrier模拟并发获取分布式锁
 */
public class CucTest {
    public static void main(String[] args) {
        int N = 4;
        CyclicBarrier barrier  = new CyclicBarrier(N);
        for(int i=0;i<N;i++){
            
            new Writer(barrier).start();
        }

        System.out.println("END");
    }
    static class Writer extends Thread{
        private CyclicBarrier cyclicBarrier;
        public Writer(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }
 
        @Override
        public void run() {
            System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据...");
            try {
                Thread.sleep(5000);      //以睡眠来模拟写入数据操作
                System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("所有线程写入完毕，继续处理其他任务...");
            //加锁
            ZKCuratorLock.acquire();
            System.out.println("线程"+ Thread.currentThread().getName() +"获得分布式锁");
            try {
                Thread.sleep(2000);
                ZKCuratorLock.release();
                System.out.println("线程"+Thread.currentThread().getName()+"释放分布式锁");
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("END");
        }
    }
}

5. 测试结果

   
   zk实现分布式锁测试结果.png

zookeeper实现原理

• 首先，zk里有一把锁，这个锁就是zk上的一个节点。然后客户端都要来获取这个锁。
• 假设客户端A先对zk发起了加分布式锁的请求，这个加锁请求是用到了zk中的一个特殊的概念，叫做“临时顺序节点”。就是直接在"my_lock"这个锁节点下，创建一个顺序节点，这个顺序节点有zk内部自行维护的一个节点序号。
• 接着他会查一下"my_lock"这个锁节点下的所有子节点，并且这些子节点是按照序号排序的，这个时候他会拿到一个集合。
• 如果是排在第一个的话，就可以加锁了。
• 客户端B过来想要加锁了，这个时候他会干一样的事儿：也先是在"my_lock"这个锁节点下创建一个临时顺序节点。
• 客户端B因为是第二个来创建顺序节点的，所以zk内部会维护序号为"2"。
• 接着客户端B会走加锁判断逻辑，查询"my_lock"锁节点下的所有子节点。同时检查自己创建的顺序节点，是不是集合中的第一个？加锁失败！
• 加锁失败以后，客户端B就会通过ZK的API，对他的上一个顺序节点加一个监听器。监听这个节点是否被删除等变化！
• 接着，客户端A加锁之后，可能处理了一些代码逻辑，然后就会释放锁。将这个节点给删除。
• 删除那个节点之后，zk会负责通知监听这个节点的监听器，也就是客户端B之前加的那个监听器，说：兄弟，你监听的那个节点被删除了，释放了锁。
• 此时客户端B的监听器感知到了上一个顺序节点被删除，也就是排在他之前的某个客户端释放了锁。

• 此时，就会通知客户端B重新尝试去获取锁，也就是获取"my_lock"节点下的子节点集合。循环以上判断。

  
  原理流程图.png