Java条记（分布式锁）

分布式锁的实现，就需要找到一个可以让所有的 JVM 访问到的公共组件，好比数据库，Redis等。
1 Redis 实现分布式锁

使用 Redis 实现分布式锁，实现原理：

• 根据差异的业务可以定名为差异的锁，好比某条商品库存的主键可视为键名，随机值作为键值，设置在 Redis 中，别的 JVM 访问到了商品库存信息时，要到 Redis 中查看是否有如许对应键值对，如果有则视为该信息已被上锁
  
• 随机值作为键值，它的用于开释锁时的校验。由于有一种环境发生，锁存入 Redis 时需要设置过期时间，这是由于业务中有可能发生异常导致没法正常开释锁，而加上过期时间就可以避免别的进程永远无法得到锁，但这时就又引入另一个题目就是，当业务是正常运作的，但运行时间过长导致键值对失效，因此另一个进程业务可以得到锁即重新设置了一条新的键值对，那么此时运行时间过长的那个业务如果不通过校验逻辑就会开释了锁，导致混乱了
  
• 如许的实现方法主要是由于 Redis 是一个单线程，同时它设置键值对时有一个 NX 特性，它表示当 key 不存在时设置成功，key存在时设置不成功，它是一个原子性操作从而达到上锁操作的原子性
  

1. @GetMapping("/index")
2. public String index() {
3.     String key = "idOfOrderTable";
4.     String value = UUID.randomUUID().toString();
5.     log.info("等待获取锁。。。");
6.     redisOperator.setIfAbsent(key, value, 15);
7.     // 检查是否当前进程成功上锁
8.     String data = redisOperator.get(key);
9.     while (!value.equals(data)) {
10.         redisOperator.setIfAbsent(key, value, 15);
11.         data = redisOperator.get(key);
12.     }
13.     log.info("------- 成功获取锁 -------");
14.     try {
15.         log.info("执行业务逻辑。。。");
16.         Thread.sleep(10000);
17.         log.info("------- 释放锁 -------");
18.         data = redisOperator.get(key);
19.         if (data != null && data.equals(value)) {
20.             redisOperator.del(key);
21.         }
22.     } catch (InterruptedException e) {
23.         e.printStackTrace();
24.     }
25.     return "result";
26. }

复制代码

1.1 Redisson

1. <dependency>
2.         <groupId>org.redisson</groupId>
3.     <artifactId>redisson-spring-boot-starter</artifactId>
4.     <version>3.11.2</version>
5. </dependency>

复制代码

Redisson 是一个高级的分布式协调Redis客服端，例子代码：

1. @Autowired
2. private RedissonClient redissonClient;
4. @GetMapping("/index5")
5. public String index5() {
6.     log.info("------- 进入方法 -------");
7.     RLock rLock = redissonClient.getLock("idOfOrderTable");
8.     try {
9.         rLock.lock(30, TimeUnit.SECONDS); // 传入锁的过期时间
10.         log.info("------- 成功获取锁 -------");
11.         log.info("执行业务逻辑。。。");
12.         Thread.sleep(10000);
13.         log.info("执行完业务逻辑。。。");
14.     } catch (Exception e) {
15.         e.printStackTrace();
16.     } finally {
17.         log.info("------- 释放锁 -------");
18.         rLock.unlock();
19.     }
20.     return "result";
21. }

复制代码

2 Zookeeper 实现分布式锁

使用 Zookeeper 实现分布式锁，要理解 Zookeeper 存储数据的布局，它的数据布局是一个树形文件布局，它拥有两种范例的节点（每个节点都可以存储数据）：

• 长期节点
  
• 瞬时节点：瞬时节点不可有子节点，会话竣事后瞬时节点自动消散
  

