zookeeper-demo.7z

package com.liuyc; import lombok.extern.slf4j.Slf4j; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.recipes.cache.NodeCache; import org.apache.curator.framework.recipes.cache.NodeCacheListener; import org.apache.curator.framework.recipes.locks.InterProcessMutex; import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.data.Stat; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import java.nio.charset.StandardCharsets; @Slf4j @SpringBootTest public class BootZkClientApplicationTests { @Autowired private CuratorFramework curatorFramework; @Test public void createNode() throws Exception { //添加持久节点 String path1 = curatorFramework.create().forPath("/curatorTest1","aaaaaaaaaaa".getBytes()); System.out.println("path1:"+path1); //添加临时序号节点 String path2 = curatorFramework.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/curatorTest1", "bbbbbbbbb".getBytes()); System.out.println("path2:"+path2); System.in.read(); } @Test public void getData() throws Exception { byte[] bytes = curatorFramework.getData().forPath("/curatorTest1"); System.out.println(new String(bytes, StandardCharsets.UTF_8)); } @Test public void setData() throws Exception { Stat stat = curatorFramework.setData().forPath("/curatorTest1", "ccccccc".getBytes()); System.out.println(stat); } /** * 创建子节点,如父节点为空，则一起创建 * @throws Exception */ @Test public void createWithParent() throws Exception { String path = curatorFramework.create().creatingParentsIfNeeded().forPath("/curatorTest2/curatorTestTest1", "ggggg".getBytes()); System.out.println(path); } /** * 删除节点 * @throws Exception */ @Test public void delete() throws Exception { Void unused = curatorFramework.delete().forPath("/curatorTest2"); System.out.println(unused); } /** * 删除节点,如有子节点则一起删除 * @throws Exception */ @Test public void deleteWith() throws Exception { Void unused = curatorFramework.delete().deletingChildrenIfNeeded().forPath("/curatorTest2"); System.out.println(unused); } /** * 监听节点 */ @Test public void addNodeWatch() throws Exception { NodeCache nodeCache = new NodeCache(curatorFramework, "/curatorTest1"); nodeCache.getListenable().addListener(new NodeCacheListener() { @Override public void nodeChanged() throws Exception { System.out.println("监听执行"); byte[] bytes = curatorFramework.getData().forPath("/curatorTest1"); System.out.println(new String(bytes, StandardCharsets.UTF_8)); } }); nodeCache.start(); System.in.read(); } /** * 读锁 */ @Test public void getGetReadLock() throws Exception { InterProcessReadWriteLock interProcessReadWriteLock =new InterProcessReadWriteLock(curatorFramework,"/lock1"); InterProcessMutex interProcessMutex = interProcessReadWriteLock.readLock(); System.out.println("等待获取读锁对象"); //获取锁 interProcessMutex.acquire(); for (int i = 0; i < 100; i++) { Thread.sleep(3000); System.out.println(i); } interProcessMutex.release(); System.out.println("等待释放读锁"); } /** * 读锁 */ @Test public void getGetReadLock1() throws Exception { InterProcessReadWriteLock interProcessReadWriteLock =new InterProcessReadWriteLock(curatorFramework,"/lock1"); InterProcessMutex interProcessMutex = interProcessReadWriteLock.readLock(); System.out.println("等待获取读锁对象"); //获取锁 interProcessMutex.acquire(); for (int i = 0; i < 100; i++) { Thread.sleep(3000); System.out.println(i); } interProcessMutex.release(); System.out.println("等待释放读锁"); } /** * 写锁 */ @Test public void getGetWriteLock() throws Exception { InterProcessReadWriteLock interProcessReadWriteLock =new InterProcessReadWriteLock(curatorFramework,"/lock1"); InterProcessMutex interProcessMutex = interProcessReadWriteLock.writeLock(); System.out.println("等待获取写锁对象"); //获取锁 interProcessMutex.acquire(); for (int i = 0; i < 100; i++) { Thread.sleep(3000); System.out.println(i); } interProcessMutex.release(); System.out.println("等待释放写锁"); } }