铭铭erom个人页面-阿里云开发者社区

基于Zookeeper的分布式锁研究

分布式环境下如何保证一个数据在并发的情况下保证同一时间在一台机器只有一个线程在执行？实现一个分布式锁需要解决的问题有以下几个主要点：资源被唯一执行良好的可重入机制：发生死锁情况下可被其他进程可重入良好的性能问题：操作简单，额外的请求不要过多容灾机制：个别的锁机器崩溃也能稳定运行在 ZooKeeper 中，节点类型可以分为持久节点（PERSISTENT ）、临时节点（EPHEMERAL），以及时序节点（SEQUENTIAL ），具体在节点创建过程中，一般是组合使用，可以生成以下 4 种节点类型。持久节点（PERSISTENT）持久顺序节点（PERSISTENT_SEQUENTIAL）临时节点（EPHEMERAL）临时顺序节点（EPHEMERAL_SEQUENTIAL）这里我们就基于临时有序节点去实现分布式锁:1、首先创建一个根节点用于存放分布式锁需要新建的临时节点2、在创建临时有序节点的时候，zk会根据client的创建请求会按顺序依次生成0000001、0000002、0000003这种类似序号的节点3、每个client判断当前自己创建的节点是否为序号最小的节点，是的话就认为拿到了锁，可以执行业务代码，否则就去监听最小节点的删除时间4、当处理完后需要删除节点，这样就能被其他线程监听到，开始竞争锁当然，聪明的同学一看就能看出这个步骤有很多的问题，首先就是最小节点的删除时间会触发大量client的启动执行，需要发送大量的通知，而且如果在去创建监听的时候出现了最小节点已经被删除的情况，这时候永远都收不到最小节点的删除事件，而且如果按不同的资源创建了太多的目录的话，是否会对整个zk的性能产生影响，如果对不同资源采用同一个目录话，那么一个目录下的节点又过多，取子节点的性能又有问题... 所以该分布式锁的实现步骤可以优化一下1、首先不必所有client都去监听最小节点的删除事件，只需要监听比自己稍小点的节点的删除事件即可2、每此执行完处理后可以尝试性删除目录，避免随着时间的增长创建了过多的目录3、另外zookeeper提供的API中设置监听器的操作与读操作是原子执行的，也就是说在读子节点列表时同时设置监听器，保证不会丢失事件，所以可以在创建监听的时候一旦发现监听的节点为空就认为节点已删除，可以拿到锁下面是代码:里面包含了测试main方法https://github.com/fengym201507411/lock package com.fym; import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat; import java.time.LocalDate; import java.time.LocalDateTime; import java.util.List; import java.util.concurrent.*; /** * Created by fengyiming on 2018/12/24. * 基于zookeeper的代码 */ public class LockServiceImpl { private static volatile ZooKeeper zooKeeper; /** * zk连接超时时间/s */ private final static int SESSION_TIMEOUT = 10000; /** * 分布式锁创建key的根路径 */ private final static String PRE_ROOT_PATH = "/zkLockRoot"; /** * /字符 */ private final static String PATH = "/"; static { try { // 连接zookeeper zooKeeper = new ZooKeeper("127.0.0.1:2181", SESSION_TIMEOUT, new Watcher() { @Override public void process(WatchedEvent event) { } }); Stat stat = zooKeeper.exists(PRE_ROOT_PATH, false); if (stat == null) { System.out.println("root path is null,create......"); zooKeeper.create(PRE_ROOT_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } System.out.println("root path ok"); } catch (Exception e) { System.out.println("加载zk信息异常" + e.getMessage()); } } public static void lock(String threadName, String key, long waitSeconds) { try { key = key + LocalDate.now().toString(); System.out.println(threadName + "begin lock"); String path = new StringBuffer(PRE_ROOT_PATH).append(PATH).append(key).toString(); Stat stat = zooKeeper.exists(path, false); if (stat == null) { System.out.println(threadName + "key path is null,create......"); try { zooKeeper.create(path, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } catch (KeeperException k) { System.out.println(threadName + "目录" + path + "已存在，无需重复创建"); } catch (Exception e) { System.out.println(threadName + "create key path error,error message:" + e.getMessage()); } } LocalDateTime begin = LocalDateTime.now(); String lockNodePre = new StringBuffer(PRE_ROOT_PATH).append(PATH).append(key).append(PATH).toString(); String lockNode = zooKeeper.create(lockNodePre, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode .EPHEMERAL_SEQUENTIAL); String lockNodeName = lockNode.substring(lockNodePre.length()); System.out.println(threadName + "创建有序临时节点:" + lockNode + ",节点名称：" + lockNodeName); // 取所有子节点 List<String> subNodes = zooKeeper.getChildren(path, false); System.out.println(threadName + "当前竞争资源下的节点数：" + subNodes.size()); //排序 subNodes.sort(String::compareTo); System.out.println(threadName + "first:" + subNodes.get(0) + ",last: " + subNodes.get(subNodes.size() - 1)); int index = subNodes.indexOf(lockNodeName); String minNodeName = subNodes.get(0); if (!lockNodeName.equals(minNodeName)) { String min1NodeName = subNodes.get(index - 1); String min1NodePath = new StringBuffer(PRE_ROOT_PATH).append(PATH).append(key).append(PATH) .append(min1NodeName).toString(); CountDownLatch countDownLatch = new CountDownLatch(1); System.out.println(threadName + "当前节点" + lockNodeName + "准备监听节点：" + min1NodeName); Stat min1Stat = zooKeeper.exists(min1NodePath, new Watcher() { @Override public void process(WatchedEvent event) { System.out.println(threadName + "节点" + event.getPath() + "，事件 : " + event.getType()); if (event.getType() == Event.EventType.NodeDeleted) { System.out.println(threadName + "节点删除"); countDownLatch.countDown(); } } }); if (min1Stat == null) { System.out.println(threadName + "节点不存在，无需等待，当前节点：" + lockNodeName + "，前一节点：" + min1NodeName); } else { System.out.println(threadName + "------wait-------"); countDownLatch.await(); } } //超时 if(LocalDateTime.now().compareTo(begin.plusSeconds(waitSeconds)) > 0){ throw new Exception("waite time out"); } System.out.println(threadName + "拿到了lock" + lockNodeName + "，do -----"); zooKeeper.delete(lockNode, -1); System.out.println(threadName + "执行完毕，解锁" + lockNodeName + "------"); String lastNodeName = subNodes.get(subNodes.size() - 1); if (lockNodeName.equals(lastNodeName)) { try { zooKeeper.delete(path, -1); System.out.println(threadName + "尝试删除该key目录成功，path" + path); } catch (KeeperException k) { System.out.println(threadName + "尝试删除该key目录，失败：" + k.getMessage()); } catch (Exception e) { System.out.println(threadName + "尝试删除该key目录，失败：" + e.getMessage()); } } } catch (Exception e) { System.out.println(threadName + "lock error" + e.getMessage()); } } public static void main(String[] args) throws InterruptedException { ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()); for (int i = 0; i < 1000; i++) { String name = new StringBuffer("ThreadName[").append(i).append("]").toString(); executor.execute(() -> { LockServiceImpl.lock(name, "firstLock",10); }); } } }

铭铭erom个人页面-阿里云开发者社区

个人介绍

擅长的技术

关于JAVA一些基础知识的总结

基于Zookeeper的分布式锁研究

算法的时间、空间复杂度计算

RabbitMQ知识一览-工作经验

分布式缓存-redis集群部署

关于JAVA一些基础知识的总结

基于Zookeeper的分布式锁研究

算法的时间、空间复杂度计算

RabbitMQ知识一览-工作经验

分布式缓存-redis集群部署

日常项目经验总结

Redis是如何淘汰数据？

ThreadPoolExecutor线程池知识总结

Hive计算时count sum partition by等方法在数据开发时的一些用法

两个并发集合ConcurrentHashMap和CopyOnWriteArrayList知识点汇集

HashMap知识点汇集

Mysql的那些事儿(部分涉及数据库知识总结)

设计模式

JAVA 常见开发问题

一种redis命令搞定基于redis的分布式锁

redis缓存常用姿势

项目里多线程常用姿势，很给力的使用方法

Mysql 查询调优记录

阿里巴巴 Java 开发手册之异常日志（二）-------我的经验

spring定时任务如何运行 schedule解析

刷新缓存如何使用SDK完成