devbox
代码大牛
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spring Boot 整合Zookeeper组件_springboot集成zookeeper

作者：代码大牛 | 2024-02-01 14:05:13

赞

踩

springboot集成zookeeper

简介

         Zookeeper是一个Apache开源的分布式的应用，为系统架构提供协调服务。从设计模式角度来审视：该组件是一个基于观察者模式设计的框架，负责存储和管理数据，接受观察者的注册，一旦数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的观察者做出相应的反应，从而实现集群中类似Master/Slave管理模式。ZooKeeper的目标就是封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户。

技术架构：

Spring Boot 2.6.3 、Zookeeper3.4.6 、 JDK1.8

导入依赖

  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>
  16.        <dependency>
  17.             <groupId>org.projectlombok</groupId>
  18.             <artifactId>lombok</artifactId>
  19.             <version>1.18.4</version>
  20.         </dependency>

application.yml配置

  1. zoo:
  2.   keeper:
  3.     #开启标志
  4.     enabled: true
  5.     #服务器地址
  6.     server: 192.168.116.100:2181
  7.     #命名空间，被称为ZNode
  8.     namespace: lx
  9.     #权限控制，加密
  10.     digest: smile:111111
  11.     #会话超时时间
  12.     sessionTimeoutMs: 3000
  13.     #连接超时时间
  14.     connectionTimeoutMs: 60000
  15.     #最大重试次数
  16.     maxRetries: 10
  17.     #初始休眠时间
  18.     baseSleepTimeMs: 1000

Zookeeper配置

  1. package com.example.zk.config;
  2.  
  3. import lombok.Data;
  4. import org.springframework.boot.context.properties.ConfigurationProperties;
  5. import org.springframework.stereotype.Component;
  6.  
  7. /**
  8.  * @author lanx
  9.  * @date 2022/3/6
  10.  */
  11. @Data
  12. @Component
  13. @ConfigurationProperties(prefix = "zoo.keeper")
  14. public class ZookeeperProperties {
  15.     private String enabled;
  16.     private String server;
  17.     private String namespace;
  18.     private String digest;
  19.     private int sessionTimeoutMs;
  20.     private int connectionTimeoutMs;
  21.     private int maxRetries;
  22.     private int baseSleepTimeMs;
  23.  
  24. }

Zookeeper 初始化

  1. package com.example.zk.config;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.curator.RetryPolicy;
  5. import org.apache.curator.framework.CuratorFramework;
  6. import org.apache.curator.framework.CuratorFrameworkFactory;
  7. import org.apache.curator.retry.ExponentialBackoffRetry;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.context.annotation.Configuration;
  10.  
  11. import javax.annotation.PostConstruct;
  12.  
  13. /**
  14.  * 配置
  15.  * @author lanx
  16.  * @date 2022/3/6
  17.  */
  18. @Slf4j
  19. @Configuration
  20. public class ZookeeperConfig {
  21.     @Autowired
  22.     private ZookeeperProperties zookeeperProperties ;
  23.  
  24.     private static CuratorFramework client = null ;
  25.     /**
  26.      * 初始化
  27.      */
  28.     @PostConstruct
  29.     public void init (){
  30.         //重试策略，初试时间1秒，重试10次
  31.         RetryPolicy policy = new ExponentialBackoffRetry(
  32.                 zookeeperProperties.getBaseSleepTimeMs(),
  33.                 zookeeperProperties.getMaxRetries());
  34.         //通过工厂创建Curator
  35.         client = CuratorFrameworkFactory.builder()
  36.                 .connectString(zookeeperProperties.getServer())
  37.                 .authorization("digest",zookeeperProperties.getDigest().getBytes())
  38.                 .connectionTimeoutMs(zookeeperProperties.getConnectionTimeoutMs())
  39.                 .sessionTimeoutMs(zookeeperProperties.getSessionTimeoutMs())
  40.                 .retryPolicy(policy).build();
  41.         //开启连接
  42.         client.start();
  43.         log.info("zookeeper 初始化完成...");
  44.     }
  45.     public static CuratorFramework getClient (){
  46.         return client ;
  47.     }
  48.     public static void closeClient (){
  49.         if (client != null){
  50.             client.close();
  51.         }
  52.     }
  53. }

