1. zk分布式锁原理
（参考：https://blog.csdn.net/xuefeng0707/article/details/80588855）
zk使用临时顺序节点+Watch机制实现分布式锁：
- 加锁：创建临时顺序节点，如果该节点是最小节点则获取到锁，否则对上一个临时节点注册监听。
- 解锁：删除节点。
下面我们详细分析下加锁和解锁的过程，暂不考虑可重入性和锁超时，参考了curator源码。
1.1 加锁
过程1：创建临时顺序节点
举例子说明吧，在下面将要用到的例子中，我们会在“/lock/test/testZkLock/”目录下创建临时顺序子节点，子节点的名字不需要我们去命名，curator框架会自动命名为“lock-”，创建完临时顺序节点之后我们能看到这样的结构：
image.png
如果是多个进程或者线程竞争锁资源，就会在“/lock/test/testZkLock/”目录下依次创建临时顺序节点，在一次运行中我们启动3个线程，看到创建了3个临时节点：
image.png
过程2：竞争锁
在这个过程中会判断当前线程创建的临时顺序节点是不是序号最小的那个，如果是则获取到锁、继续执行后面的任务，否则的话对比当前序号小的节点注册监听。举例说明，假设在某一瞬间，某一共享资源下创建的临时节点是这样的：
lock_001
lock_002
lock_003
那么在加锁时，
lock_001的线程可以获取到锁，
lock_002在lock_001上注册Watch监听，然后wait()进入等待，
同理lock_003在lock_002上注册监听。
image.png
image.png
curator源码：
image.png
1.2 解锁
不考虑可重入的话，解锁直接删除临时节点：
client.delete().guaranteed().forPath(ourPath);
当该临时节点被删除时，注册在该节点上的Watcher会监听到zk事件，然后执行process()方法：
image.png
notifyFromWatcher()其实就一行代码：
image.png
仍然举上面的例子，假设目前临时有序节点为：lock_001、lock_002、lock_003
当lock_001因为释放锁而被删除的时候，会触发zk通知机制，那么lock_002节点会收到事件通知，在收到通知的时候执行“notifyAll();”，通知当前JVM进程中所有处于wait状态的线程起来抢锁。
分布式环境下有多个进程，为什么只通知本进程的就可以？
假设：
节点2监听节点1是在进程A中进行的；
节点3监听节点2是在进程B中进行的；
当节点1删除的时候，通知到进程A，节点2就能收到；
而节点2删除的时候，通知到进程B，节点3就能收到。
意思是谁注册的监听就通知谁。
2. 使用Curator实现分布式锁
Curator是Netfix公司开源的一套ZooKeeper客户端框架，对zk底层的连接、监听等进行了良好的封装，并且还提供了分布式锁API，因此我们不必自己实现上述复杂的理论，直接使用curator框架即可。
Curator分布式锁是一种可重入锁，实现了分布式的AQS，使用ConcurrentMap实现了一个类似ThreadLocal的功能，把线程（Thread.currentThread()）作为key，锁作为value，在加锁时，如果在key中找到该线程，就对value加1，解锁时减1，减到0的时候删除临时节点、移除map中的该线程。
首先定义锁接口：
package utils.distributed.lock;
public interface DistributedLock {
/**
* 阻塞式锁
* @return
*/
void lock();
/**
* 非阻塞式锁
* @return
*/
boolean tryLock();
/**
* 带超时时间的阻塞式锁
* @param timeout
* @return
*/
boolean tryLock(long timeout);
/**
* 解锁
*/
void unLock() throws Exception;
/**
* 释放资源
*/
void shutdown();
}
然后实现一个zk锁：
package utils.distributed.lock;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
/**
* zk分布式锁
*/
public class ZKLock implements DistributedLock {
private CuratorFramework client;
private InterProcessMutex lock;
public ZKLock(String host, String bizType, String lockKey) {
client = CuratorFrameworkFactory.newClient(host,
new ExponentialBackoffRetry(ZKLockConstant.BASE_SLEEP_TIME_MS, ZKLockConstant.MAX_RETRIES));
client.start();
String path = ZKLockConstant.ZK_SEPERATOR + StringUtils.join(Arrays.asList(ZKLockConstant.ZK_LOCK_BASE_PREFIX, bizType, lockKey), ZKLockConstant.ZK_SEPERATOR);
lock = new InterProcessMutex(client, path);
}
public void lock() {
try {
lock.acquire();
} catch (Exception e) {
e.printStackTrace();
}
}
public boolean tryLock() {
return tryLock(0);
}
public boolean tryLock(long timeout) {
try {
return lock.acquire(timeout, TimeUnit.MILLISECONDS);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
public void unLock() throws Exception {
lock.release();
}
public void shutdown() {
client.close();
}
}
3. 测试
3.1 测试lock
package mytest.distributed.lock;
import utils.distributed.lock.ZKLock;
import java.util.concurrent.CountDownLatch;
public class ZkLockTest {
public static void main(String[] args) {
String zkHost = "192.168.160.128:2181";
String bizType = "test";
String lockKey = "testZkLock";
ZKLock zkLock = new ZKLock(zkHost, bizType, lockKey);
// 启动3个线程模拟分布式锁竞争
CountDownLatch conutDownLatch = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {
new Thread(() -> {
testZkLock(zkLock);
conutDownLatch.countDown();
}).start();
}
try {
conutDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
zkLock.shutdown();
}
private static void testZkLock(ZKLock zkLock) {
System.out.println("######## 开始加锁，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
zkLock.lock();
try {
System.out.println("######## 加锁成功，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} finally {
try {
zkLock.unLock();
System.out.println("######## 解锁成功，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
} catch (Exception e) {
System.out.println("######## 解锁失败，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
}
}
}
}
打印结果：
image.png
全部成功了。
3.2 测试tryLock()
复用上面的main方法，将testZkLock换成testZkLockTryLock：
private static void testZkLockTryLock(ZKLock zkLock) {
System.out.println("######## 开始加锁，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
if (!zkLock.tryLock()) {
System.out.println("######## 加锁失败，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
return;
}
try {
System.out.println("######## 加锁成功，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
} finally {
try {
zkLock.unLock();
System.out.println("######## 解锁成功，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
} catch (Exception e) {
System.out.println("######## 解锁失败，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
}
}
}
打印结果：
image.png
因为只尝试一次，所以有失败的情况发生。
3.3 测试tryLock(timeout)
private static void testZkLockTryLockTimeOut(ZKLock zkLock) {
System.out.println("######## 开始加锁，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
if (!zkLock.tryLock(3000)) {
System.out.println("######## 加锁失败，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
return;
}
try {
System.out.println("######## 加锁成功，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} finally {
try {
zkLock.unLock();
System.out.println("######## 解锁成功，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
} catch (Exception e) {
System.out.println("######## 解锁失败，线程信息：" + Thread.currentThread().getId() + "-" + Thread.currentThread().getName());
}
}
}
打印结果：
image.png
线程18获取锁之后要持有5s才会释放，而线程16、17加锁等待时间只有3s，因此会超时。
4. 分布式锁注解
实现一个基于注解的分布式锁：
4.1 定义分布式锁注解
package utils.distributed.annotation;
import java.lang.annotation.*;
/**
* ZK分布式锁注解
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ZkLock {
/**
* zk地址
* @return
*/
String zkHost();
/**
* 业务类型
*/
String bizType();
/**
* 锁名称
*
* @return
*/
String lockKey();
/**
* 超时时间
* @return
*/
long timeout();
}
4.2 定义分布式锁切面
package utils.distributed.advice;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import utils.distributed.annotation.ZkLock;
import utils.distributed.lock.ZKLock;
/**
* @Description zk分布式锁切面
* @Author lilong
* @Date 2019-04-08 13:50
*/
@Component
@Aspect
public class ZkLockAspectAdvice {
@Around(value = "@annotation(utils.distributed.annotation.ZkLock) && @annotation(zkLock)")
public Object process(ProceedingJoinPoint pjp, ZkLock zkLock) throws Throwable {
ZKLock lock = new ZKLock(zkLock.zkHost(), zkLock.bizType(), zkLock.lockKey());
boolean acquired = false;
try {
acquired = lock.tryLock(zkLock.timeout());
if (acquired) {
return pjp.proceed();
} else {
System.out.println("######## 加锁失败，线程信息：" + Thread.currentThread().getId() + "-"
+ Thread.currentThread().getName());
return null;
}
} finally {
if (acquired) {
lock.unLock();
}
}
}
}
4.3 使用注解
@Override
@ZkLock(zkHost = "192.168.160.128:2181", bizType = "test", lockKey = "queryKeyValue", timeout = 3000)
public KeyValueJsonPO queryKeyValue(String bizType, String key) {
return keyValueJsonPOMapper.queryKeyValue(bizType, key);
}
网友评论