09. Zookeeper JavaAPI-常用操作

灰太狼 2022-03-18 02:28 320阅读 0赞

Zookeeper 的JavaAPI对常用的节点操作都进行了封装, 和Shell 提供的命令差不多.

1. 创建节点API

1.1 创建默认权限节点

  1. zooKeeper.create("/node-anyone", "test node anyone".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

1.2 创建auth 权限模式节点

  1. // 添加用户认证信息
  2. zooKeeper.addAuthInfo("digest", "admin:admin".getBytes());
  3. zooKeeper.addAuthInfo("digest", "admin:123456".getBytes());
  5. // 创建节点
  6. zooKeeper.create("/node-auth", "test node auth".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);

1.3 创建digest 模式权限节点

  1. // 权限位使用或运算符拼接
  2. int aclVal = ZooDefs.Perms.CREATE | ZooDefs.Perms.DELETE | ZooDefs.Perms.READ | ZooDefs.Perms.WRITE;
  4. // 生成加密的用户认证信息
  5. String authentication = DigestAuthenticationProvider.generateDigest("admin:admin");
  7. // 创建权限模式
  8. ACL acl = new ACL(aclVal, new Id("digest", authentication));
  10. // 创建节点
  11. zooKeeper.create("/node-digest", "test node auth".getBytes(), Collections.singletonList(acl), CreateMode.PERSISTENT);

1.4 创建短暂有序节点

  1. // 创建节点, 返回节点名称. 因为有序, 所以节点名称和指定名称不同
  2. String nodeName = zooKeeper.create("/node-es", "test node anyone".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
  3. System.out.println("nodeName:" + nodeName);

2. 节点操作API

2.1 获取节点数据

  1. // 用于保存节点信息
  2. Stat stat = new Stat();
  4. // 获取节点数据
  5. byte[] data = zooKeeper.getData("/node-anyone", false, stat);
  6. System.out.println("data:" + new String(data));
  7. System.out.println("stat.version:" + stat.getVersion());

2.2 获取子节点

  1. List<String> children = zooKeeper.getChildren("/", false);

2.3 判断节点是否存在

  1. Stat stat = zooKeeper.exists("/sss", false);

2.4 删除节点

删除节点时, 必须指定节点版本号.

  1. // 获取节点状态
  2. Stat stat = new Stat();
  3. zooKeeper.getData("/test", false, stat);
  5. // 删除节点
  6. zooKeeper.delete("/test", stat.getVersion());

3. 权限操作API

3.1 获取权限

  1. List<ACL> aclList = zooKeeper.getACL("/node-auth", new Stat());

3.2 设置权限

设置权限时, 需要指定节点版本号.

  1. // 创建节点信息对象, 用于保存节点信息
  2. Stat nodeStat = new Stat();
  3. zooKeeper.getData(nodePath, false, nodeStat);
  5. // 更新节点权限
  6. zooKeeper.setACL("/node-auth", ZooDefs.Ids.READ_ACL_UNSAFE, nodeStat.getVersion());

4. 批量操作

zookeeper 提供了multi() 和 transaction()API 来进行批量操作. 而Transaction 对象其实就是将multi 进行了一层封装而已.

  1. //批量操作
  2. List<Op> opList = new ArrayList<>();
  3. opList.add(Op.create("/nodes", "nodes".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  4. opList.add(Op.create("/nodes/A", "aa".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  5. opList.add(Op.create("/nodes/B", "bb".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  6. opList.add(Op.create("/nodes/C", "cc".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  7. // 如果添加此行, 则全部失败
  8. // opList.add(Op.create("/nodes/D/dd", "dd".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  11. List<OpResult> multi = zooKeeper.multi(opList);
  12. for (OpResult opResult : multi) { 
  14. }
  15. multi.stream()
  16.         .map(opResult -> (OpResult.CreateResult) opResult)
  17.         .forEach(result -> System.out.println(result.getPath() + "-> " + result.getType()));

5. 完整测试用例

  1. import org.apache.zookeeper.*;
  2. import org.apache.zookeeper.data.ACL;
  3. import org.apache.zookeeper.data.Id;
  4. import org.apache.zookeeper.data.Stat;
  5. import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
  6. import org.junit.BeforeClass;
  7. import org.junit.Test;
  9. import java.security.NoSuchAlgorithmException;
  10. import java.util.ArrayList;
  11. import java.util.Collections;
  12. import java.util.List;
  14. /** * @Description: 测试同步API * @author: zongf * @date: 2019-02-17 10:20 */
  15. public class TestAPI { 
  17.     // zookeeper 地址, 多个地址以逗号割开
  18.     private static String zkServer = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";
  20.     // 会话超时时间
  21.     private static int sessionTimeout = 3000;
  23.     // zookeeper 客户端连接
  24.     private static ZooKeeper zooKeeper;
  26.     @BeforeClass
  27.     public static void init() throws Exception{ 
  28.         zooKeeper = new ZooKeeper(zkServer, sessionTimeout, watchedEvent ->{ });
  29.         System.out.println("******************* init start ***********************");
  30.         System.out.println("state:" + zooKeeper.getState());
  31.         System.out.println("sessionPwd:" + zooKeeper.getSessionPasswd());
  32.         System.out.println("sessionId:" + zooKeeper.getSessionId());
  33.         System.out.println("timeout:" + zooKeeper.getSessionTimeout());
  35.         System.out.println("******************* init end ***********************");
  36.     }
  39.     // 创建shell默认模式节点, 访问权限为- world:anyone:cdrwa
  40.     @Test
  41.     public void create_world() throws KeeperException, InterruptedException{ 
  42.         // 创建节点
  43.         zooKeeper.create("/node-anyone", "test node anyone".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  44.     }
  46.     // 创建auth权限模式节点, 访问权限为- auth::cdrwa
  47.     @Test
  48.     public void create_auth() throws KeeperException, InterruptedException{ 
  49.         // 添加用户认证信息
  50.         zooKeeper.addAuthInfo("digest", "admin:admin".getBytes());
  51.         zooKeeper.addAuthInfo("digest", "admin:123456".getBytes());
  53.         // 创建节点
  54.         zooKeeper.create("/node-auth", "test node auth".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
  55.     }
  57.     // 创建digest模式节点, 访问权限为: cdrw, 只能admin用户访问
  58.     @Test
  59.     public void cteate_digest() throws KeeperException, InterruptedException, NoSuchAlgorithmException { 
  61.         // 权限位使用或运算符拼接
  62.         int aclVal = ZooDefs.Perms.CREATE | ZooDefs.Perms.DELETE | ZooDefs.Perms.READ | ZooDefs.Perms.WRITE;
  64.         // 生成加密的用户认证信息
  65.         String authentication = DigestAuthenticationProvider.generateDigest("admin:admin");
  67.         // 创建权限模式
  68.         ACL acl = new ACL(aclVal, new Id("digest", authentication));
  70.         // 创建节点
  71.         zooKeeper.create("/node-digest", "test node auth".getBytes(), Collections.singletonList(acl), CreateMode.PERSISTENT);
  72.     }
  74.     // 创建短暂有序节点
  75.     @Test
  76.     public void create_ephemeral_sequential() throws KeeperException, InterruptedException{ 
  77.         // 创建节点, 返回节点名称. 因为有序, 所以节点名称和指定名称不同
  78.         String nodeName = zooKeeper.create("/node-es", "test node anyone".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
  79.         System.out.println("nodeName:" + nodeName);
  81.         // 休眠线程, 否则会话结束短暂节点就销毁了
  82.         Thread.sleep(10000);
  83.     }
  86.     // 异步创建
  87.     @Test
  88.     public void create_ansy() throws KeeperException, InterruptedException { 
  89.         zooKeeper.create("/learn2", "just for learn async".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL, new AsyncCallback.StringCallback() { 
  90.             @Override
  91.             public void processResult(int rc, String path, Object ctx, String name) { 
  92.                 System.out.println("rc:" + rc);
  93.                 System.out.println("path:" + path);
  94.                 System.out.println("name:" + name);
  95.                 System.out.println("ctx:" + ctx);
  96.             }
  97.         }, "回调结果");
  99.         Thread.sleep(2000);
  100.     }
  102.     // 获取节点数据 和 状态
  103.     @Test
  104.     public void test_getData() throws KeeperException, InterruptedException { 
  106.         // 用于保存节点信息
  107.         Stat stat = new Stat();
  109.         // 获取节点数据
  110.         byte[] data = zooKeeper.getData("/node-anyone", false, stat);
  111.         System.out.println("data:" + new String(data));
  112.         System.out.println("stat.version:" + stat.getVersion());
  113.     }
  115.     @Test
  116.     public void test_getChildren() throws KeeperException, InterruptedException { 
  117.         List<String> children = zooKeeper.getChildren("/", false);
  118.         System.out.println(children);
  119.     }
  121.     // 测试节点是否存在
  122.     @Test
  123.     public void test_exsists() throws KeeperException, InterruptedException { 
  124.         Stat stat = zooKeeper.exists("/sss", false);
  125.         System.out.println(stat != null? stat : "/sss节点不存在" );
  126.     }
  128.     // 删除节点: 必须指定版本号, 只能删除空节点. 不支持shell的rmr 操作
  129.     @Test
  130.     public void test_delete() throws KeeperException, InterruptedException { 
  131.         // 创建测试节点
  132.         zooKeeper.create("/test", "test delete".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  134.         // 获取节点状态
  135.         Stat stat = new Stat();
  136.         zooKeeper.getData("/test", false, stat);
  138.         // 删除节点
  139.         zooKeeper.delete("/test", stat.getVersion());
  141.     }
  145.     // 获取权限
  146.     @Test
  147.     public void test_getACL() throws KeeperException, InterruptedException { 
  148.         Stat nodeStat = new Stat();
  149.         List<ACL> aclList = zooKeeper.getACL("/node-auth", nodeStat);
  150.         aclList.forEach(acl  -> System.out.println(acl.getId().toString().trim() + "->" + convertPerms(acl.getPerms())));
  151.     }
  153.     // 更新权限
  154.     @Test
  155.     public void test_setACL() throws KeeperException, InterruptedException { 
  156.         String nodePath = "/node-anyone";
  158.         // 首先认证用户
  159.         zooKeeper.addAuthInfo("digest", "admin:123456".getBytes());
  161.         // 输出节点权限
  162.         List<ACL> aclList = zooKeeper.getACL(nodePath, new Stat());
  163.         aclList.forEach(acl  -> System.out.println(acl.getId().toString().trim() + "->" + convertPerms(acl.getPerms())));
  165.         // 创建节点信息对象, 用于保存节点信息
  166.         Stat nodeStat = new Stat();
  167.         zooKeeper.getData(nodePath, false, nodeStat);
  169.         // 更新节点权限
  170.         zooKeeper.setACL("/node-auth", ZooDefs.Ids.READ_ACL_UNSAFE, nodeStat.getVersion());
  172.         System.out.println("修改后:");
  174.         // 重新输出节点权限
  175.         List<ACL> aclListAfter = zooKeeper.getACL(nodePath, new Stat());
  176.         aclListAfter.forEach(acl  -> System.out.println(acl.getId().toString().trim() + "->" + convertPerms(acl.getPerms())));
  178. // 输出结果
  179. // 'digest,'admin:0uek/hZ/V9fgiM35b0Z2226acMQ=->rwcda
  180. // 'digest,'admin:x1nq8J5GOJVPY6zgzhtTtA9izLc=->rwcda
  181. // 修改后:
  182. // 'world,' anyone -> r
  184.     }
  186.     // 将权限位转换为shell 字母
  187.     public static String convertPerms(int perms) { 
  189.         if (perms == ZooDefs.Perms.ALL) { 
  190.             return "rwcda";
  191.         }
  193.         StringBuffer sb = new StringBuffer();
  195.         if (perms - ZooDefs.Perms.DELETE >= 0) { 
  196.             sb.append("d");
  197.             perms = perms - ZooDefs.Perms.DELETE;
  198.         }
  200.         if (perms - ZooDefs.Perms.CREATE >= 0) { 
  201.             sb.append("c");
  202.             perms = perms - ZooDefs.Perms.CREATE;
  203.         }
  205.         if (perms - ZooDefs.Perms.WRITE >= 0) { 
  206.             sb.append("w");
  207.             perms = perms - ZooDefs.Perms.WRITE;
  208.         }
  210.         if (perms - ZooDefs.Perms.READ >= 0) { 
  211.             sb.append("r");
  212.             perms = perms - ZooDefs.Perms.READ;
  213.         }
  215.         return sb.toString();
  216.     }
  218.     // 批量执行: 批量创建节点, 要么全部成功, 要么全部失败
  219.     @Test
  220.     public void multi() throws KeeperException, InterruptedException { 
  222.         //批量操作
  223.         List<Op> opList = new ArrayList<>();
  224.         opList.add(Op.create("/nodes", "nodes".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  225.         opList.add(Op.create("/nodes/A", "aa".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  226.         opList.add(Op.create("/nodes/B", "bb".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  227.         opList.add(Op.create("/nodes/C", "cc".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  228. // 如果添加此行, 则全部失败
  229. // opList.add(Op.create("/nodes/D/dd", "dd".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
  232.         List<OpResult> multi = zooKeeper.multi(opList);
  233.         for (OpResult opResult : multi) { 
  235.         }
  236.         multi.stream()
  237.                 .map(opResult -> (OpResult.CreateResult) opResult)
  238.                 .forEach(result -> System.out.println(result.getPath() + "-> " + result.getType()));
  240.     }
  242. }

发表评论

表情：
评论列表 （有 0 条评论，320人围观）

还没有评论，来说两句吧...

相关阅读