Zookeeper节点数据监听

  1. package com.example.zk.config;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.curator.framework.CuratorFramework;
  5. import org.apache.curator.framework.recipes.cache.*;
  6. import org.springframework.boot.ApplicationArguments;
  7. import org.springframework.boot.ApplicationRunner;
  8. import org.springframework.stereotype.Component;
  9.  
  10. import java.util.concurrent.ExecutorService;
  11. import java.util.concurrent.Executors;
  12.  
  13. /**
  14.  * 节点数据监听
  15.  * @author lanx
  16.  * @date 2022/3/6
  17.  */
  18. @Component
  19. @Slf4j
  20. public class ZookListening implements ApplicationRunner {
  21.  
  22.     @Override
  23.     public void run(ApplicationArguments args) throws Exception {
  24.         /**
  25.          * 在注册监听器的时候，如果传入此参数，当事件触发时，逻辑由线程池处理
  26.          */
  27.         ExecutorService pool = Executors.newFixedThreadPool(2);
  28.         /**
  29.          * 监听数据节点的变化情况
  30.          */
  31.         final NodeCache nodeCache = new NodeCache(ZookeeperConfig.getClient(), "/lanxi", false);
  32.         nodeCache.start(true);
  33.         nodeCache.getListenable().addListener(
  34.                 /**
  35.                  * 此方法只监听创建节点和更新节点，在删除节点是不触发此操作
  36.                  */
  37.                 new NodeCacheListener() {
  38.                     @Override
  39.                     public void nodeChanged() throws Exception {
  40.                         log.info("Node data is changed, new data: {}" ,
  41.                                 new String(nodeCache.getCurrentData().getData()));
  42.                     }
  43.                 },
  44.                 pool
  45.         );
  46.  
  47.  
  48.         /**
  49.          * 监听子节点的变化情况
  50.          *
  51.          * 建立一个PathChildrenCache的缓存，第三个参数为是否接收节点数据内容，如果为 false 则不接受
  52.          */
  53.         final PathChildrenCache childrenCache = new PathChildrenCache(ZookeeperConfig.getClient(), "/llx", true);
  54.         //在初始时就进行缓存监听
  55.         childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
  56.         childrenCache.getListenable().addListener(
  57.                 new PathChildrenCacheListener() {
  58.                     /**
  59.                      * 监听节点的变更。 新建、修改、删除
  60.                      * @param client
  61.                      * @param event
  62.                      * @throws Exception
  63.                      */
  64.                     @Override
  65.                     public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
  66.                             throws Exception {
  67.                         switch (event.getType()) {
  68.                             case CHILD_ADDED:
  69.                                 System.out.println("CHILD_ADDED: " + event.getData().getPath());
  70.                                 break;
  71.                             case CHILD_REMOVED:
  72.                                 System.out.println("CHILD_REMOVED: " + event.getData().getPath());
  73.                                 break;
  74.                             case CHILD_UPDATED:
  75.                                 System.out.println("CHILD_UPDATED: " + event.getData().getPath());
  76.                                 break;
  77.                             default:
  78.                                 break;
  79.                         }
  80.                     }
  81.                 },
  82.                 pool
  83.         );
  84.  
  85.     }
  86. }

curator框架，使用 DistributedAtomicInteger 作为分布式计数器

  1.  //curator框架，使用 DistributedAtomicInteger 作为分布式计数器
  2.         DistributedAtomicInteger atomicInteger =
  3.                 new DistributedAtomicInteger(ZookeeperConfig.getClient(), "/llx", new RetryNTimes(3, 1000));
  4.         //复位
  5.         atomicInteger.forceSet(0);
  6.         AtomicValue<Integer> value = atomicInteger.add(1);
  7.         log.info("Boolean:{}", value.succeeded());
  8.         log.info("最新值:{}", value.postValue());
  9.         log.info("原始值:{}", value.preValue());

curator 框架，使用 InterProcessMutex 分布式锁

  1.  for (int i = 0; i < 10; i++) {
  2.             new Thread(new Runnable() {
  3.                 @Override
  4.                 public void run() {
  5.                     //分布式锁
  6.                     final InterProcessMutex lock = new InterProcessMutex(ZookeeperConfig.getClient(), "/llx");
  7.                     try {
  8.                         //获取锁
  9.                         lock.acquire();
  10.                         log.info("线程:{},执行业务逻辑", Thread.currentThread().getName());
  11.                         Thread.sleep(1000*10);
  12.                     } catch (Exception e) {
  13.                         e.printStackTrace();
  14.                     } finally {
  15.                         try {
  16.                             //释放锁
  17.                             lock.release();
  18.                         } catch (Exception e) {
  19.                             e.printStackTrace();
  20.                         }
  21.                     }
  22.                 }
  23.             }).start();
  24.         }

