diff --git a/net/README.md b/net/README.md index 0082bf67..637839db 100644 --- a/net/README.md +++ b/net/README.md @@ -33,3 +33,14 @@ task，通用任务线程池 - [单机服务器教程](src/test/java/com/zfoo/net/core/tcp/server/TcpServerTest.java) - [RPC教程](src/test/java/com/zfoo/net/core/provider/ProviderTest.java) - [网关教程](src/test/java/com/zfoo/net/core/gateway/GatewayTest.java) + +### Ⅴ. 依赖的中间件zookeeper教程 + +- [zk基础的增删改查](src/test/java/com/zfoo/net/zookeeper/base) +- [curator的增删改查](src/test/java/com/zfoo/net/zookeeper/curator) +- [zk分布式ID](src/test/java/com/zfoo/net/zookeeper/recipes/atomicint) +- [zk分布式Barrier](src/test/java/com/zfoo/net/zookeeper/recipes/distributedbarrier) +- [zk分布式锁](src/test/java/com/zfoo/net/zookeeper/recipes/distributedbarrier) +- [zk分布式选举](src/test/java/com/zfoo/net/zookeeper/recipes/mastersel) +- [zk数据发布/订阅](src/test/java/com/zfoo/net/zookeeper/recipes/nodecache) + diff --git a/net/src/test/java/com/zfoo/net/zookeeper/ZookeeperConstantTest.java b/net/src/test/java/com/zfoo/net/zookeeper/ZookeeperConstantTest.java new file mode 100644 index 00000000..c19fa8a2 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/ZookeeperConstantTest.java @@ -0,0 +1,29 @@ +package com.zfoo.net.zookeeper; + +/** + * 引子和趣闻： + * Zookeeper名字的由来是比较有趣的，下面的片段摘抄自《从PAXOS到ZOOKEEPER分布式一致性原理与实践》一书： + * Zookeeper最早起源于雅虎的研究院的一个研究小组。在当时，研究人员发现，在雅虎内部很多大型的系统需要依赖一个类似的系统进行分布式协调， + * 但是这些系统往往存在分布式单点问题。所以雅虎的开发人员就试图开发一个通用的无单点问题的分布式协调框架。 + * 在立项初期，考虑到很多项目都是用动物的名字来命名的(例如著名的Pig项目)，雅虎的工程师希望给这个项目也取一个动物的名字。 + * 时任研究院的首席科学家Raghu Ramakrishnan开玩笑说：再这样下去，我们这儿就变成动物园了。 + * 此话一出，大家纷纷表示就叫动物园管理员吧——因为各个以动物命名的分布式组件放在一起，雅虎的整个分布式系统看上去就像一个大型的动物园了， + * 而Zookeeper正好用来进行分布式环境的协调——于是，Zookeeper的名字由此诞生了。 + *

+ * Curator无疑是Zookeeper客户端中的瑞士军刀，它译作"馆长"或者''管理者''，不知道是不是开发小组有意而为之， + * 笔者猜测有可能这样命名的原因是说明Curator就是Zookeeper的馆长(脑洞有点大：Curator就是动物园的园长)。 + *

