五、ZK实战之高性能分布式锁_java zk 分布式锁 生产

前言：

数据库和redis也可实现分布式锁，本文只将三者做一个对比，不实现数据库和redis，将实现ZK的分布式锁

1、数据库、redis、zk实现分布式锁的特点

2、ZK实现高性能分布式锁

2.1实现思路

2.2 具体实现code

• lock接口

package cn.enjoy.zk;

/*
 * @author wangle25
 * @description lock接口
 * @date 20:38 2020-07-12
 * @param 
 * @return 
 **/

public interface Lock {
    //获取到锁的资源
    public void getLock();
    // 释放锁
    public void unLock();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

• 抽象接口

package cn.enjoy.zk;

/*
 * @author wangle25
 * @description 抽象接口
 * @date 20:39 2020-07-12
 * @param 
 * @return 
 **/

public abstract class AbstractLock  implements  Lock{

    public void getLock() {

        // 任务通过竞争获取锁才能对该资源进行操作(①竞争锁)；
        // 当有一个任务在对资源进行更新时（②占有锁），
        // 其他任务都不可以对这个资源进行操作（③任务阻塞），
        // 直到该任务完成更新(④释放锁)
        // 尝试获得锁资源

        // 尝试获取锁
        if (tryLock()) {
            System.out.println("##获取lock锁的资源####");
        } else {

            // 任务阻塞
            waitLock();

            // 重新获取锁资源
            getLock();
        }
    }

    // ②占有锁
    public abstract boolean tryLock();

    // 等待
    public abstract void waitLock();
}

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

• zk分布式锁抽象类

package cn.enjoy.zk;

import org.I0Itec.zkclient.ZkClient;


/**
 * @author wangle25
 * @description ，zk分布式锁抽象类，将重复代码写入子类中，模版方法
 * @date 20:43 2020-07-12
 * @param 
 * @return 
 **/

public abstract class ZookeeperAbstractLock extends AbstractLock {
    // zk连接地址
    private static final String CONNECTSTRING = "127.0.0.1:2181";
    // 创建zk连接
    protected ZkClient zkClient = new ZkClient(CONNECTSTRING);
    protected static final String PATH = "/lock";

    protected static final String PATH2 = "/lock2";
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

• 实现类

package scmp.portal.zkLock;

import org.I0Itec.zkclient.IZkDataListener;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/*
 * @author wangle25
 * @description 基于zk的分布式锁的实现
 * @date 20:21 2020-07-12
 * @param
 * @return
 **/

public class ZookeeperDistrbuteLock2 extends ZookeeperAbstractLock {

    // 未获得锁时阻塞等待工具
    private CountDownLatch countDownLatch= null;

    // 当前请求的节点前一个节点
    private String beforePath;

    // 当前请求的节点
    private String currentPath;

    // 构造函数，默认的去获取一次锁
    public ZookeeperDistrbuteLock2() {
        if (!this.zkClient.exists(PATH2)) {
            this.zkClient.createPersistent(PATH2);
        }
    }


    @Override
    public boolean  tryLock() {
        // 如果currentPath为空则为第一次尝试加锁，第一次加锁赋值currentPath
        if(currentPath == null || currentPath.length()<= 0){
            // 创建一个临时顺序节点
            currentPath = this.zkClient.createEphemeralSequential(PATH2 + '/',"lock");
        }
        // 获取所有临时节点并排序，临时节点名称为自增长的字符串如：0000000400
        List<String> childrens = this.zkClient.getChildren(PATH2);

        // 子节点进行排序
        Collections.sort(childrens);

        // 如果当前节点在所有节点中排名第一则获取锁成功
        if (currentPath.equals(PATH2 + '/'+childrens.get(0))) {
            return true;
        }
        // 如果当前节点在所有节点中排名中不是排名第一，则获取前面的节点名称，并赋值给beforePath
        else {
            int wz = Collections.binarySearch(childrens,
                    currentPath.substring(7));
            beforePath = PATH2 + '/'+childrens.get(wz-1);
        }
        return false;

    }

    @Override
    public void waitLock() {

        // 监听前一个节点的删除事件
        IZkDataListener listener = new IZkDataListener() {
            public void handleDataDeleted(String dataPath) throws Exception {
                // 前一个节点被删除后，唤醒当前节点
                if(countDownLatch!=null){
                    countDownLatch.countDown();
                }
            }
        };

        // 给排在前面的的节点增加数据删除的watcher,本质是启动另外一个线程去监听前置节点
        this.zkClient.subscribeDataChanges(beforePath, listener);

        // 如果前一个节点存在，监听前一个节点的原因主要是避免羊群效应
        if(this.zkClient.exists(beforePath)){
            // 阻塞等待
            countDownLatch=new CountDownLatch(1);
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 取消监听
        this.zkClient.unsubscribeDataChanges(beforePath, listener);
    }


    // 解锁
    public void unLock() {
        //删除当前临时节点
        zkClient.delete(currentPath);
        zkClient.close();
    }
}
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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101