Zookeeper:分布式锁案例

什么叫做分布式锁呢？

比如说"进程 1"在使用该资源的时候，会先去获得锁，"进程 1"获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源，"进程 1"用完该资源以后就将锁释放掉，让其他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

代码实现分布式锁

package com.tinstu.zk.case2;

import javafx.print.Collation;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributeLock {
    private static String connectString = "180.76.106.99:2181,120.48.39.100:2181,152.136.194.161:2181";
    private static int sessionTimeout = 200000;
    private final ZooKeeper zk;
    private CountDownLatch countDownLatch = new CountDownLatch(1);
    private CountDownLatch waitLath = new CountDownLatch(1);
    private String waitPath;
    // 当前 client 创建的子节点
    private String s;
    public DistributeLock() throws IOException, InterruptedException, KeeperException {
        //获取链接
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                //countDownLatch 如果链接上zk，可以释放
                if(watchedEvent.getState()== Event.KeeperState.SyncConnected){
                    countDownLatch.countDown();
                }
                //waitLath 需要释放
                if(watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){
                    waitLath.countDown();
                }
            }
        });
        //等待zk正常链接后，往下走程序
        countDownLatch.await();

        //判断根节点是否存在
        Stat exists = zk.exists("/locks", false);
        if (exists == null ){
            //需要创建根节点
            zk.create("/locks","locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        }

    }
    //对zk加锁
    public void zklock(){
        //创建对应的临时节点
        try {
            s = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            //判断创建的节点是否是最小的序号节点，如果是就获取到，如果不是，监听他的前一个节点.
            List<String> children = zk.getChildren("/locks", false);
            //如果children只有一个值，就直接获取，如果有多个值，需要判断，谁最小
            if(children.size()==1){
                return ;
            }else{
                //排序
                Collections.sort(children);
                //获取节点名称 seq-00000000
                String thisNode = s.substring("/locks/".length());
                //通过seq-0000000 获取节点在children集合中的位置
                int index = children.indexOf(thisNode);
                //判断
                if(index== -1){
                    System.out.println("数据异常");
                }else if(index == 0){
                    //就一个节点
                    return;
                }else{
                    //监听他前一个节点的变化
                    waitPath = "/locks/" + children.get(index - 1);
                    zk.getData(waitPath,true,null);
                    //等待监听
                    waitLath.await();
                    return;
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
   //解锁
    public void unzklock(){
        try {
            zk.delete(s,-1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
}

 测试类测试

package com.tinstu.zk.case2;

import org.apache.zookeeper.KeeperException;
import java.io.IOException;
public class DistributedLockTest {
    public static void main(String[] args) throws
            InterruptedException, IOException, KeeperException {
        // 创建分布式锁 1
        final DistributeLock lock1 = new DistributeLock();
        // 创建分布式锁 2
        final DistributeLock lock2 = new DistributeLock();
        new Thread(new Runnable() {
            @Override
            public void run() {
                // 获取锁对象
                try {
                    lock1.zklock();
                    System.out.println("线程 1 获取锁");
                    Thread.sleep(5 * 1000);
                    lock1.unzklock();
                    System.out.println("线程 1 释放锁");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                // 获取锁对象
                try {
                    lock2.zklock();
                    System.out.println("线程 2 获取锁");
                    Thread.sleep(5 * 1000);
                    lock2.unzklock();
                    System.out.println("线程 2 释放锁");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}