Zookeeper 能实现分布式锁主要利用 Zookeeper 的瞬时有序节点的特性，多线程高并发创建有序瞬时节点的方式，根据先后次序生成名称包罗序号的节点，每个节点可以添加观察器监听事件，监听本序号的前一个序号节点的状态。
生成最早节点的线程视为已经得到锁，其余线程在生成节点之后进入休眠状态等待后续激活，当执行完业务逻辑后，删除对应的节点（视为开释锁），此时序号紧接的节点触发了监听事件，该监听事件的主要作用就是重新激活线程的休眠状态（视为得到了锁）。
   观察器只能监控一次，再监控需要重新设置
  引入的包需要和服务器的 Zookeeper 一致：

1. <dependency>
2.     <groupId>org.apache.zookeeper</groupId>
3.     <artifactId>zookeeper</artifactId>
4.     <version>3.4.6</version>
5. </dependency>

复制代码

工具类代码：

1. /**
2. * Zookeeper 线程锁的工具类，实现了资源自动关闭接口和 Zookeeper 的观察器接口
3. */
4. @Slf4j
5. public class ZkLock implements AutoCloseable, Watcher {
7.     private ZooKeeper zooKeeper; // zookeeper 客户端
8.     private String connectStr = "192.168.212.128:2181"; // zookeeper 地址
9.     private String zNode; // 子节点路径
11.     /**
12.      * 构造函数
13.      * @throws IOException
14.      */
15.     public ZkLock() throws IOException {
16.             // 初始化 zookeeper 客户端，因为本类实现了 watcher 接口，所以传入 watcher 时，可以直接传入 this
17.             this.zooKeeper = new ZooKeeper(connectStr, 10000, this);
18.     }
20.     /**
21.      * 获取锁
22.      * @param bussinessCode 业务代码
23.      * @return
24.      */
25.     public Boolean getLock(String bussinessCode) {
26.         // 创建业务节点
27.         String fatherPath = "/" + bussinessCode; // 父节点路径
28.         // 创建持久化业务根节点（就像归类存储文件一样，比如说商品信息，可以先建一个商品根节点作为类别区分，详情信息就全是其子节点）
29.         try {
30.             // 根节点是否存在
31.             Stat stat = zooKeeper.exists(fatherPath, false); // 第二个参数为是否添加观察器
32.             // 如果根节点不存在，则创建一个
33.             if (stat == null){
34.                 // 参数分别为：路径，数据内容，权限（此时设置无需账号即可访问），节点类型（持久节点）
35.                 zooKeeper.create(
36.                         fatherPath, // 节点路径
37.                         fatherPath.getBytes(), // 数据内容，可有可无
38.                         ZooDefs.Ids.OPEN_ACL_UNSAFE, // 权限，设置无需账号即可访问
39.                         CreateMode.PERSISTENT // 节点类型，持久节点
40.                 );
41.             }
42.             // 创建瞬时有序节点
43.             String sonPath = fatherPath + "/" + bussinessCode + "_"; // 瞬时有序节点路径，当节点创建成功后，zookeeper 会在节点名称后添加序号
44.             zNode = zooKeeper.create(
45.                     sonPath,
46.                     sonPath.getBytes(),
47.                     ZooDefs.Ids.OPEN_ACL_UNSAFE,
48.                     CreateMode.EPHEMERAL_SEQUENTIAL // 瞬时有序节点
49.             ); // 返回节点路径，如 /order/order_0001
50.             // 判断本节点是否为序号最小的节点
51.             List<String> childrenNodes = zooKeeper.getChildren(fatherPath, false);
52.             Collections.sort(childrenNodes);
53.             String firstNode = childrenNodes.get(0); // 排序后，最小序号的节点名称，注意不是完整路径，如 order_0001
54.             // 如果是序号最小的节点，直接返回 true，调用方获取 true 就意味着获得了锁
55.             if (zNode.endsWith(firstNode)) {
56.                 return true;
57.             }
58.             // 如果不是序号最小的节点，则添加观察器后进入休眠状态，等待观察器被触发，重新激活，此时调用方会进入阻塞状态，等待获取锁
59.             else {
60.                 String lastNode = firstNode;
61.                 // 添加观察器
62.                 for (String nodeItem : childrenNodes)
63.                 {
64.                     if (zNode.endsWith(nodeItem))
65.                     {
66.                         // 添加前一个节点观察器，观察其是否还存在
67.                         zooKeeper.exists(sonPath + lastNode, true);
68.                         break;
69.                     }
70.                     lastNode = nodeItem;
71.                 }
72.             }
73.             // 将该线程挂起（将该线程加入到本实例对象的等待队列中）
74.             synchronized (this) {
75.                 wait();
76.             }
77.             // 线程挂起被激活后，继续执行操作，返回 true，表示调用方获取锁
78.             return true;
79.         } catch (Exception e) {
80.             e.printStackTrace();
81.         }
82.         return false;
83.     }
85.     /**
86.      * 实现资源自动关闭接口，这样调用方就可以直接使用 try (ZkLock zklock = new ZkLock()) {...} 的形式上锁，这样执行完代码块后，会自动调用 close 方法
87.      * @throws Exception
88.      */
89.     @Override
90.     public void close() throws Exception {
91.         // 删除本节点，从而触发观察器逻辑
92.         // 第二个参数为版本号，用于校验的，这里直接使用 -1，表示匹配所有版本号
93.         zooKeeper.delete(zNode, -1);
94.         zooKeeper.close();
95.         log.info("------- 释放锁 -------");
96.     }
98.     /**
99.      * 实现 zookeeper 观察器接口
100.      * @param watchedEvent
101.      */
102.     @Override
103.     public void process(WatchedEvent watchedEvent) {
104.         // 当检测到节点被删除，激活线程（释放锁，并将本实例对象的等待列表中随意一个线程提取到入口列表中去）
105.         // 正因为是随机的，所以使用本工具类上锁时，每次需要新创建一个 ZkLock 对象，而不能使用一个公共变量存储 ZkLock 对象这样的形式去调用
106.         if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
107.             synchronized (this) {
108.                 this.notify();
109.             }
110.         }
111.     }
112. }

