# zkDisLock

**Repository Path**: kakalex/zkDisLock

## Basic Information

- **Project Name**: zkDisLock
- **Description**: 使用zookeeper实现分布式锁
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2020-12-10
- **Last Updated**: 2020-12-19

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

基于zookeeper实现分布式锁

zk有两种数据节点，一种时持久节点

另一种时瞬时节点，有序，瞬时节点不可再有子节点，会话结束后瞬时节点自动消失


zookeeper的观察器

可以设置观察器的三个方法：getData();getChildren();exist();

数据节点发生变化，发送给客户端

观察器只能监控一次，再监控需要重新设置


zk分布式锁的实现原理

利用zookeeper的瞬时有序节点的特性

多线程并发创建瞬时节点时得到有序的序列

序号最小的线程获得锁

其他的线程则监听自己序号的前一个序号

前一个线程执行完成，删除自己序号的节点

下一个序号的线程得到通知继续执行

实现代码
```java
@Slf4j
public class ZKLock implements AutoCloseable, Watcher {

    private ZooKeeper zooKeeper;
    private String znode;

    public ZKLock() throws IOException {
        // 建立连接
        this.zooKeeper = new ZooKeeper("192.168.1.4:2181", 10000, this);
    }

    public boolean getLock(String businessCode) {
        try {
            // 检查业务根节点是否存在
            Stat stat = zooKeeper.exists("/" + businessCode, false);
            if (stat == null) {
                //创建业务根节点
                zooKeeper.create("/" + businessCode, businessCode.getBytes(),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            //创建瞬时有序节点 /order/orde_00000001
            znode = zooKeeper.create("/" + businessCode + "/" + businessCode + "_", businessCode.getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            List<String> childrenNodes = zooKeeper.getChildren("/" + businessCode, false);
            Collections.sort(childrenNodes);
            String firstNode = childrenNodes.get(0);
            if (znode.endsWith(firstNode)) {
                return true;
            }
            // 不是第一个节点, 则监听前一个节点
            String lastNode = firstNode;
            for (String node : childrenNodes) {
                if (znode.endsWith(node)) {
                    Stat exists = zooKeeper.exists("/" + businessCode + "/" + lastNode, true);
                    break;

                }else {
                    lastNode = node;
                }
            }
            synchronized (this) {
                wait();
            }
            return true;

        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }


    @Override
    public void close() throws Exception {
        zooKeeper.delete(znode, -1);
        zooKeeper.close();
        log.info("我已经释放了锁");
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.NodeDeleted) {
            synchronized (this) {
                notify();
            }
        }
    }
}
```

在业务代码中使用zk分布式锁

```java
@Service
@Slf4j
public class ZKLockService {
    @Autowired
    private CuratorFramework client;

    public String zkLock() {
        log.info("我进入了方法！");
        try (ZKLock zkLock = new ZKLock()) {
            if (zkLock.getLock("order")) {
                Thread.sleep(10000);
                return "ok";
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("方法执行完成");
        return "fail";
    }
}
```

zk curator客户端实现分布式锁

curator客户端直接实现了分布式锁的方法，直接调用就可以

在spring中配置curator客户端
```java
    @Bean(initMethod="start",destroyMethod = "close")
    public CuratorFramework getCuratorFramework() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.1.4:2181", retryPolicy);
        return client;
    }
```

在业务代码中使用

```java
@Service
@Slf4j
public class ZKLockService {
    @Autowired
    private CuratorFramework client;

    public String curatorLock(){
        log.info("我进入了方法！");
        InterProcessMutex lock = new InterProcessMutex(client, "/order");
        try{
            if (lock.acquire(30, TimeUnit.SECONDS)){
                log.info("我获得了锁！！");
                Thread.sleep(10000);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            try {
                log.info("我释放了锁！！");
                lock.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        log.info("方法执行完成！");
        return "方法执行完成！";
    }
}

```