思路:提供一把全局的锁,所有来购买的请求竞争这一把锁,谁先拿到这把锁,谁就有资格执行下单,没抢到锁的请求被挂起,等待有锁的请求完成下单后释放锁,然后唤醒被挂起的请求继续去竞争这把锁…
可以把这把锁当做是zk上的一个节点,所有请求发起时,创建该节点,第一个创建该节点成功的请求就意味着获得了锁,其他请求创建都会抛出异常,然后捕获该异常,用全局的countDownLatch将该请求挂起,等获得锁的节点完成下单后,把该节点删除(释放锁),然后计数器-1,把挂起的线程都唤醒,继续去竞争该锁…
注：实现中所用到的api为自己封装后的zkClient工具类，需要的可以看我上一篇文章，里面有这个直接操作zk的工具类
实现：package com.zk.zkclient;
import org.I0Itec.zkclient.IZkChildListener;
import java.util.List;
import java.util.concurrent.CountDownLatch;
public class ZkLock {
private static final String ROOT_NODE = "/zklock";
private static final String LOCK_NODE = "/lock";
private static CountDownLatch countDownLatch = new CountDownLatch(1);
private static volatile boolean isOenRun = true;
private static volatile boolean isRun = true;
public static void lock() {
// 在zk的一个固定根节点，创建一个临时子节点
// 如果根节点不存在，则创建根节点
check();
while (true) {
try {
// 如果程序不是运行状态，则抛异常
if(!isRun){
throw new Exception();
}
// 创建锁的临时节点
ZkClientUtils.createEphemeral(ROOT_NODE + LOCK_NODE);
return;
} catch (Exception e) {
// 获取锁失败,对该节点状态进行监听
ZkClientUtils.addNodeListener(ROOT_NODE, new IZkChildListener() {
@Override
public void handleChildChange(String s, List list) throws Exception {
//锁被释放了
countDownLatch.countDown();
}
});
//如果没有获取到锁,需要重新设置同步资源值
if (countDownLatch.getCount() <= 0) {
countDownLatch = new CountDownLatch(1);
}
// 进行休眠等待
try {
countDownLatch.await();
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
}
private static void check(){
if (isOenRun) {
synchronized (ZkLock.class) {
if (isOenRun) {
if (!ZkClientUtils.exists(ROOT_NODE)) {
ZkClientUtils.createPersistent(ROOT_NODE);
}
// 注册在虚拟机关闭时的回调
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
public void run() {
System.err.println("程序异常关闭，执行回调删除锁.....");
isRun=false;
//释放锁
ZkClientUtils.delete(ROOT_NODE + LOCK_NODE);
}
}));
// 状态修改
isOenRun = false;
}
}
}
}
public static void unLock() {
//释放锁
ZkClientUtils.delete(ROOT_NODE + LOCK_NODE);
}
}
测试
package com.zk.zkclient;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class TestLock {
public static void main(String[] args) {
final ExecutorService threadpool = Executors.newCachedThreadPool();
System.out.println("开始下单...");
for (int i = 0; i < 1000; i++) {
int finalI = i;
threadpool.execute(new Runnable() {
@Override
public void run() {
System.out.println("我是:" + finalI + "开始下单...");
ZkLock.lock();
try {
System.out.println(finalI + ":正在购买中...");
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(finalI + ":我买完了,有请下一位...");
}finally {
// 解锁
ZkLock.unLock();
}
}
});
}
}
}
