[大数据]springboot整合zookeeper

前提：
springboot版本： 2.5.3
zookeeper版本：3.4.9 （centos7 安装集群）

gitee项目地址：https://gitee.com/qinenqi/zookeeper/tree/master/zookeeper01
postman测试数据：json.json

{
  "item":[

    {
      "name":"study",
      "item":[
        {
          "name":"zookeeper",
          "item":[
            {
              "name":"01 创建zookeeper节点",
              "request":{
                "method":"POST",
                "header":[

                ],
                "body":{
                  "mode":"raw",
                  "raw":"{\r\n    \"path\":\"/qinenqi\",\r\n    \"obj\":\"开始测试数据\",\r\n    \"aCLType\":\"OPEN_ACL_UNSAFE\",\r\n    \"createNodeType\":0\r\n}",
                  "options":{
                    "raw":{
                      "language":"json"
                    }
                  }
                },
                "url":{
                  "raw":"127.0.0.1:8888/zookeeperController/create",
                  "host":[
                    "127",
                    "0",
                    "0",
                    "1"
                  ],
                  "port":"8888",
                  "path":[
                    "zookeeperController",
                    "create"
                  ],
                  "query":[
                    {
                      "key":"path",
                      "value":"/qinenqi",
                      "disabled":true
                    },
                    {
                      "key":"obj",
                      "value":"",
                      "disabled":true
                    }
                  ]
                },
                "description":"01 创建zookeeper节点"
              },
              "response":[

              ]
            },
            {
              "name":"02更新节点",
              "request":{
                "method":"POST",
                "header":[

                ],
                "body":{
                  "mode":"raw",
                  "raw":"{\r\n    \"path\":\"/qinenqi\",\r\n    \"obj\":\"开始更新数据\"\r\n}",
                  "options":{
                    "raw":{
                      "language":"json"
                    }
                  }
                },
                "url":{
                  "raw":"127.0.0.1:8888/zookeeperController/updateZnode",
                  "host":[
                    "127",
                    "0",
                    "0",
                    "1"
                  ],
                  "port":"8888",
                  "path":[
                    "zookeeperController",
                    "updateZnode"
                  ]
                },
                "description":"02更新节点"
              },
              "response":[

              ]
            },
            {
              "name":"03删除节点",
              "request":{
                "method":"POST",
                "header":[

                ],
                "body":{
                  "mode":"raw",
                  "raw":"{\r\n    \"path\":\"/qinenqi\"\r\n}",
                  "options":{
                    "raw":{
                      "language":"json"
                    }
                  }
                },
                "url":{
                  "raw":"127.0.0.1:8888/zookeeperController/deleteZnode",
                  "host":[
                    "127",
                    "0",
                    "0",
                    "1"
                  ],
                  "port":"8888",
                  "path":[
                    "zookeeperController",
                    "deleteZnode"
                  ]
                },
                "description":"03删除节点"
              },
              "response":[

              ]
            },
            {
              "name":"04获取根节点下所有子节点",
              "request":{
                "method":"POST",
                "header":[

                ],
                "url":{
                  "raw":"127.0.0.1:8888/zookeeperController/getChildren",
                  "host":[
                    "127",
                    "0",
                    "0",
                    "1"
                  ],
                  "port":"8888",
                  "path":[
                    "zookeeperController",
                    "getChildren"
                  ]
                },
                "description":"04获取根节点下所有子节点"
              },
              "response":[

              ]
            }
          ]
        }
      ]
    }
  ]
}

1. 引入依赖：

 <!-- zookeeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.10</version>
        </dependency>

2. 新建 ZookeeperConfig

import lombok.Data;
import org.apache.zookeeper.*;

@Data
public class ZookeeperConfig {

    //连接的zk服务
    private static final String CONNECTION_HOSTS = "192.168.88.100:2181,192.88.110.128:2181,192.168.88.120:2181";
    //超时连接的时间
    private static final int SESSION_TIMEOUT = 2000;

