Zookeeper --- Java API簡單例項
阿新 • • 發佈：2018-12-14
1、簡單使用
import java.util.List; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; import org.junit.Before; import org.junit.Test; public class SimpleZkClient { private static final String connectString = "192.168.153.134:2181," + "192.168.153.135:2181," + "192.168.153.136:2181"; private static final int sessionTimeout = 2000; private ZooKeeper zkClient = null; @Before public void init() throws Exception { zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() { @Override public void process(WatchedEvent event) { // 收到事件通知後的回撥函式 System.out.println(event.getType() + "---" + event.getPath()); //再次執行監聽 try { zkClient.getChildren("/", true); } catch (Exception e) { e.printStackTrace(); } } }); } /** * 資料的增刪改查 */ //建立資料節點到ZK中 @Test public void testCreate() throws Exception{ //引數1：要建立的節點的路徑；引數2：節點的資料，可以是任何型別，但是需要轉成byte；引數3：節點的許可權，一般使用Ids.OPEN_ACL_UNSAFE；引數4：節點的型別 String nodeCreated = zkClient.create("/eclipse", "hello ZK".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println(nodeCreated); } //判斷znode是否存在 @Test public void testExist() throws Exception{ Stat stat = zkClient.exists("/eclipse", false); System.out.println(stat==null?"not exist":"exist"); } //獲取子節點 @Test public void testGet() throws Exception{ List<String> childrens = zkClient.getChildren("/", true); for (String child : childrens) { System.out.println(child); } } //獲取znode的資料 @Test public void getData() throws Exception{ byte[] data = zkClient.getData("/eclipse", false, null); System.out.println(new String(data)); } //獲取znode的資料 @Test public void setData() throws Exception{ zkClient.setData("/eclipse","hello zk".getBytes(),-1); } //刪除znode @Test public void deleteZnode() throws Exception{ //引數2：指定要刪除的版本，-1表示刪除所有版本 zkClient.delete("/eclipse",-1); } }
2、分散式應用系統伺服器上下線動態感知
DistributedServer
import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; public class DistributedServer { private static final String connectString = "192.168.153.134:2181," + "192.168.153.135:2181," + "192.168.153.136:2181"; private static final int sessionTimeout = 2000; private static final String parentNode = "/servers"; private ZooKeeper zk = null; /** * 建立到zk的客戶端連線 * * @throws Exception */ public void getConnect() throws Exception { zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() { @Override public void process(WatchedEvent event) { // 收到事件通知後的回撥函式（應該是我們自己的事件處理邏輯） System.out.println(event.getType() + "---" + event.getPath()); try { zk.getChildren("/", true); } catch (Exception e) { } } }); } /** * 向zk叢集註冊伺服器資訊 * * @param hostname * @throws Exception */ public void registerServer(String hostname) throws Exception { String create = zk.create(parentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println(hostname + " is online.." + create); } /** * 業務功能 * * @throws InterruptedException */ public void handleBussiness(String hostname) throws InterruptedException { System.out.println(hostname + " start working....."); Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { // 獲取zk連線 DistributedServer server = new DistributedServer(); server.getConnect(); // 利用zk連線註冊伺服器資訊 server.registerServer(args[0]); // 啟動業務功能 server.handleBussiness(args[0]); } }
DistributedClient
import java.util.ArrayList; import java.util.List; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; public class DistributedClient { private static final String connectString = "192.168.153.134:2181," + "192.168.153.135:2181," + "192.168.153.136:2181"; private static final int sessionTimeout = 2000; private static final String parentNode = "/servers"; // 注意:加volatile的意義何在？ private volatile List<String> serverList; private ZooKeeper zk = null; /** * 建立到zk的客戶端連線 * * @throws Exception */ public void getConnect() throws Exception { zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() { @Override public void process(WatchedEvent event) { // 收到事件通知後的回撥函式（應該是我們自己的事件處理邏輯） try { //重新更新伺服器列表，並且註冊了監聽 getServerList(); } catch (Exception e) { } } }); } /** * 獲取伺服器資訊列表 * * @throws Exception */ public void getServerList() throws Exception { // 獲取伺服器子節點資訊，並且對父節點進行監聽 List<String> children = zk.getChildren(parentNode, true); // 先建立一個區域性的list來存伺服器資訊 List<String> servers = new ArrayList<String>(); for (String child : children) { // child只是子節點的節點名 byte[] data = zk.getData(parentNode + "/" + child, false, null); servers.add(new String(data)); } // 把servers賦值給成員變數serverList，已提供給各業務執行緒使用 serverList = servers; //列印伺服器列表 System.out.println(serverList); } /** * 業務功能 * * @throws InterruptedException */ public void handleBussiness() throws InterruptedException { System.out.println("client start working....."); Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { // 獲取zk連線 DistributedClient client = new DistributedClient(); client.getConnect(); // 獲取servers的子節點資訊（並監聽），從中獲取伺服器資訊列表 client.getServerList(); // 業務執行緒啟動 client.handleBussiness(); } }
3、分散式共享鎖簡易實現
import java.util.Collections;
import java.util.List;
import java.util.Random;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
public class DistributedClientLock {
// 會話超時
private static final int SESSION_TIMEOUT = 2000;
// zookeeper叢集地址
private static final String hosts = "192.168.153.134:2181," + "192.168.153.135:2181,"
+ "192.168.153.136:2181";
private String groupNode = "locks";
private String subNode = "sub";
private boolean haveLock = false;
private ZooKeeper zk;
// 記錄自己建立的子節點路徑
private volatile String thisPath;
/**
* 連線zookeeper
*/
public void connectZookeeper() throws Exception {
zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {
public void process(WatchedEvent event) {
try {
// 判斷事件型別，此處只處理子節點變化事件
if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {
//獲取子節點，並對父節點進行監聽
List<String> childrenNodes = zk.getChildren("/" + groupNode, true);
String thisNode = thisPath.substring(("/" + groupNode + "/").length());
// 去比較是否自己是最小id
Collections.sort(childrenNodes);
if (childrenNodes.indexOf(thisNode) == 0) {
//訪問共享資源處理業務，並且在處理完成之後刪除鎖
doSomething();
//重新註冊一把新的鎖
thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
// 1、程式一進來就先註冊一把鎖到zk上
thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
// wait一小會，便於觀察
Thread.sleep(new Random().nextInt(1000));
// 從zk的鎖父目錄下，獲取所有子節點，並且註冊對父節點的監聽
List<String> childrenNodes = zk.getChildren("/" + groupNode, true);
//如果爭搶資源的程式就只有自己，則可以直接去訪問共享資源
if (childrenNodes.size() == 1) {
doSomething();
thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
}
}
/**
* 處理業務邏輯，並且在最後釋放鎖
*/
private void doSomething() throws Exception {
try {
System.out.println("gain lock: " + thisPath);
Thread.sleep(2000);
} finally {
System.out.println("finished: " + thisPath);
zk.delete(this.thisPath, -1);
}
}
public static void main(String[] args) throws Exception {
DistributedClientLock dl = new DistributedClientLock();
dl.connectZookeeper();
Thread.sleep(Long.MAX_VALUE);
}
}