zookeeper——分布式锁，zookeeper
zookeeper——分布式锁，zookeeper
我们可以把zookeeper看做是一个高可用的分布式文件系统。借助于zookeeper的特性，我们可以很方便的实现分布式的一些服务
其中典型的应用场景有：服务配置、分布式锁和分布式队列。
本节，我会讲解分布式锁的实现。
我们借助于zk的短暂有序节点(EPHEMERAL_SEQUENTIAL)和zk的消息通知机制实现分布式锁。
分布式锁的一般实现算法是：
1、在锁znode节点下创建名为lock-的短暂序列znode，并记住它的实际路径名(create操作的返回值)：
如：
id = zookeeper.create(dir + "/" + prefix, data,
getAcl(), EPHEMERAL_SEQUENTIAL);2、查询znode的子节点并设置观察；
3、如果步骤1中锁创建的znode在步骤2中返回的所有子节点中具有最小的顺序号，则获取锁，退出；
4、等待步骤2中所设观察的通知并转到步骤2；
虽然这个算法是正确的，但是还存在一些问题。这个算法会引起“羊群效应”，所谓的“羊群效应”是大量客户端收到同一通知，但是只有少量
客户端需要处理通知。无疑，羊群效应会造成不必要的负载。
改进方法是要优化通知条件，关键在于删除前一个顺序号的节点的时候只通知紧邻的后一个顺序号的节点，而不是删除任一节点都通知。
实现分布式锁的代码如下：
有五个类：
ProtocolSupport
WriteLock
ZNodeName
ZooKeeperOperation
LockListener
/**
*
* 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 com.ali.zk.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 com.ali.zk.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);
}
}
}
}
/**
*
* 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 com.ali.zk.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};//{18,52}
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 = 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 {//如果不存在比当前节点小的节点,那么获得锁,调用callback回调执行
if (isOwner()) {//进一步根据id和ownerid判断是否获得锁
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;
}
//根据dir创建节点
//根据配置“可复原的zookeeper应用”标准
//尝试多次提交
//具体算法看方法
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;
}
}
/**
*
* 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 com.ali.zk.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
*
*/
public 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;
}
}
/**
*
* 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 com.ali.zk.lock;
import org.apache.zookeeper.KeeperException;
/**
* A callback object which can be used for implementing retry-able operations in the
* {@link org.apache.zookeeper.recipes.lock.ProtocolSupport} class
*
*/
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;
}
/**
*
* 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 com.ali.zk.lock;
/**
* This class has two methods which are call
* back methods when a lock is acquired and
* when the lock is released.
*
*/
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();
}
单元测试代码如下：
有两个单元测试类：
ZNodeNameTest
WriteLockTest
/**
*
* 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 com.ali.zk.lock.test;
import junit.framework.TestCase;
import java.util.SortedSet;
import java.util.TreeSet;
import org.junit.Test;
import com.ali.zk.lock.ZNodeName;
/**
* test for znodenames
*/
public class ZNodeNameTest extends TestCase {
@Test
public void testOrderWithSamePrefix() throws Exception {
String[] names = { "x-3", "x-5", "x-11", "x-1" };
String[] expected = { "x-1", "x-3", "x-5", "x-11" };
assertOrderedNodeNames(names, expected);
}
@Test
public void testOrderWithDifferentPrefixes() throws Exception {
String[] names = { "r-3", "r-2", "r-1", "w-2", "w-1" };
String[] expected = { "r-1", "r-2", "r-3", "w-1", "w-2" };
assertOrderedNodeNames(names, expected);
}
protected void assertOrderedNodeNames(String[] names, String[] expected) {
int size = names.length;
assertEquals("The two arrays should be the same size!", names.length, expected.length);
SortedSet<ZNodeName> nodeNames = new TreeSet<ZNodeName>();
for (String name : names) {
nodeNames.add(new ZNodeName(name));
}
int index = 0;
for (ZNodeName nodeName : nodeNames) {
String name = nodeName.getName();
assertEquals("Node " + index, expected[index++], name);
}
}
}
/**
*
* 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 com.ali.zk.lock.test;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
//import org.apache.zookeeper.test.ClientBase;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import com.ali.zk.lock.LockListener;
import com.ali.zk.lock.WriteLock;
/**
* test for writelock
*/
public class WriteLockTest /*extends ClientBase*/ {
protected int sessionTimeout = 10 * 1000;
protected String dir = "/" + getClass().getName();
protected WriteLock[] nodes;
protected CountDownLatch latch = new CountDownLatch(1);
private boolean restartServer = true;
private boolean workAroundClosingLastZNodeFails = true;
private boolean killLeader = true;
@Test
public void testRun() throws Exception {
runTest(3);
}
class LockCallback implements LockListener {
public void lockAcquired() {
latch.countDown();
}
public void lockReleased() {
}
}
protected void runTest(int count) throws Exception {
nodes = new WriteLock[count];
for (int i = 0; i < count; i++) {
//ZooKeeper keeper = createClient();
ZooKeeper keeper = new ZooKeeper("123.123.112.123",sessionTimeout,new Watcher(){
@Override
public void process(WatchedEvent event) {
//do nothing}
}
});
WriteLock leader = new WriteLock(keeper, dir, null);
leader.setLockListener(new LockCallback());
nodes[i] = leader;
leader.lock();
}
// lets wait for any previous leaders to die and one of our new
// nodes to become the new leader
latch.await(30, TimeUnit.SECONDS);
WriteLock first = nodes[0];
dumpNodes(count);
// lets assert that the first election is the leader
Assert.assertTrue("The first znode should be the leader " + first.getId(), first.isOwner());
for (int i = 1; i < count; i++) {
WriteLock node = nodes[i];
Assert.assertFalse("Node should not be the leader " + node.getId(), node.isOwner());
}
if (count > 1) {
if (killLeader) {
System.out.println("Now killing the leader");
// now lets kill the leader
latch = new CountDownLatch(1);
first.unlock();
latch.await(30, TimeUnit.SECONDS);
//Thread.sleep(10000);
WriteLock second = nodes[1];
dumpNodes(count);
// lets assert that the first election is the leader
Assert.assertTrue("The second znode should be the leader " + second.getId(), second.isOwner());
for (int i = 2; i < count; i++) {
WriteLock node = nodes[i];
Assert.assertFalse("Node should not be the leader " + node.getId(), node.isOwner());
}
}
if (restartServer) {
// now lets stop the server
System.out.println("Now stopping the server");
//stopServer();
Thread.sleep(10000);
// TODO lets assert that we are no longer the leader
dumpNodes(count);
System.out.println("Starting the server");
//startServer();
Thread.sleep(10000);
for (int i = 0; i < count - 1; i++) {
System.out.println("Calling acquire for node: " + i);
nodes[i].lock();
}
dumpNodes(count);
System.out.println("Now closing down...");
}
}
}
protected void dumpNodes(int count) {
for (int i = 0; i < count; i++) {
WriteLock node = nodes[i];
System.out.println("node: " + i + " id: " +
node.getId() + " is leader: " + node.isOwner());
}
}
@After
public void tearDown() throws Exception {
if (nodes != null) {
for (int i = 0; i < nodes.length; i++) {
WriteLock node = nodes[i];
if (node != null) {
System.out.println("Closing node: " + i);
node.close();
if (workAroundClosingLastZNodeFails && i == nodes.length - 1) {
System.out.println("Not closing zookeeper: " + i + " due to bug!");
} else {
System.out.println("Closing zookeeper: " + i);
node.getZookeeper().close();
System.out.println("Closed zookeeper: " + i);
}
}
}
}
System.out.println("Now lets stop the server");
//super.tearDown();
}
}
个人觉得算法的核心在WriteLock中LockZooKeeperOperation内部类的execute方法的实现：
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 {//如果不存在比当前节点小的节点,那么获得锁,调用callback回调执行
if (isOwner()) {//进一步根据id和ownerid判断是否获得锁
if (callback != null) {
callback.lockAcquired();
}
return Boolean.TRUE;
}
}
}
}
}
while (id == null);
return Boolean.FALSE;
评论暂时关闭