ZooKeeper應用場景-分散式鎖設計
阿新 • • 發佈:2019-01-05
分散式鎖,這個主要得益於ZooKeeper為我們保證了資料的強一致性。鎖服務可以分為兩類,一個是保持獨佔,另一個是控制時序。
- 所謂保持獨佔,就是所有試圖來獲取這個鎖的客戶端,最終只有一個可以成功獲得這把鎖。通常的做法是把zk上的一個znode看作是一把鎖,通過create znode的方式來實現。所有客戶端都去建立 /distribute_lock 節點,最終成功建立的那個客戶端也即擁有了這把鎖。
- 控制時序,就是所有檢視來獲取這個鎖的客戶端,最終都是會被安排執行,只是有個全域性時序了。做法和上面基本類似,只是這裡 /distribute_lock 已經預先存在,客戶端在它下面建立臨時有序節點(這個可以通過節點的屬性控制:CreateMode.EPHEMERAL_SEQUENTIAL來指定)。Zk的父節點(/distribute_lock)維持一份sequence,保證子節點建立的時序性,如node_1,node_2,從而也形成了每個客戶端的全域性時序。(http://www.cnblogs.com/ggjucheng/p/3352614.html)
下面我們看看ZooKeeper的原始碼包的recipe目錄下一個互斥鎖的設計:
ProtocolSupport提供了與ZooKeeper有關的輔助類操作。例如retry重連,目錄檢查。
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); } } } }
ZooKeeperOperation:根據使用場景實現execute操作,如WriteBlock的使用。
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; }
LockListener:在使用場景中使用,控制
ackage org.apache.zookeeper.recipes.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();
}
WriteLock需要注意:
1、內部類LockZooKeeperOperation根據場景實現了ZooKeeperOperation
2、LockWatcher實現了Watcher介面
3、synchronized boolean lock() 。1)確保 /distribute_lock 已經預先存在 2)retry嘗試獲取鎖,同步獲取鎖。execute實現原理如下圖
4、synchronized void unlock() 。1)刪除node_i,2)同步釋放鎖。
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 = 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;
}
}