Zookeeper操作Service类

  1. package com.example.zk.service;
  2.  
  3. import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
  4. import org.apache.zookeeper.CreateMode;
  5.  
  6. import java.util.List;
  7.  
  8. public interface ZookeeperService {
  9.     /**
  10.      * 判断节点是否存在
  11.      */
  12.     boolean isExistNode (final String path) ;
  13.     /**
  14.      * 创建节点
  15.      */
  16.     void createNode (CreateMode mode, String path ) ;
  17.     /**
  18.      * 设置节点数据
  19.      */
  20.     void setNodeData (String path, String nodeData) ;
  21.     /**
  22.      * 创建节点
  23.      */
  24.     void createNodeAndData (CreateMode mode, String path , String nodeData) ;
  25.     /**
  26.      * 获取节点数据
  27.      */
  28.     String getNodeData (String path) ;
  29.     /**
  30.      * 获取节点下数据
  31.      */
  32.     List<String> getNodeChild (String path) ;
  33.     /**
  34.      * 是否递归删除节点
  35.      */
  36.     void deleteNode (String path,Boolean recursive) ;
  37.     /**
  38.      * 获取读写锁
  39.      */
  40.     InterProcessReadWriteLock getReadWriteLock (String path) ;
  41. }
  1. package com.example.zk.service.impl;
  2.  
  3. import com.example.zk.config.ZookeeperConfig;
  4. import com.example.zk.service.ZookeeperService;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.curator.framework.CuratorFramework;
  7. import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
  8. import org.apache.zookeeper.CreateMode;
  9. import org.apache.zookeeper.data.Stat;
  10. import org.springframework.stereotype.Service;
  11. import org.springframework.util.StringUtils;
  12.  
  13. import java.util.ArrayList;
  14. import java.util.List;
  15.  
  16. @Slf4j
  17. @Service
  18. public class ZookeeperServiceImpl implements ZookeeperService {
  19.     @Override
  20.     public boolean isExistNode(String path) {
  21.         CuratorFramework client = ZookeeperConfig.getClient();
  22.         client.sync() ;
  23.         try {
  24.             Stat stat = client.checkExists().forPath(path);
  25.             return client.checkExists().forPath(path) != null;
  26.         } catch (Exception e) {
  27.             log.error("isExistNode error...", e);
  28.             e.printStackTrace();
  29.         }
  30.         return false;
  31.     }
  32.     @Override
  33.     public void createNode(CreateMode mode, String path) {
  34.         CuratorFramework client = ZookeeperConfig.getClient() ;
  35.         try {
  36.             // 递归创建所需父节点
  37.             client.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
  38.         } catch (Exception e) {
  39.             log.error("createNode error...", e);
  40.             e.printStackTrace();
  41.         }
  42.     }
  43.     @Override
  44.     public void setNodeData(String path, String nodeData) {
  45.         CuratorFramework client = ZookeeperConfig.getClient() ;
  46.         try {
  47.             // 设置节点数据
  48.             client.setData().forPath(path, nodeData.getBytes("UTF-8"));
  49.         } catch (Exception e) {
  50.             log.error("setNodeData error...", e);
  51.             e.printStackTrace();
  52.         }
  53.     }
  54.     @Override
  55.     public void createNodeAndData(CreateMode mode, String path, String nodeData) {
  56.         CuratorFramework client = ZookeeperConfig.getClient() ;
  57.         try {
  58.             // 创建节点，关联数据
  59.             client.create().creatingParentsIfNeeded().withMode(mode)
  60.                   .forPath(path,nodeData.getBytes("UTF-8"));
  61.         } catch (Exception e) {
  62.             log.error("createNode error...", e);
  63.             e.printStackTrace();
  64.         }
  65.     }
  66.     @Override
  67.     public String getNodeData(String path) {
  68.         CuratorFramework client = ZookeeperConfig.getClient() ;
  69.         try {
  70.             // 数据读取和转换
  71.             byte[] dataByte = client.getData().forPath(path) ;
  72.             String data = new String(dataByte,"UTF-8") ;
  73.             if (!StringUtils.isEmpty(data)){
  74.                 return data ;
  75.             }
  76.         }catch (Exception e) {
  77.             log.error("getNodeData error...", e);
  78.             e.printStackTrace();
  79.         }
  80.         return null;
  81.     }
  82.     @Override
  83.     public List<String> getNodeChild(String path) {
  84.         CuratorFramework client = ZookeeperConfig.getClient() ;
  85.         List<String> nodeChildDataList = new ArrayList<>();
  86.         try {
  87.             // 节点下数据集
  88.             nodeChildDataList = client.getChildren().forPath(path);
  89.         } catch (Exception e) {
  90.             log.error("getNodeChild error...", e);
  91.             e.printStackTrace();
  92.         }
  93.         return nodeChildDataList;
  94.     }
  95.     @Override
  96.     public void deleteNode(String path, Boolean recursive) {
  97.         CuratorFramework client = ZookeeperConfig.getClient() ;
  98.         try {
  99.             if(recursive) {
  100.                 // 递归删除节点
  101.                 client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
  102.             } else {
  103.                 // 删除单个节点
  104.                 client.delete().guaranteed().forPath(path);
  105.             }
  106.         } catch (Exception e) {
  107.             log.error("deleteNode error...", e);
  108.             e.printStackTrace();
  109.         }
  110.     }
  111.     @Override
  112.     public InterProcessReadWriteLock getReadWriteLock(String path) {
  113.         CuratorFramework client = ZookeeperConfig.getClient() ;
  114.         // 写锁互斥、读写互斥
  115.         InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, path);
  116.         return readWriteLock ;
  117.     }
  118. }

