前言：Zookeeper官方也提供了客户端，但是我们在日常工作中不怎么用，目前用的多的是Zookeeper客户端框架是Curator，本节我们主要讲解如何集成Curator，Zookeeper集群搭建方法也会总结。 ZooKeeper有很多可视化数据查看工具，比如基于CS架构的 ZooInspector，还有基于BS结构的如：ZKUI，感兴趣的可自行学习，这里不做介绍。

1、Apache Curator开源客户端

基本介绍

Curator 是一套由netflix 公司开源的，Java 语言编程的 ZooKeeper 客户端框架，Curator项目 是现在ZooKeeper 客户端中使用最多，对ZooKeeper 版本支持最好的第三方客户端，并推荐使用。
Curator 把我们平时常用的很多 ZooKeeper 服务开发功能做了封装，例如 Leader 选举、 分布式计数器、分布式锁。减少了技术人员在使用 ZooKeeper 时的大部分底层细节开发工 作。

如何集成

引入jar包:

  1. <dependency>
  2.             <groupId>org.apache.curator</groupId>
  3.             <artifactId>curator-framework</artifactId>
  4.             <version>2.12.0</version>
  5.         </dependency>
  6.         <dependency>
  7.             <groupId>org.apache.curator</groupId>
  8.             <artifactId>curator-recipes</artifactId>
  9.             <version>2.12.0</version>
  10.         </dependency>
  11.         <dependency>
  12.             <groupId>org.apache.curator</groupId>
  13.             <artifactId>curator-client</artifactId>
  14.             <version>2.12.0</version>
  15.         </dependency>

参数准备:

  1. zookeeper:
  2.   enabled: true
  3.   server: 127.0.0.1:2181
  4.   namespace: lwq
  5.   digest: lwq:123456
  6.   sessionTimeoutMs: 3000
  7.   connectionTimeoutMs: 60000
  8.   maxRetries: 2
  9.   baseSleepTimeMs: 1000
  1. @Data
  2. @Component
  3. @ConfigurationProperties(prefix = "zookeeper")
  4. public class ZookeeperParam {
  5.     private boolean enabled;
  6.     private String server;
  7.     private String namespace;
  8.     private String digest;
  9.     private int sessionTimeoutMs;
  10.     private int connectionTimeoutMs;
  11.     private int maxRetries;
  12.     private int baseSleepTimeMs;
  13. }

配置类：

  1. @Configuration
  2. public class ZookeeperConfig {
  3.     private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperConfig.class);
  5.     @Resource
  6.     private ZookeeperParam zookeeperParam;
  7.     private static CuratorFramework client = null;
  9.     /**
  10.      * 初始化
  11.      */
  12.     @PostConstruct
  13.     public void init() {
  14.         // 重试策略，初试时间1秒，重试10次
  15.         RetryPolicy policy =
  16.             new ExponentialBackoffRetry(zookeeperParam.getBaseSleepTimeMs(), zookeeperParam.getMaxRetries());
  17.         // 通过工厂创建Curator
  18.         client = CuratorFrameworkFactory.builder().connectString(zookeeperParam.getServer())
  19.             .authorization("digest", zookeeperParam.getDigest().getBytes())
  20.             .connectionTimeoutMs(zookeeperParam.getConnectionTimeoutMs())
  21.             .sessionTimeoutMs(zookeeperParam.getSessionTimeoutMs()).retryPolicy(policy).build();
  22.         // 开启连接
  23.         client.start();
  24.         LOGGER.info("zookeeper 初始化完成...");
  25.     }
  27.     public static CuratorFramework getClient() {
  28.         return client;
  29.     }
  31.     public static void closeClient() {
  32.         if (client != null) {
  33.             client.close();
  34.         }
  35.     }
  36. }

