Java Spring Boot 集成Zookeeper

目录

• 集成步骤
• 1.pom.xml文件配置，引入相关jar包
• 2. 核心配置类
• 3.常用API功能
• 4.基本使用
• 常见错误和解决办法
• 问题1：调用api创建zookeeper节点时，报KeeperErrorCode = Unimplemented for /test错误。
• 问题2：启动项目的日志中会有Will not attempt to authenticate using SASL错误

集成步骤

1.pom.xml文件配置，引入相关jar包

Curator是Netflix公司开源的一套zookeeper客户端框架，解决了很多Zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等等。

 <!-- 封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式Barrier -->
 <dependency>
          <groupId>org.apache.curator</groupId>
          <artifactId>curator-recipes</artifactId>
          <version>2.10.0</version>
          <exclusions>
            <exclusion>
              <groupId>org.apache.zookeeper</groupId>
              <artifactId>zookeeper</artifactId>
             </exclusion>
          </exclusions>
     </dependency>

      <dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.4.13</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-log4j12</artifactId>
                    </exclusion>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-api</artifactId>
                    </exclusion>
                    <exclusion>
                        <groupId>log4j</groupId>
                        <artifactId>log4j</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>

     <dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-web</artifactId>
     </dependency>

特殊说明： 1.无需引入curator-framework，因为curator-recipes自动关联依赖引入curator-framework。 2.curator会默认引入zookeeper的jar报，需要检查版本与服务器的版本是否一致，如果不一致则需要排除引入 3.

2. 核心配置类

@Configuration
public class ZookeeperConfig implements Serializable
{
    private static final long serialVersionUID = -9025878246972668136L;

    private final ZooKeeperProperty zooKeeperProperty;

    public ZookeeperConfig(ZooKeeperProperty zooKeeperProperty) {
        this.zooKeeperProperty = zooKeeperProperty;
    }

    @Bean
    public CuratorFramework curatorFramework()
    {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(zooKeeperProperty.getBaseSleepTime(),
                zooKeeperProperty.getMaxRetries());
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(zooKeeperProperty.getServers())
                .connectionTimeoutMs(zooKeeperProperty.getConnectionTimeout())
                .sessionTimeoutMs(zooKeeperProperty.getSessionTimeout())
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        return client;
    }

    @Bean
    @ConditionalOnMissingBean
    public ZooKeeperUtils zooKeeperTemplate(CuratorFramework client) {
        return new ZooKeeperUtils(client);
    }
}

@ConfigurationProperties(prefix="zookeeper")
@Component
public class ZooKeeperProperty implements Serializable
{
    private static final long serialVersionUID = 8650758711482699256L;

    /**
     * zk连接集群，多个用逗号隔开
     */
    private String servers;

    /**
     * 会话超时时间
     */
    private int sessionTimeout = 60000;

    /**
     * 连接超时时间
     */
    private int connectionTimeout = 15000;

    /**
     * 初始重试等待时间(毫秒)
     */
    private int baseSleepTime = 1000;

    /**
     * 重试最大次数
     */
    private int maxRetries = 10;

    //省略get、set方法
    ......
    }

3.常用API功能

@Component
public class ZooKeeperUtils
{
    private static final Logger logger = LoggerFactory
            .getLogger(ZooKeeperUtils.class);

    /**
     * 路径分隔符
     */
    private static final String PATH_SEPARATOR = "/";

    /**
     * zk连接
     */
    private final CuratorFramework client;

    public ZooKeeperUtils(CuratorFramework client)
    {
        this.client = client;
    }

    /**
     * 创建空节点，默认持久节点
     *
     * @param path
     *            节点路径
     * @param node
     *            节点名称
     * @return 完整路径
     */
    public String createNode(String path, String node)
    {
        return createNode(path, node, CreateMode.PERSISTENT);
    }

    /**
     * 创建带类型的空节点
     *
     * @param path
     *            节点路径
     * @param node
     *            节点名称
     * @param createMode
     *            类型 CreateMode.PERSISTENT: 创建节点后，不删除就永久存在
     *            CreateMode.PERSISTENT_SEQUENTIAL：节点path末尾会追加一个10位数的单调递增的序列
     *            CreateMode.EPHEMERAL：创建后，回话结束节点会自动删除
     *            CreateMode.EPHEMERAL_SEQUENTIAL：节点path末尾会追加一个10位数的单调递增的序列
     * @return 路径
     */
    public String createNode(String path, String node, CreateMode createMode)
    {
        path = buildPath(path, node);
        logger.info("create node for path: {} with createMode: {}", path,
                createMode.name());
        try
        {

            client.create().creatingParentsIfNeeded().withMode(createMode)
                    .forPath(path);

            logger.info("create node :{} sucessfully", node);
            return path;
        }
        catch (Exception e)
        {
            logger.error(
                    "create node for path: {} with createMode: {} failed!",
                    path, createMode.name(), e);
            return null;
        }
    }

