(Java)基于Zookeeper的分布式锁

上篇博客中讲了Zookeeper的基本概念和使用，这篇博客讲基于Zookeeper的分布式锁

01. 为什么使用分布式锁

一个方法在高并发情况下的同一时间只能被同一个线程执行，在传统单体应用单机部署的情况下，可以使用Java并发处理相关的API(如ReentrantLcok或synchronized)进行互斥控制。但是，随着业务发展的需要，原单体单机部署的系统被演化成分布式系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题.

02. 基于zookeeper的分布式锁原理

一个分布式锁对应ZooKeeper的一个节点，每个需要获取这个分布式锁的客户端线程在这个节点下创建一个临时有序节点，此时有两种情况：

1. 创建的临时顺序节点是文件夹下的第一个节点，则认为是获取分布式锁成功。

2. 创建的临时顺序节点不是文件夹下的第一个节点，则认为当前锁已经被另一个客户端线程获取，此时需要进入阻塞状态，等待节点顺序中的前一个节点释放锁的时候唤醒当前线程。
   

03. 代码实现

3.1 核心类

3.2 问题描述（场景分析）

这里模拟修改用户积分(score)的操作，如果多个线程同时修改一个用户的积分就会有并发问题，因此需要加入锁来进行控制。

3.3 代码实现

具体步骤：

1. 创建UserDao，提供getScoreFromDb和updateScore两个方法；
2. 创建DistributedLockDemo类以及main方法；
3. 完善main方法：
   1. 实例化zookeeper客户端CuratorFramework
   2. 实例化分布式锁InterProcessLock
   3. 实例化UserDao
   4. 并发执行100遍修改用户积分（score）操作
   5. 输出用户最终的积分（score）
4. 执行查看结果；
5. 去除代码中锁的逻辑再次执行并查看结果。

• UserDao

/**
 * @author: sz
 * @date: Created in 2021/2/14
 * @description: 用户dao层
 * @version: 1.0
 */

public class UserDao {
    private int score = 0;

    // 模拟从数据库获取用户积分
    public int getScoreFromDb(){
        //模拟网络请求耗时1毫秒
        try {
            Thread.sleep(1L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return this.score;
    }


    // 模拟更新数据库里的用户积分
    public void updateScore(int score){
        // 模拟网络请求耗时1毫秒
        try {
            Thread.sleep(1L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        this.score = score;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

• DistributedLockDemo]

/**
 * @author: sz
 * @date: Created in 2021/2/14
 * @description: curator实现的分布式锁使用实例
 * @version: 1.0
 */

public class DistributedLockDemo {
    public static void main(String[] args) {
        // 1-构造并启动客户端
        RetryPolicy retryPolicy = new RetryNTimes(10,1000);//构造超时重试策略，每1s重试一次，重试10次
        CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181",retryPolicy);
        client.start();

        // 2-构造锁
        InterProcessLock lock = new InterProcessMutex(client,"/user/1/update");

        // 3-构造userDao
        UserDao userDao = new UserDao();


        // 4-并发修改100遍
        for (int i = 0; i < 100; i++) {
            // 开启线程
            new Thread(()->{
                try {
                    // 开启锁
                    lock.acquire();
                    // 获取积分
                    int score = userDao.getScoreFromDb();
                    // 积分+1
                    score++;
                    //更新数据库
                    userDao.updateScore(score);
                } catch (Exception e) {
                    e.printStackTrace();
                }finally {
                    try {
                        lock.release();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }

            }).start();
        }
        // 睡眠5s等待任务执行完成
        try {
            Thread.sleep(5000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        //关闭客户端
        client.close();
        System.out.println("完成后，结果："+userDao.getScoreFromDb());


    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

• 加分布式锁的执行结果：
  

• 去掉分布式锁，执行结果：
  

可见分布式锁能够较好的处理并发问题。