    private  static ZooKeeper zooKeeper = null;
    static {
        try {
            zooKeeper = new ZooKeeper(CONNECTION_HOSTS, SESSION_TIMEOUT, new Watcher() {
                public void process(WatchedEvent watchedEvent) {
                    System.out.println(watchedEvent.getPath()+";"+watchedEvent.getState());
                    try {
                        //获取对根目录节点数据的监控
                        zooKeeper.getChildren("/",true);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static ZooKeeper getZooKeeper() {
        return zooKeeper;
    }
}

新建 ZookeeperUtil

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import java.util.ArrayList;


/**
 *  Zookeeper 工具类
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ZookeeperUtil {

    private ZooKeeper zooKeeper;

    /*
    判断当前节点是否存在
    */
    public Stat isExistZnode(String path) throws Exception{
        Stat stat = zooKeeper.exists(path,true);
        return stat;
    }
    /*
      创建节点
     */
    public String createZnode(String path, Object object, ArrayList<ACL> zooDefs, CreateMode createMode) throws Exception {
        return  zooKeeper.create(path,object.toString().getBytes(),zooDefs,createMode);
    }
    /*
      修改节点
     */
    public Stat updateZnode(String path, Object object, int version) throws Exception {
        return zooKeeper.setData(path,object.toString().getBytes(),version);
    }
    /*
      删除节点
     */
    public void deleteZnode(String path,int version) throws Exception {
        zooKeeper.delete(path,version);
    }


}

新建实体类：

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import java.io.Serializable;
import java.util.ArrayList;

@Data
@AllArgsConstructor
@NoArgsConstructor
@Slf4j
public class Zookeeper implements Serializable {

    private String id;

    /**
     *  zookeeper的路径
     */
    private String path;

    /**
     *  节点对应的数据
     */
    private Object obj;

    /**
     * OPEN_ACL_UNSAFE  : 完全开放的ACL，任何连接的客户端都可以操作该属性znode
     * CREATOR_ALL_ACL : 只有创建者才有ACL权限
     * READ_ACL_UNSAFE：只能读取ACL
     */
    private String aCLType;

    /**
     *  acl的集合：ACL权限
     */
    private ArrayList<ACL> lists;

    /**
     *  0 PERSISTENT 持久化目录节点, 会话结束存储数据不会丢失
     *  2 PERSISTENT_SEQUENTIAL 顺序自动编号持久化目录节点
     *  1 EPHEMERAL 临时目录节点
     *  3 EPHEMERAL_SEQUENTIAL 临时自动编号节点
     */
    private int createNodeType;

    /**
     *  节点类型：
     *  PERSISTENT 持久化目录节点, 会话结束存储数据不会丢失
     *  PERSISTENT_SEQUENTIAL 顺序自动编号持久化目录节点
     *  EPHEMERAL 临时目录节点
     *  EPHEMERAL_SEQUENTIAL 临时自动编号节点
     */
    private CreateMode createMode;

    public void setaCLType(String aCLType) {
        this.aCLType = aCLType;
        // ArrayList<ACL> 赋值
        if("OPEN_ACL_UNSAFE".equals(aCLType)){
            lists = ZooDefs.Ids.OPEN_ACL_UNSAFE;
        }else if("CREATOR_ALL_ACL".equals(aCLType)){
            lists = ZooDefs.Ids.CREATOR_ALL_ACL;
        }else if("READ_ACL_UNSAFE".equals(aCLType)){
            lists = ZooDefs.Ids.READ_ACL_UNSAFE;
        }
    }


    public void setCreateNodeType(int createNodeType) {
        this.createNodeType = createNodeType;
        if(createNodeType == 0){
            createMode = CreateMode.PERSISTENT;
        }else if(createNodeType == 2){
            createMode = CreateMode.PERSISTENT_SEQUENTIAL;
        }else if(createNodeType == 1){
            createMode = CreateMode.EPHEMERAL;
        }else if(createNodeType == 3){
            createMode = CreateMode.EPHEMERAL_SEQUENTIAL;
        }
    }
}

新建controller ZookeeperController

import com.example.zookeeper01.entry.Zookeeper;
import com.example.zookeeper01.service.ZookeeperService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/zookeeperController")
public class ZookeeperController {



    @Autowired
    private ZookeeperService zookeeperService;

    /**
     *  创建zookeeper节点
     * @param zookeeper
     */
    @RequestMapping("/create")
    public void create(@RequestBody Zookeeper zookeeper){
        zookeeperService.create(zookeeper);
    }

    /**
     * 更新节点
     * @param zookeeper
     */
    @RequestMapping("/updateZnode")
    public void updateZnode(@RequestBody Zookeeper zookeeper){
        zookeeperService.updateZnode(zookeeper);
    }

    /**
     * 删除节点
     * @param zookeeper
     */
    @RequestMapping("/deleteZnode")
    public void deleteZnode(@RequestBody Zookeeper zookeeper){
        zookeeperService.deleteZnode(zookeeper);
    }

    /**
     * 获取根节点下所有子节点
     */
    @RequestMapping("/getChildren")
    public void getChildren(){
        zookeeperService.getChildren();
    }
}

新建 ZookeeperService

import com.example.zookeeper01.entry.Zookeeper;

public interface ZookeeperService {

    /**
     *  创建zookeeper节点
     * @param zookeeper
     */
    public void create(Zookeeper zookeeper);

    /**
     * 更新节点
     * @param zookeeper
     */
    public void updateZnode(Zookeeper zookeeper);

    /**
     * 删除节点
     * @param zookeeper
     */
    public void deleteZnode(Zookeeper zookeeper);

    /**
     * 获取根节点下所有子节点
     *
     */
    public void getChildren();
}

新建 ZookeeperServiceImpl

import com.example.zookeeper01.config.ZookeeperConfig;
import com.example.zookeeper01.entry.Zookeeper;
import com.example.zookeeper01.service.ZookeeperService;
import com.example.zookeeper01.utils.ZookeeperUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;

@Service
@Slf4j
public class ZookeeperServiceImpl implements ZookeeperService {



    /**
     *  创建zookeeper节点
     * @param zookeeper
     */
    public void create(Zookeeper zookeeper){
        try{
            String path = zookeeper.getPath();
            Object obj = zookeeper.getObj();
            ArrayList<ACL> lists = zookeeper.getLists();
            CreateMode createMode = zookeeper.getCreateMode();
            ZooKeeper zooKeeper = ZookeeperConfig.getZooKeeper();
            ZookeeperUtil zookeeperUtil = new ZookeeperUtil(zooKeeper);
            if(zookeeperUtil.isExistZnode(path) == null){
                zookeeperUtil.createZnode(path, obj, lists, createMode);
                log.info("创建zookeeper节点 成功");
            }else {
                log.info(path + " znode exist！！！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 更新节点
     * @param zookeeper
     */
    public void updateZnode(Zookeeper zookeeper){
        try{
            String path = zookeeper.getPath();
            Object obj = zookeeper.getObj();
            ArrayList<ACL> lists = zookeeper.getLists();
            CreateMode createMode = zookeeper.getCreateMode();
            ZooKeeper zooKeeper = ZookeeperConfig.getZooKeeper();
            ZookeeperUtil zookeeperUtil = new ZookeeperUtil(zooKeeper);
            Stat existZnode = zookeeperUtil.isExistZnode(path);
            if(existZnode != null){
                zookeeperUtil.updateZnode(path, obj, existZnode.getVersion());
                log.info("更新 zookeeper节点 成功");
            }else {
                log.info(path + " 不存在！！！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    /**
     * 删除节点
     * @param zookeeper
     */
    public void deleteZnode(Zookeeper zookeeper){
        try{
            String path = zookeeper.getPath();
            Object obj = zookeeper.getObj();
            ArrayList<ACL> lists = zookeeper.getLists();
            CreateMode createMode = zookeeper.getCreateMode();
            ZooKeeper zooKeeper = ZookeeperConfig.getZooKeeper();
            ZookeeperUtil zookeeperUtil = new ZookeeperUtil(zooKeeper);
            Stat existZnode = zookeeperUtil.isExistZnode(path);
            if(existZnode != null){
                zookeeperUtil.deleteZnode(path, existZnode.getVersion());
                log.info("删除 zookeeper节点 成功");
            }else {
                log.info(path + " 不存在！！！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 获取根节点下所有子节点
     *
     */
    public void getChildren(){
        try{
            ZooKeeper zooKeeper = ZookeeperConfig.getZooKeeper();
            //获取根节点下所有子节点,遍历获取对应节点的数据
            List<String> znodes = zooKeeper.getChildren("/",true);
            for(String str : znodes){
                log.info(str+">>>"+new String(zooKeeper.getData("/"+str,true,new Stat())));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}