java使用zookeeper實(shí)現(xiàn)的分布式鎖示例
使用zookeeper實(shí)現(xiàn)的分布式鎖
分布式鎖，實(shí)現(xiàn)了Lock接口
package com.concurrent;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
/**
DistributedLock lock = null;
try {
lock = new DistributedLock("127.0.0.1:2182","test");
lock.lock();
//do something...
} catch (Exception e) {
e.printStackTrace();
}
finally {
if(lock != null)
lock.unlock();
}
* @author xueliang
*
*/
public class DistributedLock implements Lock, Watcher{
private ZooKeeper zk;
private String root = "/locks";//根
private String lockName;//競(jìng)爭(zhēng)資源的標(biāo)志
private String waitNode;//等待前一個(gè)鎖
private String myZnode;//當(dāng)前鎖
private CountDownLatch latch;//計(jì)數(shù)器
private int sessionTimeout = 30000;
private List<Exception> exception = new ArrayList<Exception>();
/**
* 創(chuàng)建分布式鎖,使用前請(qǐng)確認(rèn)config配置的zookeeper服務(wù)可用
* @param config 127.0.0.1:2181
* @param lockName 競(jìng)爭(zhēng)資源標(biāo)志,lockName中不能包含單詞lock
*/
public DistributedLock(String config, String lockName){
this.lockName = lockName;
// 創(chuàng)建一個(gè)與服務(wù)器的連接
try {
zk = new ZooKeeper(config, sessionTimeout, this);
Stat stat = zk.exists(root, false);
if(stat == null){
// 創(chuàng)建根節(jié)點(diǎn)
zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
} catch (IOException e) {
exception.add(e);
} catch (KeeperException e) {
exception.add(e);
} catch (InterruptedException e) {
exception.add(e);
}
}
/**
* zookeeper節(jié)點(diǎn)的監(jiān)視器
*/
public void process(WatchedEvent event) {
if(this.latch != null) {
this.latch.countDown();
}
}
public void lock() {
if(exception.size() > 0){
throw new LockException(exception.get(0));
}
try {
if(this.tryLock()){
System.out.println("Thread " + Thread.currentThread().getId() + " " +myZnode + " get lock true");
return;
}
else{
waitForLock(waitNode, sessionTimeout);//等待鎖
}
} catch (KeeperException e) {
throw new LockException(e);
} catch (InterruptedException e) {
throw new LockException(e);
}
}
public boolean tryLock() {
try {
String splitStr = "_lock_";
if(lockName.contains(splitStr))
throw new LockException("lockName can not contains \\u000B");
//創(chuàng)建臨時(shí)子節(jié)點(diǎn)
myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(myZnode + " is created ");
//取出所有子節(jié)點(diǎn)
List<String> subNodes = zk.getChildren(root, false);
//取出所有l(wèi)ockName的鎖
List<String> lockObjNodes = new ArrayList<String>();
for (String node : subNodes) {
String _node = node.split(splitStr)[0];
if(_node.equals(lockName)){
lockObjNodes.add(node);
}
}
Collections.sort(lockObjNodes);
System.out.println(myZnode + "==" + lockObjNodes.get(0));
if(myZnode.equals(root+"/"+lockObjNodes.get(0))){
//如果是最小的節(jié)點(diǎn),則表示取得鎖
return true;
}
//如果不是最小的節(jié)點(diǎn)，找到比自己小1的節(jié)點(diǎn)
String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);
waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1);
} catch (KeeperException e) {
throw new LockException(e);
} catch (InterruptedException e) {
throw new LockException(e);
}
return false;
}
public boolean tryLock(long time, TimeUnit unit) {
try {
if(this.tryLock()){
return true;
}
return waitForLock(waitNode,time);
} catch (Exception e) {
e.printStackTrace();
}
return false;
}
private boolean waitForLock(String lower, long waitTime) throws InterruptedException, KeeperException {
Stat stat = zk.exists(root + "/" + lower,true);
//判斷比自己小一個(gè)數(shù)的節(jié)點(diǎn)是否存在,如果不存在則無(wú)需等待鎖,同時(shí)注冊(cè)監(jiān)聽(tīng)
if(stat != null){
System.out.println("Thread " + Thread.currentThread().getId() + " waiting for " + root + "/" + lower);
this.latch = new CountDownLatch(1);
this.latch.await(waitTime, TimeUnit.MILLISECONDS);
this.latch = null;
}
return true;
}
public void unlock() {
try {
System.out.println("unlock " + myZnode);
zk.delete(myZnode,-1);
myZnode = null;
zk.close();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
public void lockInterruptibly() throws InterruptedException {
this.lock();
}
public Condition newCondition() {
return null;
}
public class LockException extends RuntimeException {
private static final long serialVersionUID = 1L;
public LockException(String e){
super(e);
}
public LockException(Exception e){
super(e);
}
}
}
并發(fā)測(cè)試工具
package com.concurrent;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
/**
ConcurrentTask[] task = new ConcurrentTask[5];
for(int i=0;i<task.length;i++){
task[i] = new ConcurrentTask(){
public void run() {
System.out.println("==============");
}};
}
new ConcurrentTest(task);
* @author xueliang
*
*/
public class ConcurrentTest {
private CountDownLatch startSignal = new CountDownLatch(1);//開(kāi)始閥門(mén)
private CountDownLatch doneSignal = null;//結(jié)束閥門(mén)
private CopyOnWriteArrayList<Long> list = new CopyOnWriteArrayList<Long>();
private AtomicInteger err = new AtomicInteger();//原子遞增
private ConcurrentTask[] task = null;
public ConcurrentTest(ConcurrentTask... task){
this.task = task;
if(task == null){
System.out.println("task can not null");
System.exit(1);
}
doneSignal = new CountDownLatch(task.length);
start();
}
/**
* @param args
* @throws ClassNotFoundException
*/
private void start(){
//創(chuàng)建線程，并將所有線程等待在閥門(mén)處
createThread();
//打開(kāi)閥門(mén)
startSignal.countDown();//遞減鎖存器的計(jì)數(shù)，如果計(jì)數(shù)到達(dá)零，則釋放所有等待的線程
try {
doneSignal.await();//等待所有線程都執(zhí)行完畢
} catch (InterruptedException e) {
e.printStackTrace();
}
//計(jì)算執(zhí)行時(shí)間
getExeTime();
}
/**
* 初始化所有線程，并在閥門(mén)處等待
*/
private void createThread() {
long len = doneSignal.getCount();
for (int i = 0; i < len; i++) {
final int j = i;
new Thread(new Runnable(){
public void run() {
try {
startSignal.await();//使當(dāng)前線程在鎖存器倒計(jì)數(shù)至零之前一直等待
long start = System.currentTimeMillis();
task[j].run();
long end = (System.currentTimeMillis() - start);
list.add(end);
} catch (Exception e) {
err.getAndIncrement();//相當(dāng)于err++
}
doneSignal.countDown();
}
}).start();
}
}
/**
* 計(jì)算平均響應(yīng)時(shí)間
*/
private void getExeTime() {
int size = list.size();
List<Long> _list = new ArrayList<Long>(size);
_list.addAll(list);
Collections.sort(_list);
long min = _list.get(0);
long max = _list.get(size-1);
long sum = 0L;
for (Long t : _list) {
sum += t;
}
long avg = sum/size;
System.out.println("min: " + min);
System.out.println("max: " + max);
System.out.println("avg: " + avg);
System.out.println("err: " + err.get());
}
public interface ConcurrentTask {
void run();
}
}
測(cè)試
package com.concurrent;
import com.concurrent.ConcurrentTest.ConcurrentTask;
public class ZkTest {
public static void main(String[] args) {
Runnable task1 = new Runnable(){
public void run() {
DistributedLock lock = null;
try {
lock = new DistributedLock("127.0.0.1:2182","test1");
//lock = new DistributedLock("127.0.0.1:2182","test2");
lock.lock();
Thread.sleep(3000);
System.out.println("===Thread " + Thread.currentThread().getId() + " running");
} catch (Exception e) {
e.printStackTrace();
}
finally {
if(lock != null)
lock.unlock();
}
}
};
new Thread(task1).start();
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
ConcurrentTask[] tasks = new ConcurrentTask[60];
for(int i=0;i<tasks.length;i++){
ConcurrentTask task3 = new ConcurrentTask(){
public void run() {
DistributedLock lock = null;
try {
lock = new DistributedLock("127.0.0.1:2183","test2");
lock.lock();
System.out.println("Thread " + Thread.currentThread().getId() + " running");
} catch (Exception e) {
e.printStackTrace();
}
finally {
lock.unlock();
}
}
};
tasks[i] = task3;
}
new ConcurrentTest(tasks);
}
}
相關(guān)文章
這篇文章主要介紹了基于IntelliJ IDEA的類注釋和方法注釋操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-09-09
SpringBoot使用PageHelper插件實(shí)現(xiàn)Mybatis分頁(yè)效果
這篇文章主要介紹了SpringBoot使用PageHelper插件實(shí)現(xiàn)Mybatis分頁(yè)效果,本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作有一定的參考借鑒價(jià)值,需要的朋友可以參考下2024-02-02
這篇文章主要介紹了HashMap線程不安全問(wèn)題解析,HashMap的線程不安全體現(xiàn)在會(huì)造成死循環(huán)、數(shù)據(jù)丟失、數(shù)據(jù)覆蓋等問(wèn)題,其中死循環(huán)和數(shù)據(jù)丟失是在JDK1.7中出現(xiàn)的問(wèn)題,在JDK1.8中已經(jīng)得到解決,但是1.8中仍會(huì)有數(shù)據(jù)覆蓋這樣的問(wèn)題,需要的朋友可以參考下2023-11-11
mybatis-plus 擴(kuò)展批量新增的實(shí)現(xiàn)
本文主要介紹了mybatis-plus 擴(kuò)展批量新增的實(shí)現(xiàn)，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-01-01
這篇文章主要介紹了springBoot2.X配置全局捕獲異常的操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-07-07
Java基于rest assured實(shí)現(xiàn)接口測(cè)試過(guò)程解析
這篇文章主要介紹了Java基于rest assured實(shí)現(xiàn)接口測(cè)試過(guò)程解析,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-03-03
不寫(xiě)mybatis的@Param有的報(bào)錯(cuò)有的卻不報(bào)錯(cuò)問(wèn)題分析
這篇文章主要為大家介紹了不寫(xiě)mybatis的@Param有的報(bào)錯(cuò)有的卻不報(bào)錯(cuò)問(wèn)題分析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-09-09
這篇文章主要介紹了Java 從Set里面取出有序的記錄詳解及實(shí)例的相關(guān)資料,需要的朋友可以參考下2017-06-06
這篇文章主要為大家介紹了java?zxing合成復(fù)雜二維碼圖片示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-05-05
這篇文章主要介紹了詳解微信小程序 同步異步解決辦法的相關(guān)資料,需要的朋友可以參考下2017-06-06