本文主要包含zookeeper 分布式锁,zookeeper 锁,zookeeper分布式,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 require