亲宝软件园·资讯

展开

Spring Boot 集成Zookeeper

剑圣无痕 人气:0

集成步骤

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以后的版本没有此类问题。

加载全部内容

相关教程
猜你喜欢
用户评论