+ * curator-framework：对zookeeper的底层api的一些封装 + * curator-client：提供一些客户端的操作，例如重试策略等 + * curator-recipes：封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式计数器、分布式Barrier等 + * + * @author jaysunxiao + * @version 1.0 + * @since 2018-04-03 15:29 + */ + +public abstract class ZookeeperConstantTest { + + public static String URL = "localhost:2181"; + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeASyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeASyncTest.java new file mode 100644 index 00000000..7310298c --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeASyncTest.java @@ -0,0 +1,64 @@ +package com.zfoo.net.zookeeper.base.create; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.*; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooDefs.Ids; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; +import java.util.concurrent.CountDownLatch; + +@Ignore +public class CreateNodeASyncTest { + + + private static CountDownLatch connectedSemaphore = new CountDownLatch(1); + + // 异步创建一个节点 + @Test + public void test() throws IOException, InterruptedException { + ZooKeeper zookeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new CreateNodeASyncWatcher()); + + System.out.println(zookeeper.getState()); + + connectedSemaphore.await(); + + String content = "Hello Zookeeper!"; + + // 异步创建，会回调processResult() + zookeeper.create("/node_test", content.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new StringCallback(), "hello"); + + Thread.sleep(Integer.MAX_VALUE); + } + + static class CreateNodeASyncWatcher implements Watcher { + @Override + public void process(WatchedEvent event) { + System.out.println("接受事件" + event); + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + connectedSemaphore.countDown(); + } + } + } + } + + static class StringCallback implements AsyncCallback.StringCallback { + @Override + public void processResult(int rc, String path, Object ctx, String name) { + StringBuilder sb = new StringBuilder(); + sb.append("rc=" + rc).append("\n"); + sb.append("path=" + path).append("\n"); + sb.append("ctx=" + ctx).append("\n"); + sb.append("name=" + name); + System.out.println(sb.toString()); + + } + + + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeSyncAuthTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeSyncAuthTest.java new file mode 100644 index 00000000..4cb5e01c --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeSyncAuthTest.java @@ -0,0 +1,98 @@ +package com.zfoo.net.zookeeper.base.create; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.*; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooDefs.Perms; +import org.apache.zookeeper.data.ACL; +import org.apache.zookeeper.data.Id; +import org.apache.zookeeper.data.Stat; +import org.apache.zookeeper.server.auth.DigestAuthenticationProvider; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; + +@Ignore +public class CreateNodeSyncAuthTest { + + + private static ZooKeeper zookeeper; + private static boolean somethingDone = false; + private static Stat stat = new Stat(); + + @Test + public void test() throws IOException, InterruptedException { + zookeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new CreateNodeSyncAuthWatcher()); + System.out.println(zookeeper.getState()); + Thread.sleep(Integer.MAX_VALUE); + } + + @Test + public void generateDigestTest() throws NoSuchAlgorithmException { + String digest = DigestAuthenticationProvider.generateDigest("jaysunxiao:123456"); + System.out.println(digest); + } + + + static class CreateNodeSyncAuthWatcher implements Watcher { + + @Override + public void process(WatchedEvent event) { + System.out.println("收到事件：" + event); + if (event.getState() == KeeperState.SyncConnected) { + if (!somethingDone && event.getType() == EventType.None && null == event.getPath()) { + doSomething(); + } + } + } + + /* + * 权限模式(scheme): ip, digest + * 授权对象(ID) + * ip权限模式: 具体的ip地址 + * digest权限模式: username:Base64(SHA-1(username:password)) + * 权限(permission): create(C), DELETE(D),READ(R), WRITE(W), ADMIN(A) + * 注：单个权限，完全权限，复合权限 + * + * 权限组合: scheme + ID + permission + * */ + private void doSomething() { + try { + // ACL access control list + ACL aclIp = new ACL(Perms.READ, new Id("ip", "192.168.1.105")); + + ACL aclDigest = new ACL(Perms.READ | Perms.WRITE, new Id("digest", DigestAuthenticationProvider.generateDigest("jaysunxiao:123456")));// 用户名+密码 + + ArrayList acls = new ArrayList<>(); + acls.add(aclDigest); + acls.add(aclIp); + String path = zookeeper.create("/node_test", "Hello Zookeeper!".getBytes(), acls, CreateMode.PERSISTENT); + System.out.println("return path:" + path); + } catch (KeeperException | InterruptedException | NoSuchAlgorithmException e) { + e.printStackTrace(); + } + + try { + // 没有取得权限之前，不能访问/node_test + zookeeper.getData("/node_test", false, stat); + somethingDone = true; + } catch (Exception e) { + e.printStackTrace(); + } + + try { + // 下面两句设置ACL的哪些author可以访问 + zookeeper.addAuthInfo("digest", "jaysunxiao:123456".getBytes()); + zookeeper.getData("/node_test", false, stat); + somethingDone = true; + } catch (Exception e) { + e.printStackTrace(); + } + } + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeSyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeSyncTest.java new file mode 100644 index 00000000..e1b30b14 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/create/CreateNodeSyncTest.java @@ -0,0 +1,49 @@ +package com.zfoo.net.zookeeper.base.create; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.*; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooDefs.Ids; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; +import java.util.concurrent.CountDownLatch; + + +@Ignore +public class CreateNodeSyncTest { + + private static CountDownLatch connectedSemaphore = new CountDownLatch(1); + + // 同步创建节点 + @Test + public void test() throws IOException, InterruptedException, KeeperException { + // 这个创建zookeeper的链接是异步的 + ZooKeeper zookeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new CreateNodeSyncWatcher()); + + System.out.println(zookeeper.getState()); + + connectedSemaphore.await(); + + // 无论是同步还是异步，Zookeeper都不支持递归创建，既无法在父节点不存在的情况下创建一个子节点。如果创建的节点存在，则抛出异常。 + String path = zookeeper.create("/node_test", "Hello Zookeeper!".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + System.out.println("return path:" + path); + + Thread.sleep(Integer.MAX_VALUE); + } + + static class CreateNodeSyncWatcher implements Watcher { + @Override + public void process(WatchedEvent event) { + System.out.println("接受事件" + event); + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + connectedSemaphore.countDown(); + } + } + } + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/delete/DeleteNodeASyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/delete/DeleteNodeASyncTest.java new file mode 100644 index 00000000..770968df --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/delete/DeleteNodeASyncTest.java @@ -0,0 +1,59 @@ +package com.zfoo.net.zookeeper.base.delete; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.AsyncCallback; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; +import java.util.concurrent.CountDownLatch; + +@Ignore +public class DeleteNodeASyncTest { + + + private static CountDownLatch connectedSemaphore = new CountDownLatch(1); + + @Test + public void test() throws IOException, InterruptedException { + ZooKeeper zooKeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new DeleteNodeASyncWatcher()); + System.out.println(zooKeeper.getState().toString()); + + connectedSemaphore.await(); + + zooKeeper.delete("/node_test", -1, new IVoidCallback(), null); + + Thread.sleep(Integer.MAX_VALUE); + } + + + static class DeleteNodeASyncWatcher implements Watcher { + @Override + public void process(WatchedEvent event) { + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + connectedSemaphore.countDown(); + } + } + } + } + + static class IVoidCallback implements AsyncCallback.VoidCallback { + @Override + public void processResult(int rc, String path, Object ctx) { + StringBuilder sb = new StringBuilder(); + sb.append("rc=" + rc).append("\n"); + sb.append("path" + path).append("\n"); + sb.append("ctx=" + ctx).append("\n"); + System.out.println(sb.toString()); + + } + + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/delete/DeleteNodeSyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/delete/DeleteNodeSyncTest.java new file mode 100644 index 00000000..00f4c38b --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/delete/DeleteNodeSyncTest.java @@ -0,0 +1,45 @@ +package com.zfoo.net.zookeeper.base.delete; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; +import java.util.concurrent.CountDownLatch; + +@Ignore +public class DeleteNodeSyncTest { + + private static CountDownLatch connectedSemaphore = new CountDownLatch(1); + + @Test + public void test() throws IOException, InterruptedException, KeeperException { + ZooKeeper zooKeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new DeleteNodeSyncWatcher()); + System.out.println(zooKeeper.getState().toString()); + + connectedSemaphore.await(); + + // 如果一个节点存在至少一个子节点的话，那么该节点将无法被直接删除，必须先删除掉其它所有子节点 + zooKeeper.delete("/node_test", -1); + + Thread.sleep(Integer.MAX_VALUE); + } + + static class DeleteNodeSyncWatcher implements Watcher { + @Override + public void process(WatchedEvent event) { + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + connectedSemaphore.countDown(); + } + } + } + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetChildrenASyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetChildrenASyncTest.java new file mode 100644 index 00000000..c3d585b4 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetChildrenASyncTest.java @@ -0,0 +1,71 @@ +package com.zfoo.net.zookeeper.base.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.AsyncCallback; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; +import java.util.List; + +@Ignore +public class GetChildrenASyncTest { + + private static ZooKeeper zooKeeper; + + @Test + public void test() throws IOException, InterruptedException { + zooKeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new GetChildrenASyncWatcher()); + System.out.println(zooKeeper.getState().toString()); + Thread.sleep(Integer.MAX_VALUE); + } + + + static class GetChildrenASyncWatcher implements Watcher { + + private void doSomething(ZooKeeper zookeeper) { + try { + // 异步方法，不会阻塞 + zooKeeper.getChildren("/", true, new IChildren2Callback(), null); + System.out.println("不会阻塞"); + } catch (Exception e) { + e.printStackTrace(); + } + } + + @Override + public void process(WatchedEvent event) { + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + doSomething(zooKeeper); + } else { + if (event.getType() == EventType.NodeChildrenChanged) { + zooKeeper.getChildren(event.getPath(), true, new IChildren2Callback(), null); + } + } + } + } + } + + + static class IChildren2Callback implements AsyncCallback.Children2Callback { + @Override + public void processResult(int rc, String path, Object ctx, List children, Stat stat) { + StringBuilder sb = new StringBuilder(); + sb.append("rc=" + rc).append("\n"); + sb.append("path=" + path).append("\n"); + sb.append("ctx=" + ctx).append("\n"); + sb.append("children=" + children).append("\n"); + sb.append("stat=" + stat).append("\n"); + System.out.println(sb.toString()); + + } + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetChildrenSyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetChildrenSyncTest.java new file mode 100644 index 00000000..10d5eb84 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetChildrenSyncTest.java @@ -0,0 +1,52 @@ +package com.zfoo.net.zookeeper.base.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.CountDownLatch; + +@Ignore +public class GetChildrenSyncTest { + + private static CountDownLatch connectedSemaphore = new CountDownLatch(1); + private static ZooKeeper zk = null; + + @Test + public void test() throws IOException, InterruptedException, KeeperException { + zk = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new GetChildrenSyncWatcher()); + System.out.println(zk.getState().toString()); + + connectedSemaphore.await(); + + List childrenList = zk.getChildren("/", true); + System.out.println(childrenList); + + Thread.sleep(Integer.MAX_VALUE); + } + + static class GetChildrenSyncWatcher implements Watcher { + @Override + public void process(WatchedEvent event) { + if (KeeperState.SyncConnected == event.getState()) { + if (EventType.None == event.getType() && null == event.getPath()) { + connectedSemaphore.countDown(); + } else if (event.getType() == EventType.NodeChildrenChanged) { + try { + System.out.println("ReGet Child:" + zk.getChildren(event.getPath(), true)); + } catch (Exception e) { + e.printStackTrace(); + } + } + } + } + } +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetDataASyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetDataASyncTest.java new file mode 100644 index 00000000..914c48c6 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetDataASyncTest.java @@ -0,0 +1,63 @@ +package com.zfoo.net.zookeeper.base.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.*; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; +import java.util.concurrent.CountDownLatch; + +@Ignore +public class GetDataASyncTest { + + + private static CountDownLatch connectedSemaphore = new CountDownLatch(1); + private static ZooKeeper zk; + + // ZooKeeper API 获取节点数据内容，使用异步(async)接口。 + @Test + public void test() throws IOException, InterruptedException, KeeperException { + zk = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new GetDataASyncWatcher()); + System.out.println(zk.getState().toString()); + + String path = "/node_test"; + connectedSemaphore.await(); + + zk.getData(path, true, new IDataCallback(), null); + + Thread.sleep(Integer.MAX_VALUE); + } + + static class GetDataASyncWatcher implements Watcher { + @Override + public void process(WatchedEvent event) { + if (KeeperState.SyncConnected == event.getState()) { + if (EventType.None == event.getType() && null == event.getPath()) { + connectedSemaphore.countDown(); + } else if (event.getType() == EventType.NodeDataChanged) { + try { + zk.getData(event.getPath(), true, new IDataCallback(), null); + } catch (Exception e) { + e.printStackTrace(); + } + } + } + } + } + + static class IDataCallback implements AsyncCallback.DataCallback { + public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) { + System.out.println(rc + ", " + path + ", " + new String(data)); + System.out.println( + stat.getCzxid() + "," + + stat.getMzxid() + "," + + stat.getVersion() + ); + } + } +} + diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetDataSyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetDataSyncTest.java new file mode 100644 index 00000000..b11c2b5f --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/query/GetDataSyncTest.java @@ -0,0 +1,64 @@ +package com.zfoo.net.zookeeper.base.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; + +@Ignore +public class GetDataSyncTest { + + private static ZooKeeper zooKeeper; + + private static Stat stat = new Stat(); + + @Test + public void test() throws IOException, InterruptedException { + zooKeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new GetDataASyncWatcher()); + System.out.println(zooKeeper.getState().toString()); + Thread.sleep(Integer.MAX_VALUE); + } + + + static class GetDataASyncWatcher implements Watcher { + + private void doSomething() { + try { + System.out.println(new String(zooKeeper.getData("/node_test", true, stat))); + } catch (KeeperException | InterruptedException e) { + e.printStackTrace(); + } + + } + + @Override + public void process(WatchedEvent event) { + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + doSomething(); + } else { + if (event.getType() == EventType.NodeDataChanged) { + try { + System.out.println(new String(zooKeeper.getData(event.getPath(), true, stat))); + System.out.println("stat:" + stat); + } catch (Exception e) { + e.printStackTrace(); + } + + } + } + + } + } + } + + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/query/NodeExistsASyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/query/NodeExistsASyncTest.java new file mode 100644 index 00000000..8671898e --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/query/NodeExistsASyncTest.java @@ -0,0 +1,73 @@ +package com.zfoo.net.zookeeper.base.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.AsyncCallback; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; + +@Ignore +public class NodeExistsASyncTest { + + + private static ZooKeeper zooKeeper; + + @Test + public void test() throws IOException, InterruptedException { + zooKeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new NodeExistsASynccWatcher()); + System.out.println(zooKeeper.getState().toString()); + Thread.sleep(Integer.MAX_VALUE); + } + + + static class NodeExistsASynccWatcher implements Watcher { + + private void doSomething(ZooKeeper zookeeper) { + zooKeeper.exists("/node_test", true, new IStateCallback(), null); + } + + @Override + public void process(WatchedEvent event) { + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + doSomething(zooKeeper); + } else { + try { + if (event.getType() == EventType.NodeCreated) { + System.out.println(event.getPath() + " created"); + zooKeeper.exists(event.getPath(), true, new IStateCallback(), null); + } else if (event.getType() == EventType.NodeDataChanged) { + System.out.println(event.getPath() + " updated"); + zooKeeper.exists(event.getPath(), true, new IStateCallback(), null); + } else if (event.getType() == EventType.NodeDeleted) { + System.out.println(event.getPath() + " deleted"); + zooKeeper.exists(event.getPath(), true, new IStateCallback(), null); + } + + } catch (Exception e) { + e.printStackTrace(); + } + + } + + } + } + } + + + static class IStateCallback implements AsyncCallback.StatCallback { + @Override + public void processResult(int rc, String path, Object ctx, Stat stat) { + System.out.println("rc:" + rc); + + } + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/query/NodeExistsSyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/query/NodeExistsSyncTest.java new file mode 100644 index 00000000..c917ec9a --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/query/NodeExistsSyncTest.java @@ -0,0 +1,70 @@ +package com.zfoo.net.zookeeper.base.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; + +@Ignore +public class NodeExistsSyncTest { + + private static ZooKeeper zooKeeper; + + @Test + public void test() throws IOException, InterruptedException { + zooKeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new NodeExistsSyncWatcher()); + System.out.println(zooKeeper.getState().toString()); + Thread.sleep(Integer.MAX_VALUE); + } + + static class NodeExistsSyncWatcher implements Watcher { + + private void doSomething(ZooKeeper zooKeeper) { + try { + Stat stat = zooKeeper.exists("/node_test", true); + System.out.println(stat); + + } catch (Exception e) { + e.printStackTrace(); + } + + } + + @Override + public void process(WatchedEvent event) { + + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + doSomething(zooKeeper); + } else { + try { + if (event.getType() == EventType.NodeCreated) { + System.out.println(event.getPath() + " created"); + System.out.println(zooKeeper.exists(event.getPath(), true)); + } else if (event.getType() == EventType.NodeDataChanged) { + System.out.println(event.getPath() + " updated"); + System.out.println(zooKeeper.exists(event.getPath(), true)); + } else if (event.getType() == EventType.NodeDeleted) { + System.out.println(event.getPath() + " deleted"); + System.out.println(zooKeeper.exists(event.getPath(), true)); + } + + } catch (Exception e) { + e.printStackTrace(); + } + + } + + } + } + } + + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/update/UpdateNodeASyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/update/UpdateNodeASyncTest.java new file mode 100644 index 00000000..e9e0dc42 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/update/UpdateNodeASyncTest.java @@ -0,0 +1,59 @@ +package com.zfoo.net.zookeeper.base.update; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.AsyncCallback; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; + +@Ignore +public class UpdateNodeASyncTest { + + + private static ZooKeeper zooKeeper; + + @Test + public void test() throws IOException, InterruptedException { + zooKeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new UpdateNodeASyncWatcher()); + System.out.println(zooKeeper.getState().toString()); + Thread.sleep(Integer.MAX_VALUE); + } + + static class UpdateNodeASyncWatcher implements Watcher { + private void doSomething(WatchedEvent event) { + String updateContent = "new content!!!"; + zooKeeper.setData("/node_test", updateContent.getBytes(), -1, new IStatCallback(), null); + } + + @Override + public void process(WatchedEvent event) { + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + doSomething(event); + } + } + } + } + + static class IStatCallback implements AsyncCallback.StatCallback { + @Override + public void processResult(int rc, String path, Object ctx, Stat stat) { + StringBuilder sb = new StringBuilder(); + sb.append("rc=" + rc).append("\n"); + sb.append("path" + path).append("\n"); + sb.append("ctx=" + ctx).append("\n"); + sb.append("Stat=" + stat).append("\n"); + System.out.println(sb.toString()); + + } + + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/base/update/UpdateNodeSyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/base/update/UpdateNodeSyncTest.java new file mode 100644 index 00000000..c8f403af --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/base/update/UpdateNodeSyncTest.java @@ -0,0 +1,49 @@ +package com.zfoo.net.zookeeper.base.update; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; + +@Ignore +public class UpdateNodeSyncTest { + + private static ZooKeeper zooKeeper; + + @Test + public void test() throws IOException, InterruptedException { + zooKeeper = new ZooKeeper(ZookeeperConstantTest.URL, 5000, new UpdateNodeSyncWatcher()); + System.out.println(zooKeeper.getState().toString()); + Thread.sleep(Integer.MAX_VALUE); + } + + static class UpdateNodeSyncWatcher implements Watcher { + private void doSomething() { + try { + Stat stat = zooKeeper.setData("/node_test", "Hello Zookeeper!".getBytes(), -1); + System.out.println("stat:" + stat); + } catch (InterruptedException | KeeperException e) { + throw new RuntimeException(e); + } + } + + @Override + public void process(WatchedEvent event) { + if (event.getState() == KeeperState.SyncConnected) { + if (event.getType() == EventType.None && null == event.getPath()) { + doSomething(); + } + + } + } + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/cache/PathCacheTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/cache/PathCacheTest.java new file mode 100644 index 00000000..78495b90 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/cache/PathCacheTest.java @@ -0,0 +1,63 @@ +package com.zfoo.net.zookeeper.curator.cache; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import com.zfoo.util.ThreadUtils; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.recipes.cache.PathChildrenCache; +import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent; +import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.junit.Ignore; +import org.junit.Test; + +/** + * @author jaysunxiao + * @version 1.0 + * @since 2019-08-25 09:43 + */ +@Ignore +public class PathCacheTest { + + private static CuratorFramework curator = CuratorFrameworkFactory.builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) +// .retryPolicy(new RetryNTimes(1, 1000)) + .build(); + + @Test + public void test() { + curator.start(); + + PathChildrenCache pathCache = new PathChildrenCache(curator, "/test", true); + try { + pathCache.start(); + } catch (Exception e) { + e.printStackTrace(); + } + + + //节点变化的监听器 + pathCache.getListenable().addListener(new PathChildrenCacheListener() { + @Override + public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception { + System.out.println("pathCache ------ Type:" + event.getType() + ","); + System.out.println(event.getData().getPath()); + } + }); + + while (true) { + ThreadUtils.sleep(3000); + try { + pathCache.getCurrentData().stream() + .forEach(it -> System.out.println(it.getPath() + "->" + it.getData())); + } catch (Exception e) { + e.printStackTrace(); + } + + } + + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/cache/TreeCacheTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/cache/TreeCacheTest.java new file mode 100644 index 00000000..c082266f --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/cache/TreeCacheTest.java @@ -0,0 +1,71 @@ +package com.zfoo.net.zookeeper.curator.cache; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import com.zfoo.util.ThreadUtils; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.api.UnhandledErrorListener; +import org.apache.curator.framework.recipes.cache.TreeCache; +import org.apache.curator.framework.recipes.cache.TreeCacheEvent; +import org.apache.curator.framework.recipes.cache.TreeCacheListener; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.junit.Ignore; +import org.junit.Test; + +/** + * @author jaysunxiao + * @version 1.0 + * @since 2019-08-25 09:43 + */ +@Ignore +public class TreeCacheTest { + + private static CuratorFramework curator = CuratorFrameworkFactory.builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) +// .retryPolicy(new RetryNTimes(1, 1000)) + .build(); + + @Test + public void test() { + curator.start(); + + TreeCache treeCache = new TreeCache(curator, "/test"); + try { + treeCache.start(); + } catch (Exception e) { + e.printStackTrace(); + } + //添加错误监听器 + treeCache.getUnhandledErrorListenable().addListener(new UnhandledErrorListener() { + public void unhandledError(String s, Throwable throwable) { + System.out.println(".错误原因：" + throwable.getMessage() + "\n==============\n"); + } + }); + + //节点变化的监听器 + treeCache.getListenable().addListener(new TreeCacheListener() { + public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception { + System.out.println("treeCache ------ Type:" + treeCacheEvent.getType() + ","); + System.out.println(treeCacheEvent.getData().getPath()); + + } + }); + + while (true) { + ThreadUtils.sleep(3000); + try { + System.out.println("-------->" + treeCache.getCurrentData("/test/a")); + + treeCache.getCurrentChildren("/test").entrySet().stream() + .forEach(it -> System.out.println(it.getKey() + "->" + it.getValue())); + } catch (Exception e) { + e.printStackTrace(); + } + + } + + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeASyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeASyncTest.java new file mode 100644 index 00000000..686aceed --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeASyncTest.java @@ -0,0 +1,65 @@ +package com.zfoo.net.zookeeper.curator.create; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.api.BackgroundCallback; +import org.apache.curator.framework.api.CuratorEvent; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.apache.zookeeper.CreateMode; +import org.junit.Ignore; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * @author jaysunxiao + * @version 1.0 + * @since 2018-08-01 16:43 + */ +@Ignore +public class CreateNodeASyncTest { + + private static CuratorFramework curator = CuratorFrameworkFactory.builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .build(); + + private static CountDownLatch semaphore = new CountDownLatch(2); + + private static ExecutorService executorService = Executors.newFixedThreadPool(2); + + @Test + public void test() throws Exception { + curator.start(); + + System.out.println("Main thread: " + Thread.currentThread().getName()); + + // 此处传入了自定义的Executor + curator.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() { + @Override + public void processResult(CuratorFramework client, CuratorEvent event) throws Exception { + System.out.println("event[code: " + event.getResultCode() + ", type: " + event.getType() + "]"); + System.out.println("Thread of processResult: " + Thread.currentThread().getName()); + semaphore.countDown(); + } + }, executorService).forPath("/node_test", "Hello Zookeeper!".getBytes()); + + // 此处没有传入自定义的Executor，会默认使用Zookeeper内部的EventThread单线程处理 + curator.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() { + @Override + public void processResult(CuratorFramework client, CuratorEvent event) throws Exception { + System.out.println("event[code: " + event.getResultCode() + ", type: " + event.getType() + "]"); + System.out.println("Thread of processResult: " + Thread.currentThread().getName()); + semaphore.countDown(); + } + }).forPath("/node_test", "Hello Zookeeper!".getBytes()); + + semaphore.await(); + executorService.shutdown(); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeAuthTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeAuthTest.java new file mode 100644 index 00000000..e86c43f0 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeAuthTest.java @@ -0,0 +1,80 @@ +package com.zfoo.net.zookeeper.curator.create; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.RetryUntilElapsed; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.ZooDefs.Perms; +import org.apache.zookeeper.data.ACL; +import org.apache.zookeeper.data.Id; +import org.apache.zookeeper.data.Stat; +import org.apache.zookeeper.server.auth.DigestAuthenticationProvider; +import org.junit.Ignore; +import org.junit.Test; + +import java.util.ArrayList; + +@Ignore +public class CreateNodeAuthTest { + + @Test + public void createTest() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework curator = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + curator.start(); + + ACL aclIp = new ACL(Perms.READ, new Id("ip", "192.168.1.105")); + ACL aclDigest = new ACL(Perms.READ | Perms.WRITE, new Id("digest", DigestAuthenticationProvider.generateDigest("jaysunxiao:123456"))); + ArrayList acls = new ArrayList<>(); + acls.add(aclDigest); + acls.add(aclIp); + + // 使用jaysunxiao:123456创建一个授权节点 + String path = curator.create() + .creatingParentsIfNeeded() + .withMode(CreateMode.PERSISTENT) + .withACL(acls) + .forPath("/node_test", "Hello Zookeeper!".getBytes()); + + System.out.println(path); + + Thread.sleep(Integer.MAX_VALUE); + } + + + @Test + public void getDataWithAuthTest() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework curator = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .authorization("digest", "jaysunxiao:123456".getBytes()) // 授权访问 + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + curator.start(); + + Stat stat = new Stat(); + + byte[] ret = curator.getData().storingStatIn(stat).forPath("/node_test"); + + System.out.println(new String(ret)); + + System.out.println(stat); + } + + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeSyncTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeSyncTest.java new file mode 100644 index 00000000..3ca45dd0 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/create/CreateNodeSyncTest.java @@ -0,0 +1,47 @@ +package com.zfoo.net.zookeeper.curator.create; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.RetryUntilElapsed; +import org.apache.zookeeper.CreateMode; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore +public class CreateNodeSyncTest { + //RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); // /ekspəʊ'nenʃl/ adj. 指数的; 幂数的 + //RetryPolicy retryPolicy = new RetryNTimes(5, 1000); // 最多重试5次，每次一秒 + + @Test + public void test() throws Exception { + // 重试的时间超过最大时间后，就不再重试 + // 参数说明: + // maxElapsedTimeMs: 最大的重试时间 + // sleepMsBetweenRetries：每次重试的间隔时间 + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000);// elapsed /ɪ'læps/ v. 过去; 消逝 + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + // .namespace(basePath); 指定namespace，该客户端的如何操作，都是基于该相对目录进行的。 + .build(); + + client.start(); + + String path = client.create() + .creatingParentsIfNeeded() + .withMode(CreateMode.EPHEMERAL) + .forPath("/node_test", "Hello Zookeeper!".getBytes()); + + System.out.println(path); + + Thread.sleep(Integer.MAX_VALUE); + + + } +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/delete/DeleteNodeTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/delete/DeleteNodeTest.java new file mode 100644 index 00000000..beed07f4 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/delete/DeleteNodeTest.java @@ -0,0 +1,34 @@ +package com.zfoo.net.zookeeper.curator.delete; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.RetryUntilElapsed; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore +public class DeleteNodeTest { + + @Test + public void test() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + client.start(); + + // guaranteed() 持续删除，直到成功 + client.delete().guaranteed().deletingChildrenIfNeeded().withVersion(-1).forPath("/node_test"); + + Thread.sleep(Integer.MAX_VALUE); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/query/GetChildrenTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/query/GetChildrenTest.java new file mode 100644 index 00000000..52fa1c3c --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/query/GetChildrenTest.java @@ -0,0 +1,35 @@ +package com.zfoo.net.zookeeper.curator.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.RetryUntilElapsed; +import org.junit.Ignore; +import org.junit.Test; + +import java.util.List; + +@Ignore +public class GetChildrenTest { + + @Test + public void test() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + client.start(); + + List cList = client.getChildren().forPath("/node_test"); + + System.out.println(cList.toString()); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/query/GetDataTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/query/GetDataTest.java new file mode 100644 index 00000000..ffbca0f5 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/query/GetDataTest.java @@ -0,0 +1,38 @@ +package com.zfoo.net.zookeeper.curator.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.RetryUntilElapsed; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore +public class GetDataTest { + + @Test + public void test() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + client.start(); + + Stat stat = new Stat(); + + byte[] ret = client.getData().storingStatIn(stat).forPath("/node_test"); + + System.out.println(new String(ret)); + + System.out.println(stat); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/query/NodeExistsTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/query/NodeExistsTest.java new file mode 100644 index 00000000..40a23672 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/query/NodeExistsTest.java @@ -0,0 +1,70 @@ +package com.zfoo.net.zookeeper.curator.query; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.api.BackgroundCallback; +import org.apache.curator.framework.api.CuratorEvent; +import org.apache.curator.retry.RetryUntilElapsed; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +@Ignore +public class NodeExistsTest { + + @Test + public void test() throws Exception { + ExecutorService es = Executors.newFixedThreadPool(5); + + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + client.start(); + + client.checkExists().inBackground(new BackgroundCallback() { + + @Override + public void processResult(CuratorFramework arg0, CuratorEvent arg1) { + Stat stat = arg1.getStat(); + System.out.println(stat); + System.out.println(arg1.getContext()); + } + }, "Hello Zookeeper!", es).forPath("/node_test"); + + Thread.sleep(Integer.MAX_VALUE); + } + + + @Test + public void testNodeExist() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + client.start(); + + // 如果没有这个节点，则stat返回空 + Stat stat = client.checkExists().forPath("/zfoo/provider"); + + Thread.sleep(Integer.MAX_VALUE); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/curator/update/UpdateDataTest.java b/net/src/test/java/com/zfoo/net/zookeeper/curator/update/UpdateDataTest.java new file mode 100644 index 00000000..302223dd --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/curator/update/UpdateDataTest.java @@ -0,0 +1,35 @@ +package com.zfoo.net.zookeeper.curator.update; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.RetryUntilElapsed; +import org.apache.zookeeper.data.Stat; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore +public class UpdateDataTest { + + @Test + public void test() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + client.start(); + + Stat stat = new Stat(); + client.getData().storingStatIn(stat).forPath("/node_test"); + + client.setData().withVersion(stat.getVersion()).forPath("/node_test", "Hello Zookeeper!".getBytes()); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/recipes/atomicint/DistributedAtomicIntTest.java b/net/src/test/java/com/zfoo/net/zookeeper/recipes/atomicint/DistributedAtomicIntTest.java new file mode 100644 index 00000000..03b24fb1 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/recipes/atomicint/DistributedAtomicIntTest.java @@ -0,0 +1,43 @@ +package com.zfoo.net.zookeeper.recipes.atomicint; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.recipes.atomic.AtomicValue; +import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.apache.curator.retry.RetryNTimes; +import org.junit.Ignore; +import org.junit.Test; + +/** + * 使用Curator实现分布式计数器 + * + * @author jaysunxiao + * @version 1.0 + * @since 2018-08-03 16:09 + */ +@Ignore +public class DistributedAtomicIntTest { + + + static String distatomicint_path = "/node_test"; + static CuratorFramework curator = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .build(); + + @Test + public void test() throws Exception { + curator.start(); + + DistributedAtomicInteger atomicInteger = new DistributedAtomicInteger(curator, distatomicint_path, + new RetryNTimes(3, 1000)); + + AtomicValue atomicValue = atomicInteger.add(8); + + System.out.println("Result: " + atomicValue.succeeded()); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedbarrier/DistributedBarrierTest.java b/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedbarrier/DistributedBarrierTest.java new file mode 100644 index 00000000..22ca1e70 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedbarrier/DistributedBarrierTest.java @@ -0,0 +1,53 @@ +package com.zfoo.net.zookeeper.recipes.distributedbarrier; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.recipes.barriers.DistributedBarrier; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.junit.Ignore; +import org.junit.Test; + +/** + * 使用Curator实现分布式Barrier + * + * @author jaysunxiao + * @version 1.0 + * @since 2018-08-03 16:16 + */ +@Ignore +public class DistributedBarrierTest { + + private static String barrier_path = "/curator_recipes_barrier_path"; + private static DistributedBarrier barrier; + + @Test + public void test() throws Exception { + for (int i = 0; i < 5; i++) { + new Thread(new Runnable() { + public void run() { + try { + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .build(); + client.start(); + barrier = new DistributedBarrier(client, barrier_path); + System.out.println(Thread.currentThread().getName() + "号barrier设置"); + barrier.setBarrier(); + barrier.waitOnBarrier(); + System.err.println("启动..."); + + } catch (Exception e) { + throw new RuntimeException(e); + } + } + }).start(); + } + Thread.sleep(2000); + barrier.removeBarrier(); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedbarrier/JavaBarrierTest.java b/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedbarrier/JavaBarrierTest.java new file mode 100644 index 00000000..8c9726a7 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedbarrier/JavaBarrierTest.java @@ -0,0 +1,48 @@ +package com.zfoo.net.zookeeper.recipes.distributedbarrier; + +import org.junit.Ignore; +import org.junit.Test; + +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * @author jaysunxiao + * @version 1.0 + * @since 2018-08-03 16:13 + */ +@Ignore +public class JavaBarrierTest { + + public static CyclicBarrier barrier = new CyclicBarrier(3); + + @Test + public void test() { + ExecutorService executor = Executors.newFixedThreadPool(3); + executor.submit(new Thread(new Runner("1号选手"))); + executor.submit(new Thread(new Runner("2号选手"))); + executor.submit(new Thread(new Runner("3号选手"))); + executor.shutdown(); + } + + static class Runner implements Runnable { + private String name; + + public Runner(String name) { + this.name = name; + } + + public void run() { + System.out.println(name + " 准备好了."); + try { + JavaBarrierTest.barrier.await(); + } catch (Exception e) { + } + System.out.println(name + " 起跑!"); + } + + } + +} + diff --git a/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedlock/DistributedLockTest.java b/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedlock/DistributedLockTest.java new file mode 100644 index 00000000..67fd7f9f --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedlock/DistributedLockTest.java @@ -0,0 +1,62 @@ +package com.zfoo.net.zookeeper.recipes.distributedlock; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.recipes.locks.InterProcessMutex; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.junit.Ignore; +import org.junit.Test; + +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.concurrent.CountDownLatch; + +/** + * 使用Curator实现分布式锁功能 + * + * @author jaysunxiao + * @version 1.0 + * @since 2018-08-03 15:47 + */ +@Ignore +public class DistributedLockTest { + + static String lock_path = "/node"; + static CuratorFramework curator = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .build(); + + @Test + public void test() { + curator.start(); + final InterProcessMutex lock = new InterProcessMutex(curator, lock_path); + final CountDownLatch down = new CountDownLatch(1); + for (int i = 0; i < 30; i++) { + new Thread(new Runnable() { + public void run() { + try { + down.await(); + lock.acquire(); + } catch (Exception e) { + e.printStackTrace(); + } + + SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS"); + String orderNo = sdf.format(new Date()); + System.out.println("生成的订单号是 : " + orderNo); + + try { + lock.release(); + } catch (Exception e) { + e.printStackTrace(); + } + } + }).start(); + } + down.countDown(); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedlock/NodeLockTest.java b/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedlock/NodeLockTest.java new file mode 100644 index 00000000..d7c8caf5 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/recipes/distributedlock/NodeLockTest.java @@ -0,0 +1,38 @@ +package com.zfoo.net.zookeeper.recipes.distributedlock; + +import org.junit.Ignore; +import org.junit.Test; + +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.concurrent.CountDownLatch; + +/** + * @author jaysunxiao + * @version 1.0 + * @since 2018-08-03 15:46 + */ +@Ignore +public class NodeLockTest { + + @Test + public void test() { + final CountDownLatch down = new CountDownLatch(1); + for (int i = 0; i < 10; i++) { + new Thread(new Runnable() { + public void run() { + try { + down.await(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS"); + String orderNo = sdf.format(new Date()); + System.err.println("生成的订单号是 : " + orderNo); + } + }).start(); + } + down.countDown(); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/recipes/mastersel/MasterSelectorTest.java b/net/src/test/java/com/zfoo/net/zookeeper/recipes/mastersel/MasterSelectorTest.java new file mode 100644 index 00000000..f4fbc4fe --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/recipes/mastersel/MasterSelectorTest.java @@ -0,0 +1,45 @@ +package com.zfoo.net.zookeeper.recipes.mastersel; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.recipes.leader.LeaderSelector; +import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.junit.Ignore; +import org.junit.Test; + +/** + * @author jaysunxiao + * @version 1.0 + * @since 2018-08-03 15:32 + */ +@Ignore +public class MasterSelectorTest { + + // master选举的根节点，表面本次master选举都是在该节点下进行的 + private static String master_path = "/curator_recipes_master_path"; + + private static CuratorFramework curator = CuratorFrameworkFactory.builder() + .connectString("domain1.book.zookeeper:2181") + .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build(); + + @Test + public void test() throws InterruptedException { + curator.start(); + + // curator会在成功获取Master权利的时候回调该监听器 + LeaderSelector selector = new LeaderSelector(curator, master_path, new LeaderSelectorListenerAdapter() { + @Override + public void takeLeadership(CuratorFramework client) throws Exception { + System.out.println("成为Master角色"); + Thread.sleep(3000); + System.out.println("完成Master操作，释放Master权利"); + } + }); + + selector.autoRequeue(); + selector.start(); + Thread.sleep(Integer.MAX_VALUE); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/recipes/nodecache/NodeChildrenListenerTest.java b/net/src/test/java/com/zfoo/net/zookeeper/recipes/nodecache/NodeChildrenListenerTest.java new file mode 100644 index 00000000..50028c25 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/recipes/nodecache/NodeChildrenListenerTest.java @@ -0,0 +1,58 @@ +package com.zfoo.net.zookeeper.recipes.nodecache; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.recipes.cache.PathChildrenCache; +import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent; +import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener; +import org.apache.curator.retry.RetryUntilElapsed; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore +public class NodeChildrenListenerTest { + + @Test + public void test() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework client = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + client.start(); + + // PathChildrenCache用于监听指定Zookeeper数据节点的子节点变换的情况。 + // 注意：Zookeeper无法对二级子节点监听 + final PathChildrenCache cache = new PathChildrenCache(client, "/node_test", true); + cache.start(); + cache.getListenable().addListener(new PathChildrenCacheListener() { + + @Override + public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception { + switch (event.getType()) { + case CHILD_ADDED: + System.out.println("CHILD_ADDED:" + event.getData()); + break; + case CHILD_UPDATED: + System.out.println("CHILD_UPDATED:" + event.getData()); + break; + case CHILD_REMOVED: + System.out.println("CHILD_REMOVED:" + event.getData()); + break; + default: + break; + } + } + }); + + Thread.sleep(Integer.MAX_VALUE); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/recipes/nodecache/NodeListenerTest.java b/net/src/test/java/com/zfoo/net/zookeeper/recipes/nodecache/NodeListenerTest.java new file mode 100644 index 00000000..98480da3 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/recipes/nodecache/NodeListenerTest.java @@ -0,0 +1,55 @@ +package com.zfoo.net.zookeeper.recipes.nodecache; + +import com.zfoo.net.zookeeper.ZookeeperConstantTest; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.recipes.cache.NodeCache; +import org.apache.curator.framework.recipes.cache.NodeCacheListener; +import org.apache.curator.retry.RetryUntilElapsed; +import org.junit.Ignore; +import org.junit.Test; + + +/** + * recipe [rec·i·pe || 'resɪpɪ] n. 食谱; 处方; 烹饪法; 制作法 + *

+ * 数据发布/订阅（Publish/Subscribe）系统，一般有两种方式推（Push）模式；客户端定时轮询，发现有改变就拉（Pull）。 + *

+ * Zookeeper采用两者相结合，客户端订阅，节点有变化服务器就推送 + */ +@Ignore +public class NodeListenerTest { + + @Test + public void test() throws Exception { + RetryPolicy retryPolicy = new RetryUntilElapsed(5000, 1000); + + CuratorFramework curator = CuratorFrameworkFactory + .builder() + .connectString(ZookeeperConstantTest.URL) + .sessionTimeoutMs(5000) + .connectionTimeoutMs(5000) + .retryPolicy(retryPolicy) + .build(); + + curator.start(); + + // Cache是Curator中对事件监听的包装，其对事件的监听其实可以看做本地缓存视图和远程Zookeeper视图的对比过程。 + // 同时Recipes能够反复注册监听，从而大大简化了原生API开发的繁琐程度。 + final NodeCache cache = new NodeCache(curator, "/node_test"); + cache.start(); + + // 节点创建，数据改变都能检测到；如果节点被删除则不能检测。 + cache.getListenable().addListener(new NodeCacheListener() { + @Override + public void nodeChanged() throws Exception { + byte[] ret = cache.getCurrentData().getData(); + System.out.println("new data:" + new String(ret)); + } + }); + + Thread.sleep(Integer.MAX_VALUE); + } + +} diff --git a/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/isolation.png b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/isolation.png new file mode 100644 index 00000000..bb72d2bb Binary files /dev/null and b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/isolation.png differ diff --git a/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/load-balance.png b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/load-balance.png new file mode 100644 index 00000000..3dabd9b6 Binary files /dev/null and b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/load-balance.png differ diff --git a/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/transaction.png b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/transaction.png new file mode 100644 index 00000000..9027faa1 Binary files /dev/null and b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/transaction.png differ diff --git a/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-function.png b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-function.png new file mode 100644 index 00000000..e85d44bd Binary files /dev/null and b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-function.png differ diff --git a/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-leader-election.png b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-leader-election.png new file mode 100644 index 00000000..d7b56dc7 Binary files /dev/null and b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-leader-election.png differ diff --git a/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-process.png b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-process.png new file mode 100644 index 00000000..bbb786ea Binary files /dev/null and b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/image/zookeeper-process.png differ diff --git a/net/src/test/java/com/zfoo/net/zookeeper/zhelp/zookeeper-setup.md b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/zookeeper-setup.md new file mode 100644 index 00000000..a1f5775d --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/zookeeper-setup.md @@ -0,0 +1,137 @@ +# Zookeeper命令 + +## 一、连接Zookeeper + +/opt/zookeeper/bin/zkCli.sh -timeout 5000 -server 192.168.238.128:2181 启动zookeeper的客户端 + +## 二、查看 + +- h # 查看zookeeper的所有数据结构 + +- ls / # 列出根目录得所有节点 +- ls2 / ls+stat + +- get /node_1 # 查询/node_1节点的信息 + +- stat /node_1 # 查看/node_1节点的状态 + +``` +cZxid # 创建事务的id，create zookeeper transaction id zookeeper中每一次对数据的操作的是一个事务，会分配一个事务的id +ctime # 节点被创建的时间戳(从1970年开始) +mZxid # Mofify 节点最新一次更新发生时的zxid +mtime # 节点最后修改的时间戳(从1970年开始) +pZxid # 是与该节点的子节点（或该节点）的最近一次创建/删除子节点的事务id，修改子节点的数据内容不算 +cversion # znode子节点变化号，znode子节点修改次数 +dateVersion # znode数据变化号 +aclVersion +ephemeralOwner # ephemeral adj. 朝生暮死的, 短暂的, 短命的 临时节点 如果该节点为ephemeral节点, ephemeralOwner值表示与该节点绑定的session id. 如果该节点不是ephemeral节点, ephemeralOwner值为0 +dataLength +numChildren +``` + +## 三、更新 + +- set /node_1 "set Method" # 更新节点的信息 + +- create -e /node_2 "hello node_2" # 创建一个node_2的临时节点 +- create -s /node_3 "hello node_3" # 创建一个node_3的顺序节点，会自增主键 + +- delete /node_1 # 删除/node_1节点，如果节点下有另外的节点则不能删除 +- rmr /node_1 # 循环删除/node_1 + +- setquota -n 2 /node_1 # -n设置子节点的数目为2，如果超过了，会记录日志在bin/zookeeper.out中；-b，设置数据长度值（包括子节点） +- delquota -n /node_1 # 删除配额 + +## 四、权限控制 + +### 1. auth的授权方式 + +- addauth digest 用户名:密码明文 # 增加一个认证用户 +- create /path +- setAcl /path auth:用户名:密码明文:crwda # 设置权限 +- getAcl /path # 查看Acl设置 + +``` +ZK的节点有5种操作权限： +CREATE、READ、WRITE、DELETE、ADMIN 也就是 增、删、改、查、管理权限，这5种权限简写为crwda(即：每个单词的首字符缩写) +注： +这5种权限中，delete是指对子节点的删除权限，其它4种权限指对自身节点的操作权限 +Zookeeper对权限的控制是znode级别的，不继承即对父节点设置权限，其子节点不继承父节点的权限。 + +身份的认证有4种方式： +world：默认方式，相当于全世界都能访问 +auth：代表已经认证通过的用户(cli中可以通过addauth digest user:pwd 来添加当前上下文中的授权用户) +digest：即用户名:密码这种方式认证，这也是业务系统中最常用的 +ip：使用Ip地址认证 +``` + +### 2. digest授权方式 + +- create /path digest +- setAcl /path digest:jaysunxiao:F+YQy2Gee3i3tusJucURoYZKurE=:crwda # digest:用户名:密码明文:crwda +- addauth digest jaysunxiao:123456 +- get /path + +``` +digest:jaysunxiao:F+YQy2Gee3i3tusJucURoYZKurE=:crwda +setAcl /digest digest:jaysunxiao:123456:crwda +上下两个只能用编码后的字符串设置密码，避免明文操作，增强安全性。 +设置密码需要自己手动去编码，把密码先自己进行sha1编码然后吧结果进行base64编码。 + +``` + +# Zookeeper监控工具 + +- netflix:exhibitor 监控zookeeper，增删改查 +- zabbix + +# Zookeeper安装 + +## 一、Windows下的安装 + +``` +1.下载windows下的zookeeper安装包并解压到任意目录 +2.加入conf目录，重命名zoo_sample.cfg为zoo.cfg +3.编辑zoo.cfg： + dataDir=D:\\Java\\zookeeper-3.4.10\\zookeeper-3.4.10\\data # 存储数据的目录 + dataLogDir=D:\\Java\\zookeeper-3.4.10\\zookeeper-3.4.10\\logs # 日志目录 +``` + +## 二、Linux下的安装 + +``` +将zookeeper-3.4.10.tar.gz下载到/opt目录，这个目录是专门放下载的第三方软件的目录 +tar -xzvf zookeeper-3.4.10.tar.gz 解压 +mv zookeeper-3.4.10 zookeeper 换一个好记一点的目录 +cd zookeeper/conf 进入zookeeper配置文件目录 +cp zoo_sample.cfg zoo.cfg 建立一个配置文件，zoo_sample.cfg是一个demo +vi zoo.cfg 改变dataDir=/var/zookeeper 这个是zookeeper的共享文件放置的目录，快照文件目录 +在最后一行加上，服务器的ip，格式：server.id=host:port:port， +server.1=192.168.238.128:8888:9999 +server.2=192.168.238.129:8888:9999 +server.3=192.168.238.130:8888:9999 + +关闭防火墙 +getenforce #查看SELinux是否开启，建议关闭， +#enforcing：强制模式，代表 SELinux 运作中；disabled：关闭 +vim /etc/selinux/config #设置SELinux的开启和关闭，必须要重新启动 + +firewall-cmd --zone=public --add-port=2181/tcp --permanent +firewall-cmd --zone=public --add-port=8888/tcp --permanent +firewall-cmd --zone=public --add-port=9999/tcp --permanent +firewall-cmd --reload +开放端口 + +cd /var +mkdir zookeeper 在var目录创建一个zookeeper文件夹 +cd zookeeper +vi myid 在文件里写1 + +cd /opt/zookeeper/bin +./zkServer.sh start 启动zookeeper服务器 /opt/zookeeper/bin/zkServer.sh start +./zkServer.sh stop +./zkCli.sh timeout 5000 -server 192.168.238.128：2181 + +yum install telnet +telnet 192.168.238.128 2181 +``` diff --git a/net/src/test/java/com/zfoo/net/zookeeper/zhelp/zookeeper.md b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/zookeeper.md new file mode 100644 index 00000000..839bd1c8 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/zookeeper/zhelp/zookeeper.md @@ -0,0 +1,256 @@ +## 一.分布式系统 + +#### 分布式系统的定义： + +- 分布式系统是一个硬件或软件组分布在不同的网络计算机上，彼此之间仅仅通过消息传递进行通信和协调的系统。 + +#### 分布式系统的特点： + +1. 分布性 + 分布式系统中的多台计算机在空间上随意分布，不同城市，不同机房，不同机柜 +2. 对等性 + 没有主/从之分，没有控制整个系统的主机，也没有被控制的从机，分布式系统所有的计算机节点都是对等的 +3. 并发性 + 银行转账A转B问题 + +#### 分布式系统的各种问题： + +1. 通信异常 + 单机内存访问的延时在纳秒量级（通常10ns），而一次网络通信的延迟在0.1-1ms左右，相当于内存访问延时的100多倍 +2. 网络分区 + 网络由于发生异常，导致分布式系统中部分节点之间的网络延时不断增大，最终导致分布式系统中只有部分节点能保持正常通行，将这个现象称为网络分区。 + 这时会出现小集群，在极端情况下，这些小集群需要完成整个分布式系统所要完成的工作。 +3. 三态 + 分布式系统中每一次的请求与相应，都存在三态，成功，失败，超时。 出现超时现象无法确定当前请求是否被成功处理 +4. 节点故障 + 每个节点都有可能会出现故障，并且每天都在发生 +5. 缺乏全局时钟 + 很难定义两个时间谁先谁后 + +## 二.分布式一致性算法 + +#### 事务（Transaction） + +- ACID + +1. 原子性（Atomicity）[ˌætəˈmɪsɪti] 要么做要么不做 +2. 一致性（Consistency）事务的执行不能破坏数据库的完整性和一致性 +3. 隔离性（Isolation）[ˌaɪsəˈleʃən] 在标准的SQL规范中有4个事务隔离级别 + 未授权读取，授权读取，可重复读取，串行化。 + ![Image text](image/isolation.png) +4. 持久性（Durability）[djʊərə'bɪlətɪ] 事务一旦提交，它对数据库的状态变更就应该是永久性的。 + +#### CAP定理（CAP theorem /ˈθɪərəm/） + +- C：一致性(Consistency)，所有节点在同一时间具有相同的数据 +- A：可用性(Availability)，保证每个请求在一定时间内，不管成功或者失败都有响应 +- P：分区容错性(Partition tolerance)，系统中任意信息的丢失或失败不会影响系统的继续运作，一般指增加删除节点导致网络分区，不同分区依然可用 + +``` +CA： + CA要求在多个节点满足数据的严格一致性，且延迟较低。 + 如果要满足P，则可以容忍数据丢包，因为数据在跨节点传输时有可能会丢包，如果丢包就无法满足一致性。 + 所以CA一般单点集群，传统的数据库系统就是具有CA特征的数据库，所以在可扩展性上不强，水平扩展能力比较差。 + +CP： + 满足一致性，分区容忍性的系统，因为P会导致系统的节点无线延长，也就无法在一定时间内保证可用性。 + 通常性能不是特别高。 + +AP： + 满足可用性，分区容忍性的系统，就要放弃强一致性，保证最终一致性。因为为了保证可用性，有时候节点就会用自己的本地数据，导致数据不一致性。 + +因为是分布式系统，所以应用必然需要部署到不同的节点中，因此必然出现子网络，所以P是分布式系统首先需要满足的。 +系统架构师往往把精力花在如何根据业务特点在C和A之间寻求平衡。 +``` + +#### BASE理论： + +``` +Basically Available（基本可用），运行损失部分可用性，如响应时间加长，或者功能上的损失，如双11服务降级 +Soft state（软状态），允许在不同节点的数据副本之间进行数据同步的过程中存在延时 +Eventually consistency（最终一致性），系统中的副本，在经过一段时间的同步后，最终能够达到一个一致的状态 +三个短语的简写即为BASE，BASE是对CAP中一致性和可用性权衡的结果，核心思想是即使无法做到强一致性（Strong consistency），但每个业务可以根据自身的业务特点，采用适当的方式来使系统达到最终一致性。 +``` + +#### 分布式一致性协议： + +当一个事务需要跨多个分布式节点的时候，为了保持事务处理的ACID特性，就需要引入一个称为“协调者”的组件来统一调度所有分布式节点的执行逻辑，来保证一致性。 + +2PC：Two Phase Commit，二段提交协议，大多数关系型数据库都是采用的这个协议来完成分布式事务的 +![Image text](image/transaction.png) + +3PC：Three Phase Commit，三阶段提交协议 + +#### Paxos：分布式一致性算法 + +Proposer（提议者），Acceptor（决策者），Client（产生议题者），Learner（最终决策学习者） +现在通过一则故事来学习paxos的算法的流程(2阶段提交)，有2个Client(老板，老板之间是竞争关系)和3个Acceptor(政府官员)： +现在需要对一项议题来进行paxos过程，议题是“A项目我要中标！”，这里的“我”指每个带着他的秘书Proposer的Client老板。 +Proposer当然听老板的话了，赶紧带着议题和现金去找Acceptor政府官员。作为政府官员，当然想谁给的钱多就把项目给谁。 +Proposer-1小姐带着现金同时找到了Acceptor-1~Acceptor-3官员，1与2号官员分别收取了10比特币，找到第3号官员时，没想到遭到了3号官员的鄙视，3号官员告诉她，Proposer-2给了11比特币。 +不过没关系，Proposer-1已经得到了1,2两个官员的认可，形成了多数派(如果没有形成多数派，Proposer-1会去银行提款在来找官员们给每人20比特币，这个过程一直重复每次+10比特币，直到多数派的形成)， +满意的找老板复命去了，但是此时Proposer-2保镖找到了1,2号官员，分别给了他们11比特币，1,2号官员的态度立刻转变，都说Proposer-2的老板懂事，这下子Proposer-2放心了，搞定了3个官员， +找老板复命去了，当然这个过程是第一阶段提交，只是官员们初步接受贿赂而已。故事中的比特币是编号，议题是value。 +这个过程保证了在某一时刻，某一个proposer的议题会形成一个多数派进行初步支持。 +===============华丽的分割线，第一阶段结束================ 现在进入第二阶段提交，现在proposer-1小姐使用分身术(多线程并发) +分了3个自己分别去找3位官员，最先找到了1号官员签合同，遭到了1号官员的鄙视， +1号官员告诉他proposer-2先生给了他11比特币，因为上一条规则的性质proposer-1小姐知道proposer-2第一阶段在她之后又形成了多数派(至少有2位官员的赃款被更新了); +此时她赶紧去提款准备重新贿赂这3个官员(重新进入第一阶段)，每人20比特币。刚给1号官员20比特币， 1号官员很高兴初步接受了议题，还没来得及见到2,3号官员的时候， +这时proposer-2先生也使用分身术分别找3位官员(注意这里是proposer-2的第二阶段)，被第1号官员拒绝了告诉他收到了20比特币， +第2,3号官员顺利签了合同，这时2，3号官员记录client-2老板用了11比特币中标，因为形成了多数派，所以最终接受了Client2老板中标这个议题，对于proposer-2先生已经出色的完成了工作； + +这时proposer-1小姐找到了2号官员，官员告诉她合同已经签了，将合同给她看，proposer-1小姐是一个没有什么职业操守的聪明人，觉得跟Client1老板混没什么前途， +所以将自己的议题修改为“Client2老板中标”，并且给了2号官员20比特币，这样形成了一个多数派。顺利的再次进入第二阶段。由于此时没有人竞争了，顺利的找3位官员签合同， +3位官员看到议题与上次一次的合同是一致的，所以最终接受了，形成了多数派，proposer-1小姐跳槽到Client2老板的公司去了。 + +#### ZAB：原子消息广播协议算法 + +ZAB：Zookeeper Atomic Broadcast，原子消息广播协议，Zookeeper分布式一致性的核心算法。包括两部分：崩溃恢复和消息广播（类似二阶段的提交过程）。 +具体一点可以分为三个阶段：发现（Leader选举），同步（Leader和Follower之间的同步），广播（接受客户端新的事务请求）。 +每个节点都可能处在三种状态，looking：leader选举阶段，following：follower服务器和leader保持同步状态，leading：Leader服务器作为主进程领导状态。 + +Paxos和ZAB的本质区别在于两者的设计目标不太一样。前者用于构建一个分布式的一致性状态机系统，而后者用于构建一个高可用的分布式数据主备系统。 + +## 三.Zookeeper基本说明 + +#### Zookeeper的定义： + +- Zookeeper是一个高可用的分布式数据管理与协调框架 + +#### Zookeeper应用范围： + +分布式应用可以基于Zookeeper实现如分布式一致性特性，数据发布/订阅，负载均衡，命名服务，分布式协调/通知,集群管理，Master选举，分布式锁，分布式队列 + +#### Zookeeper特点 + +- 高性能：Zookeeper将全量数据存储在内存中，并直接服务于客户端的所有非事务请求，因此它尤其适用于以读操作为主的应用场景。Zookeeper性能压测在100%读请求的场景下可以达到12-13W的QPS。 +- 顺序访问：对于每一个客户端的请求，Zookeeper都会生成一个全局唯一的递增序号，这个序号反映了所有事务操作的先后顺序，应用程序可以使用这个特性来实现更高层次的同步原语。 +- Leader服务器在产生一个新的事务的时候，会生成一个序列号ZXID，Zookeeper Transaction ID，这个ID有64位，前32位代表事务编号，每个事务加1，后32位代表发给客户端的计数器，每发给客户端计数器加1。 + ![Image text](image/zookeeper-function.png) + +#### Zookeeper的三种选举算法：LeaderElection，AuthFastLeaderElection，FastLeaderElection + +默认的算法是FastLeaderElection： +启动时的选举算法如下： + +1. 目前有5台服务器，每台服务器均没有数据，它们的编号分别是1,2,3,4,5,按编号依次启动，它们的选择举过程如下： +2. 服务器1启动，给自己投票，然后发投票信息，由于其它机器还没有启动所以它收不到反馈信息，服务器1的状态一直属于Looking。 +3. 服务器2启动，给自己投票，同时与之前启动的服务器1交换结果，由于服务器2的编号大所以服务器2胜出，但此时投票数没有大于半数，所以两个服务器的状态依然是LOOKING。 +4. 服务器3启动，给自己投票，同时与之前启动的服务器1,2交换信息，由于服务器3的编号最大所以服务器3胜出，此时投票数正好大于半数，所以服务器3成为领导者，服务器1,2成为小弟。 +5. 服务器4启动，给自己投票，同时与之前启动的服务器1,2,3交换信息，尽管服务器4的编号大，但之前服务器3已经胜出，所以服务器4只能成为小弟。 +6. 服务器5启动，后面的逻辑同服务器4成为小弟。 + +崩溃恢复时的选举算法： +![Image text](image/zookeeper-leader-election.png) + +![Image text](image/zookeeper-process.png) + +## 四.Zookeeper的应用 + +#### 数据发布/订阅 + +- 发布/订阅一般有两种设计模式，Zookeeper采用推拉结合的方式，一旦节点数据发生变化，服务端就会向客户端发送Watcher事件通知，客户端自己拉取想要的数据 + +``` +服务器主动推（Push）：服务器主动将数据更新发送给所有订阅的客户端，这种数据量在短时间可能会很大 +客户端主动拉（Pull）：客户端主动发起请求获取最新的数据，通常采用定时轮询的方式，这种缺点就是空轮询，客户端开销大 +``` + +- 试用的场合 + +``` +数据量比较小 +数据内容在运行时会发生动态变化 +集群中各机器共享，配置一致 +``` + +#### 负载均衡 + +- 可分为硬件负载均衡和软件负载均衡 + +四层通过虚拟IP+端口接收请求，然后再分配到真实的服务器，如四层的负载均衡，就是通过发布三层的IP地址（VIP），然后加四层的端口号，来决定哪些流量需要做负载均衡， +对需要处理的流量进行NAT处理，转发至后台服务器，并记录下这个TCP或者UDP的流量是由哪台服务器处理的，后续这个连接的所有流量都同样转发到同一台服务器处理。 以常见的TCP为例，负载均衡设备在接收到第一个来自客户端的SYN +请求时，即通过上述方式选择一个最佳的服务器，并对报文中目标IP地址进行修改(改为后端服务器IP），直接转发给该服务器。 +TCP的连接建立，即三次握手是客户端和服务器直接建立的，负载均衡设备只是起到一个类似路由器的转发动作。 +在某些部署情况下，为保证服务器回包可以正确返回给负载均衡设备，在转发报文的同时可能还会对报文原来的源地址进行修改。 + +七层通过虚拟的URL或主机名接收请求，然后再分配到真实的服务器，七层的负载均衡，就是在四层的基础上（没有四层是绝对不可能有七层的），再考虑应用层的特征。 +比如同一个Web服务器的负载均衡，除了根据VIP加80端口辨别是否需要处理的流量，还可根据七层的URL、浏览器类别、语言来决定是否要进行负载均衡。 +举个例子，如果你的Web服务器分成两组，一组是中文语言的，一组是英文语言的，那么七层负载均衡就可以当用户来访问你的域名时，自动辨别用户语言，然后选择对应的语言服务器组进行负载均衡处理。 +以常见的TCP为例，负载均衡设备如果要根据真正的应用层内容再选择服务器，只能先代理最终的服务器和客户端建立连接(三次握手)后，才可能接受到客户端发送的真正应用层内容的报文， +然后再根据该报文中的特定字段，再加上负载均衡设备设置的服务器选择方式，决定最终选择的内部服务器。负载均衡设备在这种情况下，更类似于一个代理服务器。 +负载均衡和前端的客户端以及后端的服务器会分别建立TCP连接。所以从这个技术原理上来看，七层负载均衡明显的对负载均衡设备的要求更高，处理七层的能力也必然会低于四层模式的部署方式。 + +负载均衡器通常称为四层交换机或七层交换机。四层交换机主要分析IP层及TCP/UDP层，实现四层流量负载均衡。七层交换机除了支持四层负载均衡以外，还有分析应用层的信息，如HTTP协议URI或Cookie信息。 + + 1、负载均衡分为L4 switch（四层交换），即在OSI第4层工作，就是TCP层啦。此种Load Balance不理解应用协议（如HTTP/FTP/MySQL等等）。例子：LVS，F5。 + 2、另一种叫做L7 switch（七层交换），OSI的最高层，应用层。此时，该Load Balancer能理解应用协议。例子： haproxy，MySQL Proxy。 + +![Image text](image/load-balance.png) + +#### 命名服务 + +- 集群中全局唯一ID的生成，用Java自带的UUID生成的是一长串没有意义的ID，在Zookeeper中创建一个顺序生成节点可以方便的生成全局唯一ID + +#### 分布式协调/通知 + +- 在绝大多数的分布式系统中，系统机器间的通讯无外乎心跳检测，工作进度汇报和系统调度这三种类型 + +1. 心跳检测 + +``` +传统的方案是一段时间A通ping到B，来判断A和B是否相连。 +基于Zookeeper的心跳检测，A可以在Zookeeper注册一个临时节点，B监听这个节点的变化判断是否相连，这样可以极大的解耦合 +``` + +2. 工作进度汇报 + +``` +注册一个临时节点，节点的内容为自己的任务 +``` + +3. 系统调度 + +``` +通过临时节点的数据变化，控制其它监听该节点的系统 +``` + +#### 集群管理 + +- 机器的上下线，机器监控 + +#### Master选举 + +- 集群中所有的节点同事注册一个/master的临时节点，哪个注册成功哪个就是master节点 + +#### 分布式锁 + +- 排它锁 + +``` +获取锁：所有的客户端同事创建一个/lock临时节点，Zookeeper保证所有的客户端只有一个能创建成功，就认为这个客户端获取了锁，其它客户端注册监听，以便释放锁的时候重新获取锁 +释放锁：获取锁的节点宕机，临时节点就会被移除；正常的执行业务逻辑后，主动释放锁，即删除这个临时节点 +``` + +- 共享锁，又称为读锁 + +``` +加上共享锁的对象，只能对锁中的内容进行读取操作，其它的事物也只能对该对象加共享锁 +``` + +#### 分布式队列 + +- FIFO分布式队列 +- Barrier分布式屏障 + + + + + + + + + + + +