闽公网安备 35020302035485号

6.具备可重入特性,一个线程中可以多次获取同一把锁,比如一个线程在执行一个带锁的方法,该方法中又调用了另一个需要相同锁的方法,则该线程可以直接执行调用的方法,而无需重新获得锁

5.线程A处理完,删除自己的节点,线程B监听到变更事件,判断自己是最小的节点,获得锁
<!-- zookeeper 分布式锁、注意zookeeper版本 这里对应的是3.4.6-->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.10.0</version>
</dependency>
首先,我们看下InterProcessLock接口中的几个方法:/** * 堆代码 duidaima.com * 获取锁、阻塞等待、可重入 */ public void acquire() throws Exception; /** * 获取锁、阻塞等待、可重入、超时则获取失败 */ public boolean acquire(long time, TimeUnit unit) throws Exception; /** * 释放锁 */ public void release() throws Exception; /** * Returns true if the mutex is acquired by a thread in this JVM */ boolean isAcquiredInThisProcess();获取锁:
//获取锁
public void acquire() throws Exception
{
if ( !internalLock(-1, null) )
{
throw new IOException("Lost connection while trying to acquire lock: " + basePath);
}
}
private boolean internalLock(long time, TimeUnit unit) throws Exception
{
/*
实现同一个线程可重入性,如果当前线程已经获得锁,
则增加锁数据中lockCount的数量(重入次数),直接返回成功
*/
//获取当前线程
Thread currentThread = Thread.currentThread();
//获取当前线程重入锁相关数据
LockData lockData = threadData.get(currentThread);
if ( lockData != null )
{
//原子递增一个当前值,记录重入次数,后面锁释放会用到
lockData.lockCount.incrementAndGet();
return true;
}
//尝试连接zookeeper获取锁
String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
if ( lockPath != null )
{
//创建可重入锁数据,用于记录当前线程重入次数
LockData newLockData = new LockData(currentThread, lockPath);
threadData.put(currentThread, newLockData);
return true;
}
//获取锁超时或者zk通信异常返回失败
return false;
}
Zookeeper获取锁实现: String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
{
//堆代码 duidaima.com
//获取当前时间戳
final long startMillis = System.currentTimeMillis();
//如果unit不为空(非阻塞锁),把当前传入time转为毫秒
final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;
//子节点标识
final byte[] localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
//尝试次数
int retryCount = 0;
String ourPath = null;
boolean hasTheLock = false;
boolean isDone = false;
//自旋锁,循环获取锁
while ( !isDone )
{
isDone = true;
try
{
//在锁节点下创建临时且有序的子节点,例如:_c_008c1b07-d577-4e5f-8699-8f0f98a013b4-lock-000000001
ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
//如果当前子节点序号最小,获得锁则直接返回,否则阻塞等待前一个子节点删除通知(release释放锁)
hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
}
catch ( KeeperException.NoNodeException e )
{
//异常处理,如果找不到节点,这可能发生在session过期等时,因此,如果重试允许,只需重试一次即可
if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
{
isDone = false;
}
else
{
throw e;
}
}
}
//如果获取锁则返回当前锁子节点路径
if ( hasTheLock )
{
return ourPath;
}
return null;
}
private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
{
boolean haveTheLock = false;
boolean doDelete = false;
try
{
if ( revocable.get() != null )
{
client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
}
//自旋获取锁
while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
{
//获取所有子节点集合
List<String> children = getSortedChildren();
//判断当前子节点是否为最小子节点
String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
//如果是最小节点则获取锁
if ( predicateResults.getsTheLock() )
{
haveTheLock = true;
}
else
{
//获取前一个节点,用于监听
String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
synchronized(this)
{
try
{
//这里使用getData()接口而不是checkExists()是因为,如果前一个子节点已经被删除了那么会抛出异常而且不会设置事件监听器,而checkExists虽然也可以获取到节点是否存在的信息但是同时设置了监听器,这个监听器其实永远不会触发,对于Zookeeper来说属于资源泄露
client.getData().usingWatcher(watcher).forPath(previousSequencePath);
if ( millisToWait != null )
{
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
//如果设置了获取锁等待时间
if ( millisToWait <= 0 )
{
doDelete = true; // 超时则删除子节点
break;
}
//等待超时时间
wait(millisToWait);
}
else
{
wait();//一直等待
}
}
catch ( KeeperException.NoNodeException e )
{
// it has been deleted (i.e. lock released). Try to acquire again
//如果前一个子节点已经被删除则deException,只需要自旋获取一次即可
}
}
}
}
}
catch ( Exception e )
{
ThreadUtils.checkInterrupted(e);
doDelete = true;
throw e;
}
finally
{
if ( doDelete )
{
deleteOurPath(ourPath);//获取锁超时则删除节点
}
}
return haveTheLock;
}
释放锁: public void release() throws Exception
{
Thread currentThread = Thread.currentThread();
LockData lockData = threadData.get(currentThread);
//没有获取锁,你释放个球球,如果为空抛出异常
if ( lockData == null )
{
throw new IllegalMonitorStateException("You do not own the lock: " + basePath);
}
//获取重入数量
int newLockCount = lockData.lockCount.decrementAndGet();
//如果重入锁次数大于0,直接返回
if ( newLockCount > 0 )
{
return;
}
//如果重入锁次数小于0,抛出异常
if ( newLockCount < 0 )
{
throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
}
try
{
//释放锁
internals.releaseLock(lockData.lockPath);
}
finally
{
//移除当前线程锁数据
threadData.remove(currentThread);
}
}
/**
* 堆代码 duidaima.com
* 基于curator的zookeeper分布式锁
*/
public class CuratorUtil {
private static String address = "192.168.1.180:2181";
public static void main(String[] args) {
//1、重试策略:初试时间为1s 重试3次
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
//2、通过工厂创建连接
CuratorFramework client = CuratorFrameworkFactory.newClient(address, retryPolicy);
//3、开启连接
client.start();
//4 分布式锁
final InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");
//读写锁
//InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/readwriter");
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
fixedThreadPool.submit(new Runnable() {
@Override
public void run() {
boolean flag = false;
try {
//尝试获取锁,最多等待5秒
flag = mutex.acquire(5, TimeUnit.SECONDS);
Thread currentThread = Thread.currentThread();
if(flag){
System.out.println("线程"+currentThread.getId()+"获取锁成功");
}else{
System.out.println("线程"+currentThread.getId()+"获取锁失败");
}
//模拟业务逻辑,延时4秒
Thread.sleep(4000);
} catch (Exception e) {
e.printStackTrace();
} finally{
if(flag){
try {
mutex.release();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
});
}
}
}
这里我们开启5个线程,每个线程获取锁的最大等待时间为5秒,为了模拟具体业务场景,方法中设置4秒等待时间。开始执行main方法,通过ZooInspector监控/curator/lock下的节点如下图: