在分布式系统中，往往需要一些分布式同步原语来做一些协同工作，上一篇文章介绍了Zookeeper的基本原理，本文介绍下基于Zookeeper的Lock和Queue的实现，主要代码都来自Zookeeper的官方recipe。
锁（Lock）
完全分布式锁是全局同步的，这意味着在任何时刻没有两个客户端会同时认为它们都拥有相同的锁，使用 Zookeeper 可以实现分布式锁，需要首先定义一个锁节点（lock root node）。
需要获得锁的客户端按照以下步骤来获取锁：
保证锁节点（lock root node）这个父根节点的存在，这个节点是每个要获取lock客户端共用的，这个节点是PERSISTENT的。
第一次需要创建本客户端要获取lock的节点，调用 create( )，并设置 节点为EPHEMERAL_SEQUENTIAL类型，表示该节点为临时的和顺序的。
在父锁节点（lock root node）上调用 getChildren( ) ，不需要设置监视标志。 (为了避免“羊群效应”).
按照Fair竞争的原则，将步骤3中的子节点（要获取锁的节点）按照节点顺序的大小做排序，取出编号最小的一个节点做为lock的owner，判断自己的节点id
是否就为owner id，如果是则返回，lock成功。如果不是则调用 exists( )监听比自己小的前一位的id，关注它锁释放的操作（也就是exist watch）。
如果第4步监听exist的watch被触发，则继续按4中的原则判断自己是否能获取到lock。
释放锁：需要释放锁的客户端只需要删除在第2步中创建的节点即可。
注意事项：
一个节点的删除只会导致一个客户端被唤醒，因为每个节点只被一个客户端watch，这避免了“羊群效应”。
一个分布式lock的实现：
/**
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.zookeeper.recipes.lock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import static org.apache.zookeeper.CreateMode.EPHEMERAL_SEQUENTIAL;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
/**
* A <a href="package.html">protocol to implement an exclusive
* write lock or to elect a leader</a>. <p/> You invoke {@link #lock()} to
* start the process of grabbing the lock; you may get the lock then or it may be
* some time later. <p/> You can register a listener so that you are invoked
* when you get the lock; otherwise you can ask if you have the lock
* by calling {@link #isOwner()}
*
*/
public class WriteLock extends ProtocolSupport {
private static final Logger LOG = LoggerFactory.getLogger(WriteLock.class);
private final String dir;
private String id;
private ZNodeName idName;
private String ownerId;
private String lastChildId;
private byte[] data = {0x12, 0x34};
private LockListener callback;
private LockZooKeeperOperation zop;
/**
* zookeeper contructor for writelock
* @param zookeeper zookeeper client instance
* @param dir the parent path you want to use for locking
* @param acls the acls that you want to use for all the paths,
* if null world read/write is used.
*/
public WriteLock(ZooKeeper zookeeper, String dir, List<ACL> acl) {
super(zookeeper);
this.dir = dir;
if (acl != null) {
setAcl(acl);
}
this.zop = new LockZooKeeperOperation();
}
/**
* zookeeper contructor for writelock with callback
* @param zookeeper the zookeeper client instance
* @param dir the parent path you want to use for locking
* @param acl the acls that you want to use for all the paths
* @param callback the call back instance
*/
public WriteLock(ZooKeeper zookeeper, String dir, List<ACL> acl,
LockListener callback) {
this(zookeeper, dir, acl);
this.callback = callback;
}
/**
* return the current locklistener
* @return the locklistener
*/
public LockListener getLockListener() {
return this.callback;
}
/**
* register a different call back listener
* @param callback the call back instance
*/
public void setLockListener(LockListener callback) {
this.callback = callback;
}
/**
* Removes the lock or associated znode if
* you no longer require the lock. this also
* removes your request in the queue for locking
* in case you do not already hold the lock.
* @throws RuntimeException throws a runtime exception
* if it cannot connect to zookeeper.
*/
public synchronized void unlock() throws RuntimeException {
if (!isClosed() && id != null) {
// we don't need to retry this operation in the case of failure
// as ZK will remove ephemeral files and we don't wanna hang
// this process when closing if we cannot reconnect to ZK
try {
ZooKeeperOperation zopdel = new ZooKeeperOperation() {
public boolean execute() throws KeeperException,
InterruptedException {
zookeeper.delete(id, -1);
return Boolean.TRUE;
}
};
zopdel.execute();
} catch (InterruptedException e) {
LOG.warn("Caught: " + e, e);
//set that we have been interrupted.
Thread.currentThread().interrupt();
} catch (KeeperException.NoNodeException e) {
// do nothing
} catch (KeeperException e) {
LOG.warn("Caught: " + e, e);
throw (RuntimeException) new RuntimeException(e.getMessage()).
initCause(e);
}
finally {
if (callback != null) {
callback.lockReleased();
}
id = null;
}
}
}
/**
* the watcher called on
* getting watch while watching
* my predecessor
*/
private class LockWatcher implements Watcher {
public void process(WatchedEvent event) {
// lets either become the leader or watch the new/updated node
LOG.debug("Watcher fired on path: " + event.getPath() + " state: " +
event.getState() + " type " + event.getType());
try {
lock();
} catch (Exception e) {
LOG.warn("Failed to acquire lock: " + e, e);
}
}
}
/**
* a zoookeeper operation that is mainly responsible
* for all the magic required for locking.
*/
private class LockZooKeeperOperation implements ZooKeeperOperation {
/** find if we have been created earler if not create our node
*
* @param prefix the prefix node
* @param zookeeper teh zookeeper client
* @param dir the dir paretn
* @throws KeeperException
* @throws InterruptedException
*/
private void findPrefixInChildren(String prefix, ZooKeeper zookeeper, String dir)
throws KeeperException, InterruptedException {
List<String> names = zookeeper.getChildren(dir, false);
for (String name : names) {
if (name.startsWith(prefix)) {
id = dir + "/" + name;
if (LOG.isDebugEnabled()) {
LOG.debug("Found id created last time: " + id);
}
break;
}
}
if (id == null) {
id = zookeeper.create(dir + "/" + prefix, data,
getAcl(), EPHEMERAL_SEQUENTIAL);
if (LOG.isDebugEnabled()) {
LOG.debug("Created id: " + id);
}
}
}
/**
* the command that is run and retried for actually
* obtaining the lock
* @return if the command was successful or not
*/
public boolean execute() throws KeeperException, InterruptedException {
do {
if (id == null) {
long sessionId = zookeeper.getSessionId();
String prefix = "x-" + sessionId + "-";
// lets try look up the current ID if we failed
// in the middle of creating the znode
findPrefixInChildren(prefix, zookeeper, dir);
idName = new ZNodeName(id);
}
if (id != null) {
List<String> names = zookeeper.getChildren(dir, false);
if (names.isEmpty()) {
LOG.warn("No children in: " + dir + " when we've just " +
"created one! Lets recreate it...");
// lets force the recreation of the id
id = null;
} else {
// lets sort them explicitly (though they do seem to come back in order ususally :)
SortedSet<ZNodeName> sortedNames = new TreeSet<ZNodeName>();
for (String name : names) {
sortedNames.add(new ZNodeName(dir + "/" + name));
}
ownerId = sortedNames.first().getName();
SortedSet<ZNodeName> lessThanMe = sortedNames.headSet(idName);
if (!lessThanMe.isEmpty()) {
ZNodeName lastChildName = lessThanMe.last();
lastChildId = lastChildName.getName();
if (LOG.isDebugEnabled()) {
LOG.debug("watching less than me node: " + lastChildId);
}
Stat stat = zookeeper.exists(lastChildId, new LockWatcher());
if (stat != null) {
return Boolean.FALSE;
} else {
LOG.warn("Could not find the" +
" stats for less than me: " + lastChildName.getName());
}
} else {
if (isOwner()) {
if (callback != null) {
callback.lockAcquired();
}
return Boolean.TRUE;
}
}
}
}
}
while (id == null);
return Boolean.FALSE;
}
};
/**
* Attempts to acquire the exclusive write lock returning whether or not it was
* acquired. Note that the exclusive lock may be acquired some time later after
* this method has been invoked due to the current lock owner going away.
*/
public synchronized boolean lock() throws KeeperException, InterruptedException {
if (isClosed()) {
return false;
}
ensurePathExists(dir);
return (Boolean) retryOperation(zop);
}
/**
* return the parent dir for lock
* @return the parent dir used for locks.
*/
public String getDir() {
return dir;
}
/**
* Returns true if this node is the owner of the
* lock (or the leader)
*/
public boolean isOwner() {
return id != null && ownerId != null && id.equals(ownerId);
}
/**
* return the id for this lock
* @return the id for this lock
*/
public String getId() {
return this.id;
}
}
注意这里的lock，可能会失败，会尝试多次，每次失败后会Sleep一段时间。
用于辅助节点大小顺序排序的类：
/**
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.zookeeper.recipes.lock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Represents an ephemeral znode name which has an ordered sequence number
* and can be sorted in order
*
*/
class ZNodeName implements Comparable<ZNodeName> {
private final String name;
private String prefix;
private int sequence = -1;
private static final Logger LOG = LoggerFactory.getLogger(ZNodeName.class);
public ZNodeName(String name) {
if (name == null) {
throw new NullPointerException("id cannot be null");
}
this.name = name;
this.prefix = name;
int idx = name.lastIndexOf('-');
if (idx >= 0) {
this.prefix = name.substring(0, idx);
try {
this.sequence = Integer.parseInt(name.substring(idx + 1));
// If an exception occurred we misdetected a sequence suffix,
// so return -1.
} catch (NumberFormatException e) {
LOG.info("Number format exception for " + idx, e);
} catch (ArrayIndexOutOfBoundsException e) {
LOG.info("Array out of bounds for " + idx, e);
}
}
}
@Override
public String toString() {
return name.toString();
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ZNodeName sequence = (ZNodeName) o;
if (!name.equals(sequence.name)) return false;
return true;
}
@Override
public int hashCode() {
return name.hashCode() + 37;
}
public int compareTo(ZNodeName that) {
int answer = this.prefix.compareTo(that.prefix);
if (answer == 0) {
int s1 = this.sequence;
int s2 = that.sequence;
if (s1 == -1 && s2 == -1) {
return this.name.compareTo(that.name);
}
answer = s1 == -1 ? 1 : s2 == -1 ? -1 : s1 - s2;
}
return answer;
}
/**
* Returns the name of the znode
*/
public String getName() {
return name;
}
/**
* Returns the sequence number
*/
public int getZNodeName() {
return sequence;
}
/**
* Returns the text prefix before the sequence number
*/
public String getPrefix() {
return prefix;
}
}
Zookeeper统一操作ZooKeeperOperation接口：
public interface ZooKeeperOperation {
/**
* Performs the operation - which may be involved multiple times if the connection
* to ZooKeeper closes during this operation
*
* @return the result of the operation or null
* @throws KeeperException
* @throws InterruptedException
*/
public boolean execute() throws KeeperException, InterruptedException;
}
因为Zookeeper的操作会失败，这个类封装了多次尝试：
/**
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.zookeeper.recipes.lock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.recipes.lock.ZooKeeperOperation;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* A base class for protocol implementations which provides a number of higher
* level helper methods for working with ZooKeeper along with retrying synchronous
* operations if the connection to ZooKeeper closes such as
* {@link #retryOperation(ZooKeeperOperation)}
*
*/
class ProtocolSupport {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolSupport.class);
protected final ZooKeeper zookeeper;
private AtomicBoolean closed = new AtomicBoolean(false);
private long retryDelay = 500L;
private int retryCount = 10;
private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
public ProtocolSupport(ZooKeeper zookeeper) {
this.zookeeper = zookeeper;
}
/**
* Closes this strategy and releases any ZooKeeper resources; but keeps the
* ZooKeeper instance open
*/
public void close() {
if (closed.compareAndSet(false, true)) {
doClose();
}
}
/**
* return zookeeper client instance
* @return zookeeper client instance
*/
public ZooKeeper getZookeeper() {
return zookeeper;
}
/**
* return the acl its using
* @return the acl.
*/
public List<ACL> getAcl() {
return acl;
}
/**
* set the acl
* @param acl the acl to set to
*/
public void setAcl(List<ACL> acl) {
this.acl = acl;
}
/**
* get the retry delay in milliseconds
* @return the retry delay
*/
public long getRetryDelay() {
return retryDelay;
}
/**
* Sets the time waited between retry delays
* @param retryDelay the retry delay
*/
public void setRetryDelay(long retryDelay) {
this.retryDelay = retryDelay;
}
/**
* Allow derived classes to perform
* some custom closing operations to release resources
*/
protected void doClose() {
}
/**
* Perform the given operation, retrying if the connection fails
* @return object. it needs to be cast to the callee's expected
* return type.
*/
protected Object retryOperation(ZooKeeperOperation operation)
throws KeeperException, InterruptedException {
KeeperException exception = null;
for (int i = 0; i < retryCount; i++) {
try {
return operation.execute();
} catch (KeeperException.SessionExpiredException e) {
LOG.warn("Session expired for: " + zookeeper + " so reconnecting due to: " + e, e);
throw e;
} catch (KeeperException.ConnectionLossException e) {
if (exception == null) {
exception = e;
}
LOG.debug("Attempt " + i + " failed with connection loss so " +
"attempting to reconnect: " + e, e);
retryDelay(i);
}
}
throw exception;
}
/**
* Ensures that the given path exists with no data, the current
* ACL and no flags
* @param path
*/
protected void ensurePathExists(String path) {
ensureExists(path, null, acl, CreateMode.PERSISTENT);
}
/**
* Ensures that the given path exists with the given data, ACL and flags
* @param path
* @param acl
* @param flags
*/
protected void ensureExists(final String path, final byte[] data,
final List<ACL> acl, final CreateMode flags) {
try {
retryOperation(new ZooKeeperOperation() {
public boolean execute() throws KeeperException, InterruptedException {
Stat stat = zookeeper.exists(path, false);
if (stat != null) {
return true;
}
zookeeper.create(path, data, acl, flags);
return true;
}
});
} catch (KeeperException e) {
LOG.warn("Caught: " + e, e);
} catch (InterruptedException e) {
LOG.warn("Caught: " + e, e);
}
}
/**
* Returns true if this protocol has been closed
* @return true if this protocol is closed
*/
protected boolean isClosed() {
return closed.get();
}
/**
* Performs a retry delay if this is not the first attempt
* @param attemptCount the number of the attempts performed so far
*/
protected void retryDelay(int attemptCount) {
if (attemptCount > 0) {
try {
Thread.sleep(attemptCount * retryDelay);
} catch (InterruptedException e) {
LOG.debug("Failed to sleep: " + e, e);
}
}
}
}
这个类是本客户端获取到lock和释放lock的时候触发操作的接口：
public interface LockListener {
/**
* call back called when the lock
* is acquired
*/
public void lockAcquired();
/**
* call back called when the lock is
* released.
*/
public void lockReleased();
}
队列（Queue）
分布式队列是通用的数据结构，为了在 Zookeeper 中实现分布式队列，首先需要指定一个 Znode 节点作为队列节点（queue node）， 各个分布式客户端通过调用 create() 函数向队列中放入数据，调用create()时节点路径名带"qn-"结尾，并设置顺序和临时（sequence and ephemeral）节点标志。 由于设置了节点的顺序标志，新的路径名具有以下字符串模式："_path-to-queue-node_/qn-X"，X 是唯一自增号。需要从队列中移除数据的客户端首先调用 getChildren() 函数，同时在队列节点（queue node）上将 watch 设置为 true，并处理最小序号的节点（即从序号最小的节点中取数据）。客户端不需要再一次调用 getChildren()，队列中的数据获取完。如果队列节点中没有任何子节点，读取队列的客户端需要等待队列的监视事件通知。
实现步骤基本如下：
前提：需要一个队列root节点dir
入队：使用create()创建节点，将共享数据data放在该节点上，节点类型为PERSISTENT_SEQUENTIAL，临时顺序性的。
出队：因为队列可能为空，2种方式处理：一种如果为空则wait等待，一种返回异常。
等待方式：这里使用了CountDownLatch的等待和Watcher的通知机制，使用了TreeMap的排序获取节点顺序最小的数据（FIFO）。
抛出异常：getChildren()获取队列数据时，如果size==0则抛出异常。
一个分布式Queue的实现，详细代码：
/**
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.zookeeper.recipes.queue;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.TreeMap;
import java.util.concurrent.CountDownLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.ACL;
import org.apache.zookeeper.data.Stat;
/**
*
* A <a href="package.html">protocol to implement a distributed queue</a>.
*
*/
public class DistributedQueue {
private static final Logger LOG = LoggerFactory.getLogger(DistributedQueue.class);
private final String dir;
private ZooKeeper zookeeper;
private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
private final String prefix = "qn-";
public DistributedQueue(ZooKeeper zookeeper, String dir, List<ACL> acl){
this.dir = dir;
if(acl != null){
this.acl = acl;
}
this.zookeeper = zookeeper;
//Add root dir first if not exists
if (zookeeper != null) {
try {
Stat s = zookeeper.exists(dir, false);
if (s == null) {
zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
LOG.error(e.toString());
} catch (InterruptedException e) {
LOG.error(e.toString());
}
}
}
/**
* Returns a Map of the children, ordered by id.
* @param watcher optional watcher on getChildren() operation.
* @return map from id to child name for all children
*/
private TreeMap<Long,String> orderedChildren(Watcher watcher) throws KeeperException, InterruptedException {
TreeMap<Long,String> orderedChildren = new TreeMap<Long,String>();
List<String> childNames = null;
try{
childNames = zookeeper.getChildren(dir, watcher);
}catch (KeeperException.NoNodeException e){
throw e;
}
for(String childName : childNames){
try{
//Check format
if(!childName.regionMatches(0, prefix, 0, prefix.length())){
LOG.warn("Found child node with improper name: " + childName);
continue;
}
String suffix = childName.substring(prefix.length());
Long childId = new Long(suffix);
orderedChildren.put(childId,childName);
}catch(NumberFormatException e){
LOG.warn("Found child node with improper format : " + childName + " " + e,e);
}
}
return orderedChildren;
}
/**
* Find the smallest child node.
* @return The name of the smallest child node.
*/
private String smallestChildName() throws KeeperException, InterruptedException {
long minId = Long.MAX_VALUE;
String minName = "";
List<String> childNames = null;
try{
childNames = zookeeper.getChildren(dir, false);
}catch(KeeperException.NoNodeException e){
LOG.warn("Caught: " +e,e);
return null;
}
for(String childName : childNames){
try{
//Check format
if(!childName.regionMatches(0, prefix, 0, prefix.length())){
LOG.warn("Found child node with improper name: " + childName);
continue;
}
String suffix = childName.substring(prefix.length());
long childId = Long.parseLong(suffix);
if(childId < minId){
minId = childId;
minName = childName;
}
}catch(NumberFormatException e){
LOG.warn("Found child node with improper format : " + childName + " " + e,e);
}
}
if(minId < Long.MAX_VALUE){
return minName;
}else{
return null;
}
}
/**
* Return the head of the queue without modifying the queue.
* @return the data at the head of the queue.
* @throws NoSuchElementException
* @throws KeeperException
* @throws InterruptedException
*/
public byte[] element() throws NoSuchElementException, KeeperException, InterruptedException {
TreeMap<Long,String> orderedChildren;
// element, take, and remove follow the same pattern.
// We want to return the child node with the smallest sequence number.
// Since other clients are remove()ing and take()ing nodes concurrently,
// the child with the smallest sequence number in orderedChildren might be gone by the time we check.
// We don't call getChildren again until we have tried the rest of the nodes in sequence order.
while(true){
try{
orderedChildren = orderedChildren(null);
}catch(KeeperException.NoNodeException e){
throw new NoSuchElementException();
}
if(orderedChildren.size() == 0 ) throw new NoSuchElementException();
for(String headNode : orderedChildren.values()){
if(headNode != null){
try{
return zookeeper.getData(dir+"/"+headNode, false, null);
}catch(KeeperException.NoNodeException e){
//Another client removed the node first, try next
}
}
}
}
}
/**
* Attempts to remove the head of the queue and return it.
* @return The former head of the queue
* @throws NoSuchElementException
* @throws KeeperException
* @throws InterruptedException
*/
public byte[] remove() throws NoSuchElementException, KeeperException, InterruptedException {
TreeMap<Long,String> orderedChildren;
// Same as for element. Should refactor this.
while(true){
try{
orderedChildren = orderedChildren(null);
}catch(KeeperException.NoNodeException e){
throw new NoSuchElementException();
}
if(orderedChildren.size() == 0) throw new NoSuchElementException();
for(String headNode : orderedChildren.values()){
String path = dir +"/"+headNode;
try{
byte[] data = zookeeper.getData(path, false, null);
zookeeper.delete(path, -1);
return data;
}catch(KeeperException.NoNodeException e){
// Another client deleted the node first.
}
}
}
}
private class LatchChildWatcher implements Watcher {
CountDownLatch latch;
public LatchChildWatcher(){
latch = new CountDownLatch(1);
}
public void process(WatchedEvent event){
LOG.debug("Watcher fired on path: " + event.getPath() + " state: " +
event.getState() + " type " + event.getType());
latch.countDown();
}
public void await() throws InterruptedException {
latch.await();
}
}
/**
* Removes the head of the queue and returns it, blocks until it succeeds.
* @return The former head of the queue
* @throws NoSuchElementException
* @throws KeeperException
* @throws InterruptedException
*/
public byte[] take() throws KeeperException, InterruptedException {
TreeMap<Long,String> orderedChildren;
// Same as for element. Should refactor this.
while(true){
LatchChildWatcher childWatcher = new LatchChildWatcher();
try{
orderedChildren = orderedChildren(childWatcher);
}catch(KeeperException.NoNodeException e){
zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT);
continue;
}
if(orderedChildren.size() == 0){
childWatcher.await();
continue;
}
for(String headNode : orderedChildren.values()){
String path = dir +"/"+headNode;
try{
byte[] data = zookeeper.getData(path, false, null);
zookeeper.delete(path, -1);
return data;
}catch(KeeperException.NoNodeException e){
// Another client deleted the node first.
}
}
}
}
/**
* Inserts data into queue.
* @param data
* @return true if data was successfully added
*/
public boolean offer(byte[] data) throws KeeperException, InterruptedException{
for(;;){
try{
zookeeper.create(dir+"/"+prefix, data, acl, CreateMode.PERSISTENT_SEQUENTIAL);
return true;
}catch(KeeperException.NoNodeException e){
zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT);
}
}
}
/**
* Returns the data at the first element of the queue, or null if the queue is empty.
* @return data at the first element of the queue, or null.
* @throws KeeperException
* @throws InterruptedException
*/
public byte[] peek() throws KeeperException, InterruptedException{
try{
return element();
}catch(NoSuchElementException e){
return null;
}
}
/**
* Attempts to remove the head of the queue and return it. Returns null if the queue is empty.
* @return Head of the queue or null.
* @throws KeeperException
* @throws InterruptedException
*/
public byte[] poll() throws KeeperException, InterruptedException {
try{
return remove();
}catch(NoSuchElementException e){
return null;
}
}
}
基于zookeeper简单实现分布式锁
这里利用zookeeper的EPHEMERAL_SEQUENTIAL类型节点及watcher机制，来简单实现分布式锁。
主要思想：
1、开启10个线程，在disLocks节点下各自创建名为sub的EPHEMERAL_SEQUENTIAL节点；
2、获取disLocks节点下所有子节点，排序，如果自己的节点编号最小，则获取锁；
3、否则watch排在自己前面的节点，监听到其删除后，进入第2步（重新检测排序是防止监听的节点发生连接失效，导致的节点删除情况）；
4、删除自身sub节点，释放连接；
这里插播下zookeeper的4种节点类型：
public enum CreateMode {
/**
* 持久节点：节点创建后，会一直存在，不会因客户端会话失效而删除；
*/
PERSISTENT (0, false, false),
/**
* 持久顺序节点：基本特性与持久节点一致，创建节点的过程中，zookeeper会在其名字后自动追加一个单调增长的数字后缀，作为新的节点名；
*/
PERSISTENT_SEQUENTIAL (2, false, true),
/**
* 临时节点：客户端会话失效或连接关闭后，该节点会被自动删除，且不能再临时节点下面创建子节点，否则报如下错：org.apache.zookeeper.KeeperException$NoChildrenForEphemeralsException；
*/
EPHEMERAL (1, true, false),
/**
* 临时顺序节点：基本特性与临时节点一致，创建节点的过程中，zookeeper会在其名字后自动追加一个单调增长的数字后缀，作为新的节点名；
*/
EPHEMERAL_SEQUENTIAL (3, true, true);
private static final Logger LOG = LoggerFactory.getLogger(CreateMode.class);
private boolean ephemeral;
private boolean sequential;
private int flag;
CreateMode(int flag, boolean ephemeral, boolean sequential) {
this.flag = flag;
this.ephemeral = ephemeral;
this.sequential = sequential;
}
public boolean isEphemeral() {
return ephemeral;
}
public boolean isSequential() {
return sequential;
}
public int toFlag() {
return flag;
}
static public CreateMode fromFlag(int flag) throws KeeperException {
switch(flag) {
case 0: return CreateMode.PERSISTENT;
case 1: return CreateMode.EPHEMERAL;
case 2: return CreateMode.PERSISTENT_SEQUENTIAL;
case 3: return CreateMode.EPHEMERAL_SEQUENTIAL ;
default:
LOG.error("Received an invalid flag value to convert to a CreateMode");
throw new KeeperException.BadArgumentsException();
}
}
}
测试代码：
package zookeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.util.List;
import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
public class DistributedLock implements Watcher{
private int threadId;
private ZooKeeper zk = null;
private String selfPath;
private String waitPath;
private String LOG_PREFIX_OF_THREAD;
private static final int SESSION_TIMEOUT = 10000;
private static final String GROUP_PATH = "/disLocks";
private static final String SUB_PATH = "/disLocks/sub";
private static final String CONNECTION_STRING = "192.168.*.*:2181";
private static final int THREAD_NUM = 10;
//确保连接zk成功；
private CountDownLatch connectedSemaphore = new CountDownLatch(1);
//确保所有线程运行结束；
private static final CountDownLatch threadSemaphore = new CountDownLatch(THREAD_NUM);
private static final Logger LOG = LoggerFactory.getLogger(AllZooKeeperWatcher.class);
public DistributedLock(int id) {
this.threadId = id;
LOG_PREFIX_OF_THREAD = "【第"+threadId+"个线程】";
}
public static void main(String[] args) {
for(int i=0; i < THREAD_NUM; i++){
final int threadId = i+1;
new Thread(){
@Override
public void run() {
try{
DistributedLock dc = new DistributedLock(threadId);
dc.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
//GROUP_PATH不存在的话，由一个线程创建即可；
synchronized (threadSemaphore){
dc.createPath(GROUP_PATH, "该节点由线程" + threadId + "创建", true);
}
dc.getLock();
} catch (Exception e){
LOG.error("【第"+threadId+"个线程】 抛出的异常：");
e.printStackTrace();
}
}
}.start();
}
try {
threadSemaphore.await();
LOG.info("所有线程运行结束!");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 获取锁
* @return
*/
private void getLock() throws KeeperException, InterruptedException {
selfPath = zk.create(SUB_PATH,null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
LOG.info(LOG_PREFIX_OF_THREAD+"创建锁路径:"+selfPath);
if(checkMinPath()){
getLockSuccess();
}
}
/**
* 创建节点
* @param path 节点path
* @param data 初始数据内容
* @return
*/
public boolean createPath( String path, String data, boolean needWatch) throws KeeperException, InterruptedException {
if(zk.exists(path, needWatch)==null){
LOG.info( LOG_PREFIX_OF_THREAD + "节点创建成功, Path: "
+ this.zk.create( path,
data.getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT )
+ ", content: " + data );
}
return true;
}
/**
* 创建ZK连接
* @param connectString ZK服务器地址列表
* @param sessionTimeout Session超时时间
*/
public void createConnection( String connectString, int sessionTimeout ) throws IOException, InterruptedException {
zk = new ZooKeeper( connectString, sessionTimeout, this);
connectedSemaphore.await();
}
/**
* 获取锁成功
*/
public void getLockSuccess() throws KeeperException, InterruptedException {
if(zk.exists(this.selfPath,false) == null){
LOG.error(LOG_PREFIX_OF_THREAD+"本节点已不在了...");
return;
}
LOG.info(LOG_PREFIX_OF_THREAD + "获取锁成功，赶紧干活！");
Thread.sleep(2000);
LOG.info(LOG_PREFIX_OF_THREAD + "删除本节点："+selfPath);
zk.delete(this.selfPath, -1);
releaseConnection();
threadSemaphore.countDown();
}
/**
* 关闭ZK连接
*/
public void releaseConnection() {
if ( this.zk !=null ) {
try {
this.zk.close();
} catch ( InterruptedException e ) {}
}
LOG.info(LOG_PREFIX_OF_THREAD + "释放连接");
}
/**
* 检查自己是不是最小的节点
* @return
*/
public boolean checkMinPath() throws KeeperException, InterruptedException {
List<String> subNodes = zk.getChildren(GROUP_PATH, false);
Collections.sort(subNodes);
int index = subNodes.indexOf( selfPath.substring(GROUP_PATH.length()+1));
switch (index){
case -1:{
LOG.error(LOG_PREFIX_OF_THREAD+"本节点已不在了..."+selfPath);
return false;
}
case 0:{
LOG.info(LOG_PREFIX_OF_THREAD+"子节点中，我果然是老大"+selfPath);
return true;
}
default:{
this.waitPath = GROUP_PATH +"/"+ subNodes.get(index - 1);
LOG.info(LOG_PREFIX_OF_THREAD+"获取子节点中，排在我前面的"+waitPath);
try{
zk.getData(waitPath, true, new Stat());
return false;
}catch(KeeperException e){
if(zk.exists(waitPath,false) == null){
LOG.info(LOG_PREFIX_OF_THREAD+"子节点中，排在我前面的"+waitPath+"已失踪，幸福来得太突然?");
return checkMinPath();
}else{
throw e;
}
}
}
}
}
@Override
public void process(WatchedEvent event) {
if(event == null){
return;
}
Event.KeeperState keeperState = event.getState();
Event.EventType eventType = event.getType();
if ( Event.KeeperState.SyncConnected == keeperState) {
if ( Event.EventType.None == eventType ) {
LOG.info( LOG_PREFIX_OF_THREAD + "成功连接上ZK服务器" );
connectedSemaphore.countDown();
}else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
LOG.info(LOG_PREFIX_OF_THREAD + "收到情报，排我前面的家伙已挂，我是不是可以出山了？");
try {
if(checkMinPath()){
getLockSuccess();
}
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}else if ( Event.KeeperState.Disconnected == keeperState ) {
LOG.info( LOG_PREFIX_OF_THREAD + "与ZK服务器断开连接" );
} else if ( Event.KeeperState.AuthFailed == keeperState ) {
LOG.info( LOG_PREFIX_OF_THREAD + "权限检查失败" );
} else if ( Event.KeeperState.Expired == keeperState ) {
LOG.info( LOG_PREFIX_OF_THREAD + "会话失效" );
}
}
}
log配置文件：
# DEFAULT
log4j.rootLogger=INFO,CONSOLE
#
# Log INFO level and above messages to the console
#
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.Threshold=INFO
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} - %m%n
log4j.appender.COMMONSTAT=org.apache.log4j.DailyRollingFileAppender
log4j.appender.COMMONSTAT.Threshold=INFO
log4j.appender.COMMONSTAT.File=/home/zookeeper/zookeeper-test-agent/logs/test.log
log4j.appender.COMMONSTAT.DatePattern='.'yyyy-MM-dd
log4j.appender.COMMONSTAT.layout=org.apache.log4j.PatternLayout
log4j.appender.COMMONSTAT.layout.ConversionPattern=[%d{yyyy-MM-dd HH:mm:ss}] - %m%n
log4j.logger.org.displaytag=WARN
log4j.logger.org.apache.zookeeper=ERROR
log4j.logger.org.springframework=WARN
log4j.logger.org.I0Itec=WARN
log4j.logger.commonStat=INFO,COMMONSTAT
运行结果：
2014-11-19 11:34:10,894 - 【第9个线程】成功连接上ZK服务器
2014-11-19 11:34:10,895 - 【第8个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第1个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第7个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第4个线程】成功连接上ZK服务器
2014-11-19 11:34:10,895 - 【第5个线程】成功连接上ZK服务器
2014-11-19 11:34:10,896 - 【第2个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第10个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第3个线程】成功连接上ZK服务器
2014-11-19 11:34:10,895 - 【第6个线程】成功连接上ZK服务器
2014-11-19 11:34:10,910 - 【第9个线程】节点创建成功, Path: /disLocks, content: 该节点由线程9创建
2014-11-19 11:34:10,912 - 【第9个线程】创建锁路径:/disLocks/sub0000000000
2014-11-19 11:34:10,917 - 【第6个线程】创建锁路径:/disLocks/sub0000000001
2014-11-19 11:34:10,917 - 【第9个线程】子节点中，我果然是老大/disLocks/sub0000000000
2014-11-19 11:34:10,921 - 【第3个线程】创建锁路径:/disLocks/sub0000000002
2014-11-19 11:34:10,922 - 【第6个线程】获取子节点中，排在我前面的/disLocks/sub0000000000
2014-11-19 11:34:10,923 - 【第9个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:10,924 - 【第10个线程】创建锁路径:/disLocks/sub0000000003
2014-11-19 11:34:10,924 - 【第3个线程】获取子节点中，排在我前面的/disLocks/sub0000000001
2014-11-19 11:34:10,928 - 【第10个线程】获取子节点中，排在我前面的/disLocks/sub0000000002
2014-11-19 11:34:10,929 - 【第1个线程】创建锁路径:/disLocks/sub0000000004
2014-11-19 11:34:10,932 - 【第5个线程】创建锁路径:/disLocks/sub0000000005
2014-11-19 11:34:10,935 - 【第1个线程】获取子节点中，排在我前面的/disLocks/sub0000000003
2014-11-19 11:34:10,936 - 【第2个线程】创建锁路径:/disLocks/sub0000000006
2014-11-19 11:34:10,936 - 【第5个线程】获取子节点中，排在我前面的/disLocks/sub0000000004
2014-11-19 11:34:10,940 - 【第4个线程】创建锁路径:/disLocks/sub0000000007
2014-11-19 11:34:10,941 - 【第2个线程】获取子节点中，排在我前面的/disLocks/sub0000000005
2014-11-19 11:34:10,943 - 【第8个线程】创建锁路径:/disLocks/sub0000000008
2014-11-19 11:34:10,944 - 【第4个线程】获取子节点中，排在我前面的/disLocks/sub0000000006
2014-11-19 11:34:10,945 - 【第7个线程】创建锁路径:/disLocks/sub0000000009
2014-11-19 11:34:10,946 - 【第8个线程】获取子节点中，排在我前面的/disLocks/sub0000000007
2014-11-19 11:34:10,947 - 【第7个线程】获取子节点中，排在我前面的/disLocks/sub0000000008
2014-11-19 11:34:12,923 - 【第9个线程】删除本节点：/disLocks/sub0000000000
2014-11-19 11:34:12,926 - 【第6个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:12,928 - 【第6个线程】子节点中，我果然是老大/disLocks/sub0000000001
2014-11-19 11:34:12,930 - 【第9个线程】释放连接
2014-11-19 11:34:12,930 - 【第6个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:14,930 - 【第6个线程】删除本节点：/disLocks/sub0000000001
2014-11-19 11:34:14,937 - 【第3个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:14,941 - 【第3个线程】子节点中，我果然是老大/disLocks/sub0000000002
2014-11-19 11:34:14,943 - 【第6个线程】释放连接
2014-11-19 11:34:14,946 - 【第3个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:16,946 - 【第3个线程】删除本节点：/disLocks/sub0000000002
2014-11-19 11:34:16,949 - 【第10个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:16,951 - 【第10个线程】子节点中，我果然是老大/disLocks/sub0000000003
2014-11-19 11:34:16,953 - 【第3个线程】释放连接
2014-11-19 11:34:16,953 - 【第10个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:18,953 - 【第10个线程】删除本节点：/disLocks/sub0000000003
2014-11-19 11:34:18,957 - 【第1个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:18,960 - 【第10个线程】释放连接
2014-11-19 11:34:18,961 - 【第1个线程】子节点中，我果然是老大/disLocks/sub0000000004
2014-11-19 11:34:18,964 - 【第1个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:20,964 - 【第1个线程】删除本节点：/disLocks/sub0000000004
2014-11-19 11:34:20,967 - 【第5个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:20,969 - 【第5个线程】子节点中，我果然是老大/disLocks/sub0000000005
2014-11-19 11:34:20,971 - 【第1个线程】释放连接
2014-11-19 11:34:20,971 - 【第5个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:22,971 - 【第5个线程】删除本节点：/disLocks/sub0000000005
2014-11-19 11:34:22,974 - 【第2个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:22,978 - 【第2个线程】子节点中，我果然是老大/disLocks/sub0000000006
2014-11-19 11:34:22,979 - 【第5个线程】释放连接
2014-11-19 11:34:22,981 - 【第2个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:24,981 - 【第2个线程】删除本节点：/disLocks/sub0000000006
2014-11-19 11:34:24,985 - 【第4个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:24,989 - 【第2个线程】释放连接
2014-11-19 11:34:24,989 - 【第4个线程】子节点中，我果然是老大/disLocks/sub0000000007
2014-11-19 11:34:24,995 - 【第4个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:26,995 - 【第4个线程】删除本节点：/disLocks/sub0000000007
2014-11-19 11:34:26,998 - 【第8个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:27,000 - 【第8个线程】子节点中，我果然是老大/disLocks/sub0000000008
2014-11-19 11:34:27,004 - 【第8个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:27,004 - 【第4个线程】释放连接
2014-11-19 11:34:29,004 - 【第8个线程】删除本节点：/disLocks/sub0000000008
2014-11-19 11:34:29,007 - 【第7个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:29,009 - 【第7个线程】子节点中，我果然是老大/disLocks/sub0000000009
2014-11-19 11:34:29,010 - 【第8个线程】释放连接
2014-11-19 11:34:29,011 - 【第7个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:31,011 - 【第7个线程】删除本节点：/disLocks/sub0000000009
2014-11-19 11:34:31,017 - 【第7个线程】释放连接
2014-11-19 11:34:31,017 - 所有线程运行结束!
以上是小编为您精心准备的的内容，在的博客、问答、公众号、人物、课程等栏目也有的相关内容，欢迎继续使用右上角搜索按钮进行搜索分布式
， this
， null
lock
zookeeper 分布式队列、zookeeper 分布式锁、zookeeper 分布式事务、zookeeper 伪分布式、zookeeper 分布式调度，以便于您获取更多的相关知识。
时间： 2024-10-25 05:01:41