分布式系统中经常需要协调多进程，多个jvm，或者多台机器之间的同步问题，得益于zookeeper，实现了一个分布式的共享锁，方便在多台服务器之间竞争资源时，来协调各系统之间的协作和同步。
package com.zookeeper.lock.server;
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;//竞争资源的标志
private String waitNode;//等待前一个锁
private String myZnode;//当前锁
private CountDownLatch latch;//计数器
private int sessionTimeout = 30000;
private List<Exception> exception = new ArrayList<Exception>();
/**
* 创建分布式锁,使用前请确认config配置的zookeeper服务可用
* @param config 127.0.0.1:2181
* @param lockName 竞争资源标志,lockName中不能包含单词lock
*/
public DistributedLock(String config, String lockName){
this.lockName = lockName;
// 创建一个与服务器的连接
try {
zk = new ZooKeeper(config, sessionTimeout, this);
Stat stat = zk.exists(root, false);
if(stat == null){
// 创建根节点
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节点的监视器
*/
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");
//创建临时子节点
myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(myZnode + " is created ");
//取出所有子节点
List<String> subNodes = zk.getChildren(root, false);
//取出所有lockName的锁
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))){
//如果是最小的节点,则表示取得锁
return true;
}
//如果不是最小的节点，找到比自己小1的节点
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);
//判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听
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);
}
}
}
package com.zookeeper.lock.client;
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 ConcurrentClient {
private CountDownLatch startSignal = new CountDownLatch(1);//开始阀门
private CountDownLatch doneSignal = null;//结束阀门
private CopyOnWriteArrayList<Long> list = new CopyOnWriteArrayList<Long>();
private AtomicInteger err = new AtomicInteger();//原子递增
private ConcurrentTask[] task = null;
public ConcurrentClient(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(){
//创建线程，并将所有线程等待在阀门处
createThread();
//打开阀门
startSignal.countDown();//递减锁存器的计数，如果计数到达零，则释放所有等待的线程
try {
doneSignal.await();//等待所有线程都执行完毕
} catch (InterruptedException e) {
e.printStackTrace();
}
//计算执行时间
getExeTime();
}
/**
* 初始化所有线程，并在阀门处等待
*/
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();//使当前线程在锁存器倒计数至零之前一直等待
long start = System.currentTimeMillis();
task[j].run();
long end = (System.currentTimeMillis() - start);
list.add(end);
} catch (Exception e) {
err.getAndIncrement();//相当于err++
}
doneSignal.countDown();
}
}).start();
}
}
/**
* 计算平均响应时间
*/
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();
}
}
package com.zookeeper.lock.test;
import com.zookeeper.lock.client.ConcurrentClient;
import com.zookeeper.lock.client.ConcurrentClient.ConcurrentTask;
import com.zookeeper.lock.server.DistributedLock;
public class Test {
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[5];
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:2181","test2");
lock.lock();
System.out.println("Thread " + Thread.currentThread().getId() + " running");
} catch (Exception e) {
e.printStackTrace();
}
finally {
lock.unlock();
}
}
};
tasks[i] = task3;
}
new ConcurrentClient(tasks);
}
}
测试结果:
/locks/test1_lock_0000004356 is created /locks/test1_lock_0000004356==test1_lock_0000004356 Thread 8 /locks/test1_lock_0000004356 get lock true /locks/test2_lock_0000004357 is created /locks/test2_lock_0000004359 is created /locks/test2_lock_0000004358 is created /locks/test2_lock_0000004363 is created /locks/test2_lock_0000004361 is created /locks/test2_lock_0000004360 is created /locks/test2_lock_0000004362 is created /locks/test2_lock_0000004366 is created /locks/test2_lock_0000004365 is created /locks/test2_lock_0000004364 is created /locks/test2_lock_0000004357==test2_lock_0000004357 Thread 14 /locks/test2_lock_0000004357 get lock true Thread 14 running unlock /locks/test2_lock_0000004357 /locks/test2_lock_0000004358==test2_lock_0000004357 /locks/test2_lock_0000004361==test2_lock_0000004357 /locks/test2_lock_0000004359==test2_lock_0000004357 /locks/test2_lock_0000004362==test2_lock_0000004357 Thread 12 waiting for /locks/test2_lock_0000004360 /locks/test2_lock_0000004366==test2_lock_0000004357 Thread 18 waiting for /locks/test2_lock_0000004357 /locks/test2_lock_0000004363==test2_lock_0000004357 Thread 18 running unlock /locks/test2_lock_0000004358 Thread 13 waiting for /locks/test2_lock_0000004362 /locks/test2_lock_0000004365==test2_lock_0000004358 Thread 16 waiting for /locks/test2_lock_0000004361 Thread 19 waiting for /locks/test2_lock_0000004358 /locks/test2_lock_0000004360==test2_lock_0000004358 Thread 15 waiting for /locks/test2_lock_0000004365 /locks/test2_lock_0000004364==test2_lock_0000004358 Thread 11 waiting for /locks/test2_lock_0000004364 Thread 20 waiting for /locks/test2_lock_0000004359 Thread 19 running unlock /locks/test2_lock_0000004359 Thread 17 waiting for /locks/test2_lock_0000004363 Thread 20 running unlock /locks/test2_lock_0000004360 Thread 12 running unlock /locks/test2_lock_0000004361 Thread 16 running unlock /locks/test2_lock_0000004362 Thread 13 running unlock /locks/test2_lock_0000004363 Thread 17 running unlock /locks/test2_lock_0000004364 Thread 11 running unlock /locks/test2_lock_0000004365 Thread 15 running unlock /locks/test2_lock_0000004366 min: 506 max: 1481 avg: 968 err: 0 ===Thread 8 running unlock /locks/test1_lock_0000004356
关于zookeeper的很好的文章：
https://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/
这个分布式共享锁就是参考这篇文章实现的。
来自：http://my.oschina.net/shenxueliang/blog/135865
相关推荐
如何操作Redis和zookeeper实现分布式锁 在分布式场景下，有很多种情况都需要实现最终一致性。在设计远程上下文的领域事件的时候，为了保证最终一致性，在通过领域事件进行通讯的方式中，可以共享存储（领域模型和...
本文深入探讨了Zookeeper在分布式系统中的关键应用，特别是在实现分布式锁（包括非公平锁、公平锁和共享锁）、Leader选举和Spring Cloud Zookeeper注册中心等方面的实战应用。通过具体案例，我们理解了Zookeeper的...
通过redisson实现分布式锁 通过curator框架实现共享锁，读写锁 项目下载下载，启动本地redis和zookeeper即可； 代码供大家参考，如有不对的地方希望大家指出来。
分布式锁的一般实现方法是在应用服务器之外通过一个共享的存储服务器存储锁资源，同一时刻只有一个客户端能占有锁资源来完成。通常有基于Zookeeper，Redis，或数据库三种实现形式。本文介绍基于Redis的实现方案。
3、Zookeeper所提供的服务涵盖：主从协调、服务器节点动态上下线、统一配置管理、分布式共享锁、统> 一名称服务等 4、虽然说可以提供各种服务，但是zookeeper在底层其实只提供了两个功能： 管理(存储，读取)用户...
在分布式环境中实现Leader选举、互斥锁和读写锁通常涉及到协调服务，如etcd、Zookeeper或Consul。这些服务提供了必要的原语来处理节点间的协调和数据一致性。以下是实现这些功能的一般步骤： Leader选举: 使用协调...
分布式同步：ZooKeeper提供了分布式锁和顺序节点等特性，可以帮助开发者实现复杂的分布式同步机制。 组服务：ZooKeeper支持创建临时节点，可以用于实现分布式队列、成员管理和领导者选举等功能。 ZooKeeper被广泛...
Zookeeper的Wiki页面展示了如何使用Zookeeper来处理事件通知，队列，优先队列，锁，共享锁，可撤销的共享锁，两阶段提交。 那么Zookeeper能帮我们作什么事情呢？简单的例子：假设我们我们有个20个搜索引擎的服务器...
基于Zookeeper实现分布式服务器动态上下线感知 深入分析Zookeeper Zab协议及选举机制源码解读 Dubbo 使用Dubbo对单一应用服务化改造 Dubbo管理中心及及监控平台安装部署 Dubbo分布式服务模块划分（领域驱动） ...
因为分布式系统本身的复杂特性，以及对于容错性的要求，这些技术通常是重量级的，比如 Paxos 算法，欺负选举算法，ZooKeeper 等，侧重于消息的通信而不是共享内存，通常也是出了名的复杂和难以理解，当在具体的实现...
ZooKeeper实现服务器集群的节点数据的共享、Leader仲裁选举。常用做数据发布与订阅、负载均衡、命名服务（Naming Service）、分布式通知/协调、集群管理与Master选举及分布式锁/队列等。
使用zookeeper分布式锁，读取和写入共享资源Redis，测试高并发下，多个程序同时读取和写入Redis同一个键值数据
Zookeeper 非公平锁/公平锁/共享锁demo代码
该项目已成功集成 actuator(监控)、admin(可视化监控)、...job(分布式定时任务)、swagger(API接口管理测试)、security(基于RBAC的动态权限认证)、SpringSession(Session共享)、Zookeeper(结合AOP实现分布式锁)、Ra
在分布式系统中，如何管理节点间的状态一直是一个难题，etcd像是专门为集群环境的服务发现和注册而设计，它提供了数据TTL失效、数据改变监视、多值、目录监听、分布式锁原子操作等功能，可以方便的跟踪并管理集群...
Doorman 是一个客户端速率限制的解决方案，客户端与共享资源进行通讯，包括数据库、gRPC 服务、RESTful API...其高可用特性需要一个分布式的锁管理器，当前支持 etcd，也可使用 Zookeeper 替代。 标签：Doorman
job(分布式定时任务)、swagger(API接口管理测试)、security(基于RBAC的动态权限认证)、SpringSession(Session共享)、Zookeeper(结合AOP实现分布式锁)、RabbitMQ(消息队列)、Kafka(消息队列)、websocket(服务端推送...
job(`分布式定时任务`)、swagger(`API接口管理测试`)、security(`基于RBAC的动态权限认证`)、SpringSession(`Session共享`)、Zookeeper(`结合AOP实现分布式锁`)、RabbitMQ(`消息队列`)、Kafka(`消息队列`)、...