    /**
     * 创建节点，默认持久节点
     *
     * @param path
     *            节点路径
     * @param node
     *            节点名称
     * @param value
     *            节点值
     * @return 完整路径
     */
    public String createNode(String path, String node, String value)
    {
        return createNode(path, node, value, CreateMode.PERSISTENT);
    }

    /**
     * 创建节点，默认持久节点
     *
     * @param path
     *            节点路径
     * @param node
     *            节点名称
     * @param value
     *            节点值
     * @param createMode
     *            节点类型
     * @return 完整路径
     */
    public String createNode(String path, String node, String value,
            CreateMode createMode)
    {
        if (Objects.isNull(value))
        {
            logger.error("ZooKeeper节点值不能为空!");
        }
        path = buildPath(path, node);
        logger.info("create node for path: {}, value: {}, with createMode: {}",
                path, value, createMode.name());
        try
        {
            client.create().creatingParentsIfNeeded().withMode(createMode)
                    .forPath(path, value.getBytes());
            return path;
        }
        catch (Exception e)
        {
            logger.error(
                    "create node for path: {}, value: {}, with createMode: {} failed!",
                    path, value, createMode.name(), e);
        }
        return null;
    }

    /**
     * 获取节点数据
     *
     * @param path
     *            路径
     * @param node
     *            节点名称
     * @return 完整路径
     */
    public String get(String path, String node)
    {
        path = buildPath(path, node);
        try
        {
            byte[] bytes = client.getData().forPath(path);
            if (bytes.length > 0)
            {
                return new String(bytes);
            }
        }
        catch (Exception e)
        {
            logger.error("get value for path: {}, node: {} failed!", path,
                    node, e);
        }
        return null;
    }

    /**
     * 更新节点数据
     *
     * @param path
     *            节点路径
     * @param node
     *            节点名称
     * @param value
     *            更新值
     * @return 完整路径
     */
    public String update(String path, String node, String value)
    {
        if (Objects.isNull(value))
        {
            logger.error("ZooKeeper节点值不能为空!");
        }
        path = buildPath(path, node);
        logger.info("update path: {} to value: {}", path, value);

        try
        {
            client.setData().forPath(path, value.getBytes());
            return path;
        }
        catch (Exception e)
        {
            logger.error("update path: {} to value: {} failed!", path, value);
        }
        return null;
    }

    /**
     * 删除节点，并且递归删除子节点
     *
     * @param path
     *            路径
     * @param node
     *            节点名称
     * @return 路径
     */
    public boolean delete(String path, String node)
    {
        path = buildPath(path, node);
        logger.info("delete node for path: {}", path);

        try
        {
            client.delete().deletingChildrenIfNeeded().forPath(path);
            return true;
        }
        catch (Exception e)
        {
            logger.error("delete node for path: {} failed!", path);
        }
        return false;
    }

    /**
     * 获取子节点
     *
     * @param path
     *            节点路径
     * @return
     */
    public List<String> getChildren(String path)
    {
        if (StringUtils.isEmpty(path))
        {
            return null;
        }

        if (!path.startsWith(PATH_SEPARATOR))
        {
            path = PATH_SEPARATOR + path;
        }

        try
        {
            return client.getChildren().forPath(path);
        }
        catch (Exception e)
        {
            logger.error("get children path:{} error", path, e);
        }
        return null;
    }

    /**
     * 判断节点是否存在
     *
     * @param path
     *            路径
     * @param node
     *            节点名称
     * @return 结果
     */
    public boolean exists(String path, String node)
    {
        try
        {
            List<String> list = getChildren(path);
            return !CollectionUtils.isEmpty(list) && list.contains(node);
        }
        catch (Exception e)
        {
            return false;
        }
    }

    /**
     * 申请锁，指定请求等待时间
     *
     * @param path
     *            加锁zk节点
     * @param time
     *            时间
     * @param unit
     *            时间单位
     * @param runnable
     *            执行方法
     */
    public void lock(String path, long time, TimeUnit unit, Runnable runnable)
    {
        try
        {
            InterProcessMutex lock = new InterProcessMutex(client, path);
            if (lock.acquire(time, unit))
            {
                try
                {
                    runnable.run();
                }
                finally
                {
                    lock.release();
                }
            }
            else
            {
                logger.error("获取锁超时：{}!", path);
            }
        }
        catch (Exception e)
        {
            logger.error("获取锁失败: {}!", path);
        }
    }

    /**
     * 申请锁，指定请求等待时间
     *
     * @param path
     *            加锁zk节点
     * @param time
     *            时间
     * @param unit
     *            时间单位
     * @param callable
     *            执行方法
     * @return .
     */
    public <T> T lock(String path, long time, TimeUnit unit,
            Callable<T> callable)
    {
        try
        {
            InterProcessMutex lock = new InterProcessMutex(client, path);
            if (lock.acquire(time, unit))
            {
                try
                {
                    return callable.call();
                }
                finally
                {
                    lock.release();
                }
            }
            else
            {
                logger.error("获取锁超时：{}!", path);
            }
        }
        catch (Exception e)
        {
            logger.error("获取锁失败: {}!", path);
        }
        return null;
    }