Znode节点操作：

  1. public interface ZookeeperService {
  2.     /**
  3.      * 判断节点是否存在
  4.      */
  5.     boolean isExistNode(final String path);
  7.     /**
  8.      * 创建节点
  9.      */
  10.     void createNode(CreateMode mode, String path);
  12.     /**
  13.      * 设置节点数据
  14.      */
  15.     void setNodeData(String path, String nodeData);
  17.     /**
  18.      * 创建节点
  19.      */
  20.     void createNodeAndData(CreateMode mode, String path, String nodeData);
  22.     /**
  23.      * 获取节点数据
  24.      */
  25.     String getNodeData(String path);
  27.     /**
  28.      * 获取节点下数据
  29.      */
  30.     List<String> getNodeChild(String path);
  32.     /**
  33.      * 是否递归删除节点
  34.      */
  35.     void deleteNode(String path, Boolean recursive);
  37. }
  1. @Service
  2. public class ZookeeperServiceImpl implements ZookeeperService {
  3.     private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperServiceImpl.class);
  5.     @Override
  6.     public boolean isExistNode(String path) {
  7.         CuratorFramework client = ZookeeperConfig.getClient();
  8.         client.sync();
  9.         try {
  10.             Stat stat = client.checkExists().forPath(path);
  11.             return client.checkExists().forPath(path) != null;
  12.         } catch (Exception e) {
  13.             LOGGER.error("isExistNode error...", e);
  14.             e.printStackTrace();
  15.         }
  16.         return false;
  17.     }
  19.     @Override
  20.     public void createNode(CreateMode mode, String path) {
  21.         CuratorFramework client = ZookeeperConfig.getClient();
  22.         try {
  23.             // 递归创建所需父节点
  24.             client.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
  25.         } catch (Exception e) {
  26.             LOGGER.error("createNode error...", e);
  27.             e.printStackTrace();
  28.         }
  29.     }
  31.     @Override
  32.     public void setNodeData(String path, String nodeData) {
  33.         CuratorFramework client = ZookeeperConfig.getClient();
  34.         try {
  35.             // 设置节点数据
  36.             client.setData().forPath(path, nodeData.getBytes("UTF-8"));
  37.         } catch (Exception e) {
  38.             LOGGER.error("setNodeData error...", e);
  39.             e.printStackTrace();
  40.         }
  41.     }
  43.     @Override
  44.     public void createNodeAndData(CreateMode mode, String path, String nodeData) {
  45.         CuratorFramework client = ZookeeperConfig.getClient();
  46.         try {
  47.             // 创建节点，关联数据
  48.             client.create().creatingParentsIfNeeded().withMode(mode).forPath(path, nodeData.getBytes("UTF-8"));
  49.         } catch (Exception e) {
  50.             LOGGER.error("createNode error...", e);
  51.             e.printStackTrace();
  52.         }
  53.     }
  55.     @Override
  56.     public String getNodeData(String path) {
  57.         CuratorFramework client = ZookeeperConfig.getClient();
  58.         try {
  59.             // 数据读取和转换
  60.             byte[] dataByte = client.getData().forPath(path);
  61.             String data = new String(dataByte, "UTF-8");
  62.             if (StringUtils.isNotEmpty(data)) {
  63.                 return data;
  64.             }
  65.         } catch (Exception e) {
  66.             LOGGER.error("getNodeData error...", e);
  67.             e.printStackTrace();
  68.         }
  69.         return null;
  70.     }
  72.     @Override
  73.     public List<String> getNodeChild(String path) {
  74.         CuratorFramework client = ZookeeperConfig.getClient();
  75.         List<String> nodeChildDataList = new ArrayList<>();
  76.         try {
  77.             // 节点下数据集
  78.             nodeChildDataList = client.getChildren().forPath(path);
  79.         } catch (Exception e) {
  80.             LOGGER.error("getNodeChild error...", e);
  81.             e.printStackTrace();
  82.         }
  83.         return nodeChildDataList;
  84.     }
  86.     @Override
  87.     public void deleteNode(String path, Boolean recursive) {
  88.         CuratorFramework client = ZookeeperConfig.getClient();
  89.         try {
  90.             if (recursive) {
  91.                 // 递归删除节点
  92.                 client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
  93.             } else {
  94.                 // 删除单个节点
  95.                 client.delete().guaranteed().forPath(path);
  96.             }
  97.         } catch (Exception e) {
  98.             LOGGER.error("deleteNode error...", e);
  99.             e.printStackTrace();
  100.         }
  101.     }
  102. }

Controller调用：

  1. @RestController
  2. public class ZookeeperApi {
  3.     @Resource
  4.     private ZookeeperService zookeeperService;
  6.     @GetMapping("/getNodeData")
  7.     public String getNodeData(String path) {
  8.         return zookeeperService.getNodeData(path);
  9.     }
  11.     @GetMapping("/isExistNode")
  12.     public boolean isExistNode(final String path) {
  13.         return zookeeperService.isExistNode(path);
  14.     }
  16.     @GetMapping("/createNode")
  17.     public String createNode(CreateMode mode, String path) {
  18.         zookeeperService.createNode(mode, path);
  19.         return "success";
  20.     }
  22.     @GetMapping("/setNodeData")
  23.     public String setNodeData(String path, String nodeData) {
  24.         zookeeperService.setNodeData(path, nodeData);
  25.         return "success";
  26.     }
  28.     @GetMapping("/createNodeAndData")
  29.     public String createNodeAndData(CreateMode mode, String path, String nodeData) {
  30.         zookeeperService.createNodeAndData(mode, path, nodeData);
  31.         return "success";
  32.     }
  34.     @GetMapping("/getNodeChild")
  35.     public List<String> getNodeChild(String path) {
  36.         return zookeeperService.getNodeChild(path);
  37.     }
  39.     @GetMapping("/deleteNode")
  40.     public String deleteNode(String path, Boolean recursive) {
  41.         zookeeperService.deleteNode(path, recursive);
  42.         return "success";
  43.     }
  44. }

集成Curator，完成!

2、Zookeeper集群搭建

集群搭建

Zookeeper 集群模式一共有三种类型的角色

  • Leader: 处理所有的事务请求（写请求），可以处理读请求，集群中只能有一个Leader
  • Follower：只能处理读请求，同时作为 Leader的候选节点，即如果Leader宕机，Follower节点 要参与到新的Leader选举中，有可能成为新的Leader节点。
  • Observer：只能处理读请求。不能参与选举

搭建集群示例：
Zookeeper 集群配置.docx