复制代码

调用方：

1. @GetMapping("/index")
2.     public String index() {
3.         log.info("------- 进入方法 -------");
4.         try (ZkLock zkLock = new ZkLock()) {
5.             Boolean getLock = zkLock.getLock("order");
6.             if (getLock) {
7.                 log.info("------- 成功获取锁 -------");
8.                 log.info("执行业务逻辑。。。");
9.                 Thread.sleep(10000);
10.                 log.info("执行完业务逻辑。。。");
11.             } else {
12.                 log.info("------- 获取锁失败 -------");
13.             }
14.         } catch (Exception e) {
15.             e.printStackTrace();
16.         }
17.         return "result";
18.     }

复制代码

2.1 curator

1. <dependency>
2.     <groupId>org.apache.curator</groupId>
3.     <artifactId>curator-recipes</artifactId>
4.     <version>4.2.0</version>
5. </dependency>

复制代码

curator 是 apache 封装 Zookeeper 的一个高级包，例子代码如下：

1. @GetMapping("/index3")
2.     public String index3() {
3.         log.info("------- 进入方法 -------");
4.         // 获取连接
5.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); // 重试策略
6.         CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.212.128:2181", retryPolicy); // 客户端连接
7.         client.start();
8.         InterProcessMutex lock = new InterProcessMutex(client, "/order");
9.         // 获取锁，传入超时时间
10.         try {
11.             if (lock.acquire(30, TimeUnit.SECONDS)) {
12.                 log.info("------- 成功获取锁 -------");
13.                 log.info("执行业务逻辑。。。");
14.                 Thread.sleep(10000);
15.                 log.info("执行完业务逻辑。。。");
16.             }
17.         } catch (Exception e) {
18.             e.printStackTrace();
19.         }
20.         finally {
21.             try {
22.                 lock.release();
23.                 log.info("------- 释放锁 -------");
24.             } catch (Exception e) {
25.                 e.printStackTrace();
26.             }
27.         }
28.         client.close();
29.         return "result";
30.     }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。