Zookeeper操作Controller类

  1. package com.example.zk.web;
  2.  
  3. import com.example.zk.service.ZookeeperService;
  4. import org.apache.zookeeper.CreateMode;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.web.bind.annotation.GetMapping;
  7. import org.springframework.web.bind.annotation.RestController;
  8.  
  9. import java.util.List;
  10.  
  11. import static org.apache.zookeeper.CreateMode.PERSISTENT;
  12.  
  13. /**
  14.  * Zookeeper操作
  15.  * @author lanx
  16.  * @date 2022/3/6
  17.  */
  18. @RestController
  19. public class ZookeeperApi {
  20.     @Autowired
  21.     private ZookeeperService zookeeperService ;
  22.  
  23.     /**
  24.      * 获取节点下数据
  25.      * @param path
  26.      * @return
  27.      */
  28.     @GetMapping("/getNodeData")
  29.     public String getNodeData (String path) {
  30.         return zookeeperService.getNodeData(path) ;
  31.     }
  32.  
  33.     /**
  34.      * 判断节点是否存在
  35.      * @param path
  36.      * @return
  37.      */
  38.     @GetMapping("/isExistNode")
  39.     public boolean isExistNode (final String path){
  40.         return zookeeperService.isExistNode(path) ;
  41.     }
  42.  
  43.     /**
  44.      * 创建节点
  45.      * @param path
  46.      * @return
  47.      */
  48.     @GetMapping("/createNode")
  49.     public String createNode (String path ){
  50.         zookeeperService.createNode(CreateMode.PERSISTENT,path) ;
  51.         return "success" ;
  52.     }
  53.  
  54.     /**
  55.      * 节点设置数据
  56.      * @param path
  57.      * @param nodeData
  58.      * @return
  59.      */
  60.     @GetMapping("/setNodeData")
  61.     public String setNodeData (String path, String nodeData) {
  62.         zookeeperService.setNodeData(path,nodeData) ;
  63.         return "success" ;
  64.     }
  65.  
  66.     /**
  67.      * 创建节点并保存数据
  68.      * @param path
  69.      * @param nodeData
  70.      * @return
  71.      */
  72.     @GetMapping("/createNodeAndData")
  73.     public String createNodeAndData ( String path , String nodeData){
  74.         zookeeperService.createNodeAndData(PERSISTENT,path,nodeData) ;
  75.         return "success" ;
  76.     }
  77.  
  78.     /**
  79.      *获取节点下数据
  80.      * @param path
  81.      * @return
  82.      */
  83.     @GetMapping("/getNodeChild")
  84.     public List<String> getNodeChild (String path) {
  85.         return zookeeperService.getNodeChild(path) ;
  86.     }
  87.  
  88.     /**
  89.      * 是否递归删除节点
  90.      * @param path
  91.      * @param recursive
  92.      * @return
  93.      */
  94.     @GetMapping("/deleteNode")
  95.     public String deleteNode (String path,Boolean recursive) {
  96.         zookeeperService.deleteNode(path,recursive) ;
  97.         return "success" ;
  98.     }
  99. }

接口测试

http://127.0.0.1:8088/createNode?path=/llx/222

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/article/detail/53967
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号