    /* *//**
     * 对一个节点进行监听，监听事件包括指定的路径节点的增、删、改的操作
     *
     * @param path
     *            节点路径
     * @param listener
     *            回调方法
     * @throws Exception
     */

    public void watchNode(String path,boolean dataIsCompressed,final ZooKeeperCallback zooKeeperCallback)throws Exception
    {
        try
        {
            final NodeCache nodeCache = new NodeCache(client, path,dataIsCompressed);
            nodeCache.getListenable().addListener(new NodeCacheListener()
            {
                public void nodeChanged() throws Exception
                {
                    ChildData childData = nodeCache.getCurrentData();
                    logger.info("ZNode节点状态改变, path={}", childData.getPath());
                    logger.info("ZNode节点状态改变, data={}", childData.getData());
                    logger.info("ZNode节点状态改变, stat={}", childData.getStat());

                    //处理业务逻辑
                    zooKeeperCallback.call();
                }
            });      

            nodeCache.start();
        }
        catch (Exception e)
        {
            logger.error("创建NodeCache监听失败, path={}",path);
        }
    }

    /**
     * 对指定的路径节点的一级子目录进行监听，不对该节点的操作进行监听，对其子目录的节点进行增、删、改的操作监听
     *
     * @param path
     *            节点路径
     * @param listener
     *            回调方法
     */
    public void watchChildren(String path, PathChildrenCacheListener listener)
    {
        try
        {
            PathChildrenCache pathChildrenCache = new PathChildrenCache(client,
                    path, true);
            pathChildrenCache.start(PathChildrenCache.StartMode.NORMAL);
            pathChildrenCache.getListenable().addListener(listener);
        }
        catch (Exception e)
        {
            logger.error("watch children failed for path: {}", path, e);
        }
    }

    /**
     * 将指定的路径节点作为根节点（祖先节点），对其所有的子节点操作进行监听，呈现树形目录的监听，可以设置监听深度，最大监听深度为2147483647（
     * int类型的最大值）
     *
     * @param path
     *            节点路径
     * @param maxDepth
     *            回调方法
     * @param listener
     *            监听
     */
    public void watchTree(String path, int maxDepth, TreeCacheListener listener)
    {
        try
        {
            TreeCache treeCache = TreeCache.newBuilder(client, path)
                    .setMaxDepth(maxDepth).build();
            treeCache.start();
            treeCache.getListenable().addListener(listener);
        }
        catch (Exception e)
        {
            logger.error("watch tree failed for path: {}", path, e);
        }
    }

    public String buildPath(String path, String node)
    {
        if (StringUtils.isEmpty(path) || StringUtils.isEmpty(node))
        {
            logger.error("ZooKeeper路径或者节点名称不能为空！");
        }

        if (!path.startsWith(PATH_SEPARATOR))
        {
            path = PATH_SEPARATOR + path;
        }

        if (PATH_SEPARATOR.equals(path))
        {
            return path + node;
        }
        else
        {
            return path + PATH_SEPARATOR + node;
        }
    }
}

4.基本使用

@Autowired
    private ZooKeeperUtils zooKeeperUtil;

    @RequestMapping("/addNode")
    public String addNode()
    {
       String path= zooKeeperUtil.createNode("/zookeeper", "node1");
       return path;
    }

特殊说明：关于zookeeper的分布式锁，后续讲解常用分布式锁的时候，会详细说明。

常见错误和解决办法

问题1：调用api创建zookeeper节点时，报KeeperErrorCode = Unimplemented for /test错误。

原因：服务器安装zookeeper的版本与程序中的zookeeper版本不一致。

解决方案： 登录服务器，查看zookeeper安装版本，执行如下命令：

 echo stat|nc 127.0.0.1 2181

当前引入的zookeeper版本为3.4.13，而zookeeper的版本与curator对应关系如下：

  Curator 2.x.x - compatible with both ZooKeeper 3.4.x and ZooKeeper 3.5.x
  Curator 4.x.x - compatible only with ZooKeeper 3.5.x and includes support for new features such as dynamic reconfiguration, etc.
  Curator 5.x.x compatible only with ZooKeeper 3.6.x+

问题2：启动项目的日志中会有Will not attempt to authenticate using SASL错误

起初我认为是zookeeper需要进行SASL认证，但是通过查阅相关资料后，才知道3.4之前版本，zookeeper默认会采用SASL认证,3.4以后的版本没有此类问题。

到此这篇关于Java Spring Boot 集成Zookeeper的文章就介绍到这了,更多相关Spring Boot 集成Zookeeper内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！