使用三大核心语言Java、Python、Golang玩转zookeeper golang

本文视频教程 https://www.bilibili.com/vide...
关注公众号学习更多精彩课程
文章图片

使用三大核心语言Java、Python、Golang玩转zookeeper 使用Java原生api操作zookeeper 创建一个maven项目创建一个maven项目
添加项目依赖

org.apache.zookeeper zookeeper 3.4.8

https://mvnrepository.com/

实现代码

package com.duoke360; import org.apache.zookeeper.*; import java.io.IOException; import java.util.concurrent.CountDownLatch; public class Test { public static void main(String[] args) throws IOException, InterruptedException, KeeperException { String connStr = "192.168.18.128:2181"; CountDownLatch countDown = new CountDownLatch(1); Watcher watcher= event -> { if (event.getState() == Watcher.Event.KeeperState.SyncConnected) { System.err.println("eventType:"+event.getType()); if(event.getType()== Watcher.Event.EventType.None){ countDown.countDown(); }else if(event.getType()== Watcher.Event.EventType.NodeCreated){ System.out.println("listen:节点创建"); }else if(event.getType()== Watcher.Event.EventType.NodeChildrenChanged){ System.out.println("listen:子节点修改"); } } }; ZooKeeper zookeeper = new ZooKeeper(connStr, 5000,watcher); countDown.await(); //注册监听,每次都要重新注册，否则监听不到 // 先创建一个根节点root zookeeper.exists("/root/ghz", watcher); // 创建节点 String result = zookeeper.create("/root/ghz", "老郭".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println(result); Thread.sleep(10); // 获取节点 byte[] bs = zookeeper.getData("/root/ghz", true, null); result = new String(bs); System.out.println("创建节点后的数据是:" + result); // 修改节点 zookeeper.setData("/root/ghz", "多课网-老郭".getBytes(), -1); Thread.sleep(10); bs = zookeeper.getData("/root/ghz", true, null); result = new String(bs); System.out.println("修改节点后的数据是:" + result); // 删除节点 zookeeper.delete("/root/ghz", -1); System.out.println("节点删除成功"); } }

运行结果

eventType:None eventType:NodeCreated listen:节点创建 /root/ghz 创建节点后的数据是:老郭 eventType:NodeDataChanged 修改节点后的数据是:多课网-老郭节点删除成功 eventType:NodeDeleted

使用java zkclient库操作zookeeper.md 创建一个maven项目创建一个maven项目
添加依赖

com.101tec zkclient 0.10

实现代码

package com.duoke360; import org.I0Itec.zkclient.IZkDataListener; import org.I0Itec.zkclient.ZkClient; import org.apache.zookeeper.*; import java.util.List; public class Test { public static void main(String[] args) throws InterruptedException { String connStr = "192.168.18.128:2181"; ZkClient zk = new ZkClient(connStr); // 注册【数据】事件 zk.subscribeDataChanges("/root/ghz", new IZkDataListener() {@Override public void handleDataDeleted(String arg0) throws Exception { System.err.println("数据删除:" + arg0); }@Override public void handleDataChange(String arg0, Object arg1) throws Exception { System.err.println("数据修改:" + arg0 + "------" + arg1); } }); zk.subscribeChildChanges("/root", (arg0, arg1) -> { System.err.println("子节点发生变化：" + arg0); arg1.forEach(f -> { System.out.println("content：" + f); }); }); List list = zk.getChildren("/"); list.forEach(e -> { System.out.println(e); }); String res = zk.create("/root/ghz", "多课网-老郭", CreateMode.PERSISTENT); System.out.println("创建节点/root/ghz成功:" + res); zk.writeData("/root/ghz", "多课网-zookeeper"); System.out.println("修改节点/root/ghz数据成功"); res = zk.readData("/root/ghz"); System.out.println("节点数据:" + res); Thread.sleep(1000); zk.delete("/root/ghz"); System.out.println("删除节点/root/ghz成功"); Thread.sleep(1000); System.out.println("------------------------------------------------"); for (int i = 0; i < 10; i++) { zk.create("/root/ghz", "多课网-老郭", CreateMode.PERSISTENT); Thread.sleep(1000); zk.delete("/root/ghz"); Thread.sleep(1000); } } }

运行结果

node20000000002 zookeeper node10000000001 root 创建节点/root/ghz成功:/root/ghz 修改节点/root/ghz数据成功节点数据:多课网-zookeeper content：ghz 数据修改:/root/ghz------多课网-zookeeper 子节点发生变化：/root 数据修改:/root/ghz------多课网-zookeeper 删除节点/root/ghz成功数据删除:/root/ghz 子节点发生变化：/root ------------------------------------------------ 数据修改:/root/ghz------多课网-老郭子节点发生变化：/root content：ghz 数据删除:/root/ghz 子节点发生变化：/root 数据修改:/root/ghz------多课网-老郭子节点发生变化：/root content：ghz 数据删除:/root/ghz 子节点发生变化：/root 数据修改:/root/ghz------多课网-老郭子节点发生变化：/root

使用Java curator库操作zookeeper 创建一个maven项目创建一个maven项目
添加依赖

org.apache.curator curator-recipes 4.2.0

实现代码

package com.duoke360; 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.apache.zookeeper.CreateMode; import org.apache.zookeeper.data.Stat; public class Test { public static void main(String[] args) throws Exception { String connStr = "192.168.18.128:2181"; CuratorFramework cur= CuratorFrameworkFactory.builder() .connectString(connStr) .connectionTimeoutMs(5000) .retryPolicy(new ExponentialBackoffRetry(1000,3)) .build(); cur.start(); //连接//创建监听 PathChildrenCache cache=new PathChildrenCache(cur,"/root",true); cache.start(); cache.rebuild(); cache.getListenable().addListener(new PathChildrenCacheListener() { @Override public void childEvent(CuratorFramework framwork, PathChildrenCacheEvent event) throws Exception { System.err.println("节点发生变化:"+event.getType()); } }); Stat stat=cur.checkExists().forPath("/root/ghz"); if(stat!=null){ System.out.println("/root/ghz 节点存在，直接删除"); cur.delete().forPath("/root/ghz"); }System.in.read(); System.out.println("准备创建 /root/ghz"); cur.create().withMode(CreateMode.PERSISTENT) .forPath("/root/ghz", "多课网-老郭".getBytes()); System.out.println("节点 /root/ghz 创建成功"); Thread.sleep(1000); byte[] bs=cur.getData().forPath("/root/ghz"); System.out.println("数据:"+new String(bs)); Thread.sleep(1000); cur.delete().forPath("/root/ghz"); Thread.sleep(1000); } }

运行结果

test 准备创建 /root/ghz 节点 /root/ghz 创建成功节点发生变化:CHILD_ADDED 数据:多课网-老郭节点发生变化:CHILD_REMOVED

使用Python操作zookeeper 安装

pip install kazoo

连接 ZooKeeper
可通过 KazooClient 类直接连接 ZooKeeper ，支持多个 host ，端口默认 2181。

import json from kazoo.client import KazooClientzk = KazooClient(hosts='192.168.18.128:2181') zk.start()

创建节点
先看下 create() 方法定义

def create(self, path, value=https://www.it610.com/article/b"", acl=None, ephemeral=False, sequence=False, makepath=False): :param path: Path of node. :param value: Initial bytes value of node. :param acl: :class:`~kazoo.security.ACL` list. :param ephemeral: Boolean indicating whether node is ephemeral (tied to this session). :param sequence: Boolean indicating whether path is suffixed with a unique index. :param makepath: Whether the path should be created if it doesn't exist.

我们来解释下这些参数：

path：节点路径
value：节点对应的值，注意值的类型是 bytes
ephemeral：若为 True 则创建一个临时节点，session 中断后自动删除该节点。默认 False
sequence: 若为 True 则在你创建节点名后面增加10位数字（例如：你创建一个 testplatform/test 节点，实际创建的是 testplatform/test0000000003，这串数字是顺序递增的）。默认 False
makepath：若为 False 父节点不存在时抛 NoNodeError。若为 True 父节点不存在则创建父节点。默认 False

实例：

from kazoo.client import KazooClientzk = KazooClient(hosts='192.168.18.128:2181') zk.start() # 创建节点：makepath 设置为 True ，父节点不存在则创建，其他参数不填均为默认 zk.create('/root/ghz',b'',makepath=True) # 操作完后，别忘了关闭zk连接 zk.stop()

查看节点
KazooClient 类用提供 get_children() 和 get() 方法获取子节点和节点对应的值

from kazoo.client import KazooClientzk = KazooClient(hosts='192.168.18.128:2181') zk.start() # 获取某个节点下所有子节点 node = zk.get_children('/root') # 获取某个节点对应的值 value = https://www.it610.com/article/zk.get('/root/ghz') # 操作完后，别忘了关闭zk连接 zk.stop() print(node,value)

更改节点
更改上文创建的 node 值，使用 set() 方法

from kazoo.client import KazooClientzk = KazooClient(hosts='192.168.18.128:2181') zk.start() # 更改节点对应的value zk.set('/root/ghz', b'duoke-ghz') # 获取某个节点对应的值 value = https://www.it610.com/article/zk.get('/root/ghz') zk.stop() print(value)

删除节点
删除上文创建的节点，使用 delete() 方法

from kazoo.client import KazooClientzk = KazooClient(hosts='192.168.18.128:2181') zk.start() # 删除节点对应的value zk.delete('/root/ghz',recursive=False) zk.stop()

参数 recursive：若为 False，当需要删除的节点存在子节点，会抛异常 NotEmptyError 。若为True，则删除此节点以及删除该节点的所有子节点

watches 事件
zookeeper 所有读操作都有设置 watch 选项（get_children() 、get() 和 exists()）。watch 是一个触发器，当检测到 zookeeper 有子节点变动或者节点value发生变动时触发。下面以 get() 方法为例。

from kazoo.client import KazooClientzk = KazooClient(hosts='192.168.18.128:2181') zk.start()def test(event): print('触发事件')if __name__ == "__main__": # 节点必须存在，否则保存 zk.get('/root/ghz', watch=test) print("第一次获取value") zk.set('/root/ghz', b'duoke-ghz') zk.get('/root/ghz', watch=test) zk.set('/root/ghz', b'duoke-ghz2') print("第二次获取value")zk.stop()

运行结果

第一次获取value 触发事件第二次获取value

使用golang操作zookeeper 下载包

go get github.com/samuel/go-zookeeper

连接到Server

package mainimport ( "fmt" "time""github.com/samuel/go-zookeeper/zk" )func conn() *zk.Conn { var hosts = []string{"192.168.18.128:2181"} conn, _, err := zk.Connect(hosts, time.Second*5) defer conn.Close() if err != nil { fmt.Println(err) return nil } else { fmt.Println("连接成功！") return conn } }func main() { conn() }

创建节点

func create() { var conn *zk.Conn = conn() defer conn.Close() var path = "/home" var data = https://www.it610.com/article/[]byte("多课网") var flags int32 = 0 //flags有4种取值： //0:永久，除非手动删除 //zk.FlagEphemeral = 1:短暂，session断开则改节点也被删除 //zk.FlagSequence= 2:会自动在节点后面添加序号 //3:Ephemeral和Sequence，即，短暂且自动添加序号 var acls = zk.WorldACL(zk.PermAll) //控制访问权限模式p, err_create := conn.Create(path, data, flags, acls) if err_create != nil { fmt.Println(err_create) return } fmt.Println("create:", p) }

修改节点

func set() { var conn *zk.Conn = conn() defer conn.Close() var path = "/home" var data = https://www.it610.com/article/[]byte("多课网-老郭") conn.Set(path, data, -1) b, _, _ := conn.Get(path) fmt.Println("数据：" + string(b)) }

删除节点

func del() { var conn *zk.Conn = conn() defer conn.Close() var path = "/home" err := conn.Delete(path, -1) if err != nil { fmt.Println("删除失败！") } else { fmt.Println("删除成功！") } }

watch

func callback(event zk.Event) { fmt.Println("*******************") fmt.Println("path:", event.Path) fmt.Println("type:", event.Type.String()) fmt.Println("state:", event.State.String()) fmt.Println("*******************") }func watch() {var hosts = []string{"192.168.18.128:2181"} option := zk.WithEventCallback(callback)conn, _, err := zk.Connect(hosts, time.Second*5, option) defer conn.Close() if err != nil { fmt.Println(err) return } var path = "/home" _, _, _, err = conn.ExistsW(path) if err != nil { fmt.Println(err) return }// 创建 create(conn)time.Sleep(time.Second * 2)_, _, _, err = conn.ExistsW(path) if err != nil { fmt.Println(err) return } // 删除 del(conn)}

关注公众号学习更多精彩课程 【使用三大核心语言Java、Python、Golang玩转zookeeper】

文章图片