public interface lock { void getLock(); void unLock();
}
public abstract class ZkAbstractLock implements lock{
//zk连接地址
private static final String CONNECTSTRING = "127.0.0.1:2181";
//创建zk连接
protected ZkClient zkClient = new ZkClient(CONNECTSTRING);
//PATH
protected static final String PATH = "/lock";
public abstract boolean tryLock();
public abstract void waitLock();
public void getLock() {
if (tryLock()) {
System.out.println("###成功获取锁###");
} else {
waitLock();
getLock();
}
} public void unLock() {
if(zkClient!=null){
zkClient.close();
System.out.println("###释放所资源###");
}
}
}
**
* @Auther:
* @Date: 星期一:2019/4/29 15
* @Description: zk基于异常的阻塞式锁
*/
public class ZKExceptionLock extends ZkAbstractLock {
private String lockPath;
private CountDownLatch countDownLatch = new CountDownLatch(1); public ZKExceptionLock(String lockPath) {
this.lockPath = PATH + "/" + lockPath;
}
public boolean tryLock() {
try {
zkClient.createEphemeral(lockPath);
return true;
}catch (Exception e){
return false;
}
} public void waitLock() {
zkClient.subscribeDataChanges(lockPath, new IZkDataListener() {
public void handleDataChange(String s, Object o) throws Exception {
System.out.println("---handleDataChange---");
} public void handleDataDeleted(String s) throws Exception {
countDownLatch.countDown();
}
}); try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void unLock(){
zkClient.delete(lockPath);
}
}
/**
* @Auther:
* @Date: 星期一:2019/4/29 12
* @Description: zk基于监听的阻塞式锁
*/
public class ZkListenerLock extends ZkAbstractLock {
private static final CountDownLatch LATCH = new CountDownLatch(1);
private String lockPath;
private String currentPath;
private String beforePath; public ZkListenerLock(String lockPath) {
if (!zkClient.exists(PATH)) {
zkClient.createPersistent(PATH);
}
this.lockPath = PATH+"/"+lockPath;
if (!zkClient.exists(this.lockPath)) {
System.out.println("-----------");
zkClient.createPersistent(this.lockPath);
}
} public boolean tryLock() {
if (null == currentPath || currentPath.length()<=0 ) {
currentPath = zkClient.createEphemeralSequential(lockPath + "/", "lock");
}
List<String> children = zkClient.getChildren(lockPath);//0000000007
Collections.sort(children);
System.out.println(children);
if (currentPath.equals(lockPath + "/"+children.get(0))) {
return true;
}
String key = currentPath.substring(lockPath.length() + 1);
int location = Collections.binarySearch(children, key);
this.beforePath = lockPath + "/" + children.get(location - 1);
return false;
} public void waitLock() {
IZkDataListener listener = new IZkDataListener() {
public void handleDataChange(String s, Object o) throws Exception {
System.out.println("--handleDataChange--");
}
public void handleDataDeleted(String s) throws Exception {
LATCH.countDown();
}
};
zkClient.subscribeDataChanges(this.beforePath,listener);
if (!zkClient.exists(this.beforePath)) {
return;
}
try {
LATCH.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
zkClient.unsubscribeDataChanges(this.beforePath,listener);
} @Override
public void unLock() {
zkClient.delete(currentPath);
}
}
public class TestLock {
public static void main(String[] args) { int i = 0;
while (i < 10) {
People people = new People();
people.start();
i++;
}
System.out.println(People.totalMoney);
}
static class People extends Thread{
private ZkListenerLock lock = new ZkListenerLock("order");
private ZKExceptionLock lock2 = new ZKExceptionLock("order2");
public static int totalMoney = 50;
public void run(){
try {
//获取锁资源
lock2.getLock();
totalMoney--;
} catch (Exception e) {
e.printStackTrace();
} finally {
//释放锁资源
lock2.unLock();
}
}
}
}
基于监听的zk锁:L值就是建立有序节点后返回的序列号
基于异常的zk锁:
pom.xml:
<dependencies>
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.10</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.3.2</version>
</dependency>
</dependencies>