zookeeper单一节点实现分布式锁

Posted 2021-02-18 java-cxh

分布式锁的实现：注意一定要创建临时节点

package com.wiggin.lock;

import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import static org.apache.zookeeper.CreateMode.EPHEMERAL;
import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE;

/**
* 基于zk实现分布式锁
*/
public class ZkLock {

private ZooKeeper zooKeeper;

private static CountDownLatch countDownLatch = new CountDownLatch(1);

private ZkLock() {
try {
zooKeeper = new ZooKeeper("192.1.1.101:2181,192.1.1.102:2181,192.1.1.103:2181", 5000, new ZkWatcher());
System.out.println(zooKeeper.getState());
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("与zk建立连接=====>"+zooKeeper.getState());

} catch (IOException e) {
e.printStackTrace();
}
}

public static ZkLock getInstance() {
return Singleton.getInstance();
}

private class ZkWatcher implements Watcher {
@Override
public void process(WatchedEvent event) {
System.out.println("接收到监听事件=====》"+event);
if (Event.KeeperState.SyncConnected == event.getState()) {
countDownLatch.countDown();
}
}
}

public void lock(Integer id) {
String path = "/程序猴" + id;
//创建临时节点,如果创建成功的话，就表示获取锁，如果失败，则不断尝试
try {
zooKeeper.create(path,"".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("成功获取到锁");
} catch (Exception e) {
while (true) {
try {
Thread.sleep(500L);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
try {
zooKeeper.create(path,"".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
} catch (Exception e1) {
continue;
}
break;
}
}
}

/**
* 释放锁，直接删除zk节点
* @param id
*/
public void unLock(Integer id) {
String path = "/xdclass-product-lock-" + id;
try {
zooKeeper.delete(path,-1);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}

private static class Singleton {

private static ZkLock instance;
static {
instance = new ZkLock();
}

private static ZkLock getInstance() {
return instance;
}

}

}

 

分布式锁的测试：

package com.wiggin.lock;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* 线程不安全操作代码实例
*/
public class UnSafeThread {

private static int num = 0;

private static CountDownLatch countDownLatch = new CountDownLatch(10);
private static ZkLock lock = ZkLock.getInstance();

/**y
* 每次调用对num进行++操作
*/
public static void inCreate() {
lock.lock(1);
num++;
System.out.println(num);
lock.unLock(1);
}

public static void test() {
System.out.println(Thread.currentThread().getName());
}

public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(() -> {
for (int j = 0; j < 100; j++) {
inCreate();
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//每个线程执行完成之后，调用countdownLatch
countDownLatch.countDown();
}).start();
}

while (true) {
if (countDownLatch.getCount() == 0) {
System.out.println(num);
break;
}
}
}
}