实现分布式锁的方式主要有Redis和Zookeeper, Zookeeper实现分布式锁相对于Redis比较简单, Zookeeper有一个特性: 多个线程在Zookeeper里创建同一个节点时, 只会有一个线程执行成功.
Zookeeper的节点分为两大类: 临时节点, 持久化节点
临时节点: 会话失效或连接异常时, 节点会被自动删除;
持久化节点: 客户端一旦创建节点, 即使会话结束或发生异常节点也不会被删除, 只有客户端主动请求删除.
有序节点: 在创建节点时会有一个序号, 序号自增. 分为临时有序节点, 持久化有序节点
实现分布式锁
使用临时顺序节点这一特性
- 客户端A发起加锁请求, 向Zookeeper中的order_locak节点下创建有序临时节点order_0000000;
- 判断客户端A是不是order_locak节点下序号最小的节点(当前只有一个节点, 该节点本身就是最小节点), 是: 加锁 ----> 执行业务逻辑----> 执行完毕, 释放锁(删除节点).
- 客户端B发起加锁请求, 向Zookeeper中的order_locak节点下创建临时节点order_0000001, 序号自增;
- . 假设现在客户端A还没释放锁, 那么根节点下有两个节点, 使当前节点监听上一个节点, 一旦监听到上一个节点被删除(释放锁), 判断客户端B是不是order_locak节点下序号最小的节点, 是: 加锁, 重复上面的流程, 否: 继续等待判断是不是当前节点是不是序号最小的节点.
代码实现
org.apache.zookeeper zookeeper3.5.4-beta
实现:
@Slf4j
public class ZookeeperLock implements AutoCloseable, Watcher {
private ZooKeeper zooKeeper;
private String currentNode;
public ZookeeperLock() {
try {
//参数：链接字符串 ， 会话超时 ， 监听器
this.zooKeeper = new ZooKeeper("192.168.237.204:2181", 10000, this);
} catch (IOException e) {
e.printStackTrace();
}
}
public boolean getLock(String code) {
try {
// 1. 创建根节点，如果不存在，stat为null 就创建
String path = "/" + code;
Stat exists = zooKeeper.exists(path, false);
// 创建节点
if (null == exists) {
// 节点路径 ，节点值 ，权限：不需要账户密码就能链接 ， 创建模式：顺序临时节点
zooKeeper.create(path, code.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
// 2. 进来一个线程，就为线程在zookeeper创建一个临时有序节点
// 把当前节点保存为成员变量，后面用来做判断
currentNode = zooKeeper.create(path + path, code.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// 3. 对所有的子节点进行排序
List children = zooKeeper.getChildren(path, false);
Collections.sort(children);
// 4. 如果排序过的children中的第一个和 currentNode 一致，拿到锁
String firstNode = children.get(0);
if (currentNode.endsWith(firstNode)) {
return true;
}
// 5. 如果不是第一个节点，需要监听前一个节点
// 用一个临时变量记录当前节点的上一个节点
String previousNode = firstNode;
for (String node : children) {
if (currentNode.endsWith(node)) {
// 第一个参数是监听的节点，第二个参数是是否要监听，zooKeeper在初始化的时候设置好了监听器
log.info("监听上一个节点：{}", node);
zooKeeper.exists(path + "/" + previousNode, true);
} else {
//把children中的节点复制给上一个节点
previousNode = node;
}
}
// 等待被唤醒
synchronized (this) {
//wait会释放锁
wait();
}
// 到这里说明被唤醒，说明获取到锁
log.info("拿到锁：{}", currentNode);
return true;
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
return false;
}
@Override
public void close() {
// 释放锁
log.info("释放节点：{}", currentNode);
if (null != currentNode) {
try {
zooKeeper.delete(currentNode, -1);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
}
@Override
public void process(WatchedEvent watchedEvent) {
// 当节点被释放 立刻被监听到
if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
synchronized (this) {
//唤醒等待的线程
log.info("当前节点：{}，唤醒", watchedEvent.getPath());
notify();
}
}
}
}
测试:
public class Test {
public static void main(String[] args) {
try {
testZK();
} catch (Exception e) {
e.printStackTrace();
}
}
public static void testZK() throws Exception {
for(int i = 0 ; i < 10 ; i++){
new Thread(()->{
ZookeeperLock zookeeperLock = new ZookeeperLock();
boolean getLock = zookeeperLock.getLock("order");
System.out.println("是否获取到锁："+getLock);
zookeeperLock.close();
}).start();
}
Thread.sleep(5000);
}
}
Apache提供了便于实现分布式锁的组件
org.apache.curator
curator-recipes
4.2.0
配置:
@Configuration
public class AppConfig {
@Bean(initMethod = "start",destroyMethod = "close")
public CuratorFramework curatorFramework(){
//重试策略
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
//创建客户端
CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.237.204:2181", retryPolicy);
return client;
}
}
测试:
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = DistributedLockApplication.class)
public class HuzzTest {
@Autowired
private CuratorFramework curatorFramework;
@Test
public void testCurator() throws Exception {
for(int i = 0 ; i < 10 ; i++){
new Thread(()-> testCuratorLock()).start();
}
Thread.sleep(5000);
}
public void testCuratorLock(){
//分布式锁
InterProcessMutex lock = new InterProcessMutex(curatorFramework, "/order");
try {
if ( lock.acquire(1, TimeUnit.SECONDS) ){
//处理业务逻辑
log.info("获取到锁");
}
} catch (Exception e) {
e.printStackTrace();
} finally{
try {
//释放锁
log.info("释放锁");
lock.release();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
