Spring Boot 集成 ZooKeeper，及其简单应用和分布式锁_springboot集成zookeeper

一、Spring Boot 集成 ZooKeeper

配置文件pom.xml:

        <!-- ZooKeeper 之 Curator-->
        <!-- ZooKeeper版本号为4的话，机器安装zookeeper的版本要求是3.5及其以上的版本-->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipesss</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.1</version>
        </dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

配置文件application.properties：

##ZooKeeper 集成 Curator
zk.url = 127.0.0.1:2181

1
2
3

java class文件：

package com.example.zp.controller.zookeeper.configure;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;



/**
 * @Author: zp
 * @Date: 2019/5/7 10:57
 * @Description:
 */
@Configuration
@Slf4j
public class ZooKeeperConf {

    @Value("${zk.url}")
    private String zkUrl;

    @Bean
    public CuratorFramework getCuratorFramework() {
        // 用于重连策略，1000毫秒是初始化的间隔时间，3代表尝试重连次数
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(zkUrl, retryPolicy);
        //必须调用start开始连接ZooKeeper
        client.start();


//        /**
//         * 使用Curator，可以通过LeaderSelector来实现领导选取；
//         * 领导选取：选出一个领导节点来负责其他节点；如果领导节点不可用，则在剩下的机器里再选出一个领导节点
//         */
//        // 构造一个监听器
//        LeaderSelectorListenerAdapter listener = new LeaderSelectorListenerAdapter() {
//            @Override
//            public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
//                log.info("get leadership");
//                // 领导节点，方法结束后退出领导。zk会再次重新选择领导
//
//            }
//        };
//        LeaderSelector selector = new LeaderSelector(client, "/schedule", listener);
//        selector.autoRequeue();
//        selector.start();



        return client;
    }



}


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

启动项目后可以看见如图说明集成成功：

====================================================================================================================================================

二、简单应用和分布式锁

java class文件（controller层）：

package com.example.zp.controller.zookeeper;

import com.example.shopgoods.controller.zookeeper.zookeeperImpl.ZooKeeperImpl;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

/**
 * @Author: zp
 * @Date: 2019/5/7 11:43
 * @Description:
 */
@RestController
@RequestMapping("/zookeeper")
public class ZooKeeperController {

    @Autowired
    private CuratorFramework zkClient;

    @Autowired
    private ZooKeeperImpl zooKeeper;


    /**
     * zookeeper 获取节点下的数据
     *
     * post请求： http://localhost:8082/zookeeper/makeOrder?path=/task
     *
     * @param path
     * @return
     */
    @PostMapping("/getData")
    public String getData(@RequestParam String path) {
        byte[] bytes = null;
        try {
            bytes = zkClient.getData().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }

        String str = new String(bytes);
        return str;
    }


    @PostMapping("/create")
    public String create(@RequestParam String path) {
        try {
            zkClient.create().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "success";
    }


    @PostMapping("/delete")
    public String delete(@RequestParam String path) {
        try {
            zkClient.delete().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "success";
    }


    @PostMapping("/setData")
    public String setData(@RequestParam(value = "path") String path, @RequestParam(value = "data") String data) {
        try {
            zkClient.setData().forPath(path, data.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "success";
    }


    @PostMapping("/check")
    public String check(@RequestParam(value = "path") String path) {
        Stat stat = null;
        try {
            stat = zkClient.checkExists().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "stat" + stat;
    }


    @PostMapping("/children")
    public String children(@RequestParam(value = "path") String path) {
        List<String> children = null;
        try {
            children = zkClient.getChildren().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "children" + children;
    }

    @PostMapping("/watch")
    public String watch(@RequestParam(value = "path") String path) {
        Stat stat = null;
        try {
            stat = zkClient.checkExists().watched().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "watch " + stat;
    }


    /**
     * zookeeper分布式锁
     *
     * @param product
     * @return
     */
    @PostMapping("/makeOrder")
    public String makeOrder(@RequestParam(value = "product") String product) {
        zooKeeper.makeOrder(product);
        return "success";
    }
}

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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

java class文件（impl层）：

package com.example.zp.controller.zookeeper.zookeeperImpl;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

/**
 * @Author: zp
 * @Date: 2019/5/7 14:11
 * @Description:
 */
@Service
@Slf4j
public class ZooKeeperImpl {

    private static final String lockPath = "/lock/order";
    @Autowired
    private CuratorFramework zkClient;


    public void makeOrder(String product) {
        log.info("try do job for " + product);
        String path = lockPath + "/" + product;

        try {
        	// InterProcessMutex 构建一个分布式锁
            InterProcessMutex lock = new InterProcessMutex(zkClient, path);

            try {
                if (lock.acquire(5, TimeUnit.HOURS)) {
                	// 模拟业务处理耗时5秒
                    Thread.sleep(5*1000);
                    log.info("do job " + product + "done");
                }
            } finally {
            	// 释放该锁
                lock.release();
            }

        } catch (Exception e) {
        	// zk异常
            e.printStackTrace();
        }
    }


}


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

项目启动后，执行效果：
修改代码，假使业务处理耗时为50秒，用postman多次调用该方法，如下图可以看出来线程4和线程6都在等待线程3执行完毕；从zookeeper客户端可以看出来，book节点下面有3个节点等待执行；

预计3分钟过后：
如下图，方法依次执行，从zookeeper客户端可以看出来节点book下的子节点也会随着每个会话的结束而删除，最后节点book也会被删除！

idea 关于zookeeper插件：

参考文献：

springboot集成zookeeper报错： KeeperErrorCode = Unimplemented异常的解决方法

疑问：

1.zookeeper的基本命令有：
①create：创建节点；
②get：获取节点数据；
③delete：删除节点；
④set path data：设置节点数据；等命令。
不知道是否有修改节点名称的命令？