快捷導(dǎo)航

SpringBoot3集成Zookeeper的代碼詳解

更新時(shí)間：2024年01月24日 08:09:56 作者：知了一笑

ZooKeeper是一個(gè)集中的服務(wù),用于維護(hù)配置信息、命名、提供分布式同步、提供組服務(wù),分布式應(yīng)用程序以某種形式使用所有這些類(lèi)型的服務(wù),本文將給大家介紹SpringBoot3集成Zookeeper的代碼,文中通過(guò)代碼示例介紹的非常詳細(xì),需要的朋友可以參考下

一、簡(jiǎn)介

ZooKeeper是一個(gè)集中的服務(wù)，用于維護(hù)配置信息、命名、提供分布式同步、提供組服務(wù)。分布式應(yīng)用程序以某種形式使用所有這些類(lèi)型的服務(wù)。

二、環(huán)境搭建

1、修改配置文件

# 1、拷貝一份樣本配置文件
cp zookeeper-3.8.3/conf/zoo_sample.cfg zookeeper-3.8.3/conf/zoo.cfg

# 2、修改數(shù)據(jù)文件地址，注意這里用本地路徑
dataDir=/local-path/zookeeper-3.8.3/data

# 3、添加一個(gè)配置，處理啟動(dòng)日志的提示：ZooKeeper audit is disabled.
audit.enable=true

2、服務(wù)啟動(dòng)

# 1、啟動(dòng)服務(wù)端
zookeeper-3.8.3/bin/zkServer.sh start

# 2、停止服務(wù)端
zookeeper-3.8.3/bin/zkServer.sh stop

# 3、啟動(dòng)客戶端
zookeeper-3.8.3/bin/zkCli.sh

3、客戶端測(cè)幾個(gè)增刪查的命令

[zk: localhost:2181(CONNECTED) 0] create /cicada smile1
Created /cicada
[zk: localhost:2181(CONNECTED) 1] get /cicada
smile1
[zk: localhost:2181(CONNECTED) 2] ls /
[cicada, zookeeper]
[zk: localhost:2181(CONNECTED) 3] delete /cicada

三、工程搭建

1、工程結(jié)構(gòu)

2、依賴管理

Curator是一組Java庫(kù)，它讓ZooKeeper的使用變得更加容易，這里的依賴實(shí)際是查詢匹配版本的時(shí)候走了個(gè)捷徑，也可以參考integration-redis包，熟悉下Spring的封裝策略。

<!-- Zookeeper組件 -->
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>${zookeeper.version}</version>
</dependency>
<!-- 包含Curator組件 -->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-zookeeper</artifactId>
    <version>${spring-integration.version}</version>
</dependency>

3、配置文件

配置腳本

zookeeper:
  #服務(wù)器地址
  connectString: 127.0.0.1:2181
  #會(huì)話超時(shí)時(shí)間
  sessionTimeoutMs: 3000
  #連接超時(shí)時(shí)間
  connectionTimeoutMs: 60000
  #最大重試次數(shù)
  maxRetries: 3
  #初始休眠時(shí)間
  baseSleepTimeMs: 1000

配置類(lèi)

@Configuration
public class ZookeeperConfig {

    @Value("${zookeeper.connectString}")
    private String connectString;
    
    @Value("${zookeeper.baseSleepTimeMs}")
    private int baseSleepTimeMs;

    @Value("${zookeeper.maxRetries}")
    private int maxRetries ;

    @Value("${zookeeper.connectionTimeoutMs}")
    int connectionTimeoutMs ;

    @Value("${zookeeper.sessionTimeoutMs}")
    int sessionTimeoutMs ;

    private static CuratorFramework client = null ;
    /**
     * 初始化
     */
    @PostConstruct
    public void init (){
        // 重試策略
        RetryPolicy policy = new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries);
        // 創(chuàng)建Curator
        client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .connectionTimeoutMs(connectionTimeoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .retryPolicy(policy).build();
        //開(kāi)啟連接
        client.start();
    }

    @Bean
    public CuratorFramework getClient (){
        return client ;
    }
}

四、ZooKeeper用法

測(cè)試幾個(gè)API方法，節(jié)點(diǎn)創(chuàng)建和添加數(shù)據(jù)，以及判斷和查詢數(shù)據(jù)，還有就是基于ZooKeeper提供的讀寫(xiě)鎖能力。

public class ConfigTest {

    @Autowired
    private CuratorFramework client ;

    @Test
    public void testCreate () throws Exception {
        // 創(chuàng)建一個(gè)持久化節(jié)點(diǎn)，斷開(kāi)連接時(shí)不會(huì)自動(dòng)刪除
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/path1");
    }

    @Test
    public void testExists () throws Exception {
        // 判斷節(jié)點(diǎn)是否存在，path2不存在所以stat2是null
        Stat stat1 = client.checkExists().forPath("/path1");
        System.out.println(stat1);
        Stat stat2 = client.checkExists().forPath("/path2");
        System.out.println(stat2);
    }

    @Test
    public void testSetData () throws Exception {
        // 設(shè)置節(jié)點(diǎn)數(shù)據(jù)
        client.setData().forPath("/path1", "data1".getBytes(StandardCharsets.UTF_8));
    }

    @Test
    public void testCreateAndSet () throws Exception {
        // 創(chuàng)建一個(gè)持久化節(jié)點(diǎn)并設(shè)置節(jié)點(diǎn)數(shù)據(jù)
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
                .forPath("/path3","data3".getBytes(StandardCharsets.UTF_8));
    }

    @Test
    public void testGetData () throws Exception {
        // 查詢節(jié)點(diǎn)數(shù)據(jù)
        byte[] data = client.getData().forPath("/path3");
        System.out.println(new String(data,StandardCharsets.UTF_8));
    }

    @Test
    public void testDelete () throws Exception {
        // 刪除節(jié)點(diǎn)
        client.delete().guaranteed().deletingChildrenIfNeeded().forPath("/path3");
    }

    @Test
    public void testReadLock () throws Exception {
        // 讀寫(xiě)鎖-讀
        InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client,"/lock-read");
        lock.readLock().acquire();
        System.out.println("獲取-ReadLock");
        lock.readLock().release();
    }

    @Test
    public void testWriteLock () throws Exception {
        // 讀寫(xiě)鎖-寫(xiě)
        InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client,"/lock-write");
        lock.writeLock().acquire();
        System.out.println("獲取-WriteLock");
        lock.writeLock().release();
    }
}