ZooKeeper Java Example

2024-05-25 13:18
文章标签 java zookeeper example

本文主要是介绍ZooKeeper Java Example，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

要求

客户端有四个要求: 　　　　
这需要作为参数: zookpeer的服务地址、znode的名字、将输出写入到一个文件的名称、一个可执行的参数。　　
它与znode获取相关的数据并开始执行。　　
如果znode发生变化,重启客户端重新提取内容和可执行文件。　　
如果znode消失,客户端可进行线程销毁。

程序设计

一般来说,zookpeer应用被分解成两个部分,一个保持连接,另负责监控数据。在这个应用程序中,这个类称为执行者保持zookpeer联系,和另一个类DataMonitor监控树中的数据。此外,Executor包含主线程和包含执行逻辑。它负责小用户交互是什么,以及交互exectuable计划你在作为参数,该示例根据znode状态进行关闭和重新启动。
 

// from the Executor class...public static void main(String[] args) {if (args.length < 4) {System.err.println("USAGE: Executor hostPort znode filename program [args ...]");System.exit(2);}String hostPort = args[0];String znode = args[1];String filename = args[2];String exec[] = new String[args.length - 3];System.arraycopy(args, 3, exec, 0, exec.length);try {new Executor(hostPort, znode, filename, exec).run();} catch (Exception e) {e.printStackTrace();}}public Executor(String hostPort, String znode, String filename,String exec[]) throws KeeperException, IOException {this.filename = filename;this.exec = exec;zk = new ZooKeeper(hostPort, 3000, this);dm = new DataMonitor(zk, znode, null, this);}public void run() {try {synchronized (this) {while (!dm.dead) {wait();}}} catch (InterruptedException e) {}}

回想一下,执行程序的工作是启动和停止我传递的名字。它是zookpeer事件的相应对象。正如你所看到的在上面的代码中,执行者通过引用Zookpeer本的构造函数。它还通过引用DataMonitor DataMonitorListener参数的构造函数。没当程序执行的时候,就实现了这两个接口: 
public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
...

Watcher接口是Zookpeer的Java API中定义的。Zookpeer使用它与容器进行通信。它支持一个方法process()。执行程序在这个例子简单地将这些事件转发到DataMonitor决定如何处理它们。它只是为了说明这一点,按照惯例,遗嘱执行人或一些Executor-like对象“拥有”Zookpeer的连接,(后面详细讨论)。 
    public void process(WatchedEvent event) {dm.process(event);}

DataMonitorListener接口不是Zookpeer API的一部分。它是一个自定义的interface,为这个示例应用程序而设计的。DataMonitor对象通信使用它回到它的容器,这也是执行程序对象。DataMonitorListener界面如下所示: 
public interface DataMonitorListener {/*** The existence status of the node has changed.*/void exists(byte data[]);/*** The ZooKeeper session is no longer valid.* * @param rc* the ZooKeeper reason code*/void closing(int rc);
}

DataMonitor中定义该接口类和执行程序中实现类。当Executor.exists()调用,执行程序决定是否启动或关闭的要求。 　　　
当Executor.closing()调用,执行程序决定是否关闭自己的Zookpeer连接。　　　　
您可能已经猜到,DataMonitor的对象调用这些方法,以应对变化的Zookpeer状态。　　　　
以下是Executor的DataMonitorListener.exists实现()和DataMonitorListener.closing:
 
public void exists( byte[] data ) {if (data == null) {if (child != null) {System.out.println("Killing process");child.destroy();try {child.waitFor();} catch (InterruptedException e) {}}child = null;} else {if (child != null) {System.out.println("Stopping child");child.destroy();try {child.waitFor();} catch (InterruptedException e) {e.printStackTrace();}}try {FileOutputStream fos = new FileOutputStream(filename);fos.write(data);fos.close();} catch (IOException e) {e.printStackTrace();}try {System.out.println("Starting child");child = Runtime.getRuntime().exec(exec);new StreamWriter(child.getInputStream(), System.out);new StreamWriter(child.getErrorStream(), System.err);} catch (IOException e) {e.printStackTrace();}}
}public void closing(int rc) {synchronized (this) {notifyAll();}
}

DataMonitor类

DataMonitor类是Zookpeer的主要逻辑。它主要是异步和事件驱动的。DataMonitor构造函数:
 
public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,DataMonitorListener listener) {this.zk = zk;this.znode = znode;this.chainedWatcher = chainedWatcher;this.listener = listener;// Get things started by checking if the node exists. We are going// to be completely event drivenzk.exists(znode, true, this, null);
}

调用ZooKeeper.exists()检查znode的存在,设置一个坚挺,和通过引用本身(这)作为完成回调对象。在这个意义上,它开始做事了,因为真正的watch被触发。 
Don't confuse the completion callback with the watch callback. The ZooKeeper.exists() completion callback, which happens to be the method StatCallback.processResult() implemented in the DataMonitor object, is invoked when the asynchronous setting of the watch operation (by ZooKeeper.exists()) completes on the server.The triggering of the watch, on the other hand, sends an event to the Executor object, since the Executor registered as the Watcher of the ZooKeeper object.As an aside, you might note that the DataMonitor could also register itself as the Watcher for this particular watch event. This is new to ZooKeeper 3.0.0 (the support of multiple Watchers). In this example, however, DataMonitor does not register as the Watcher.

当ZooKeeper.exists()操作在服务器上完成,Zookpeer API回调客户端: 
public void processResult(int rc, String path, Object ctx, Stat stat) {boolean exists;switch (rc) {case Code.Ok:exists = true;break;case Code.NoNode:exists = false;break;case Code.SessionExpired:case Code.NoAuth:dead = true;listener.closing(rc);return;default:// Retry errorszk.exists(znode, true, this, null);return;}byte b[] = null;if (exists) {try {b = zk.getData(znode, false, null);} catch (KeeperException e) {// We don't need to worry about recovering now. The watch// callbacks will kick off any exception handlinge.printStackTrace();} catch (InterruptedException e) {return;}}     if ((b == null && b != prevData)|| (b != null && !Arrays.equals(prevData, b))) {listener.exists(b);prevData = b;}
}

代码首先检查znode存在的错误代码,致命错误,可恢复错误。如果文件(或znode)存在,它从znode获取数据,如果状态改变,调用执行者的exist()。注意,它没有任何异常处理getData调用,因为 watches等待任何可能导致一个错误:如果节点被删除之前调用ZooKeeper.getData(),观察事件的ZooKeeper.exists()触发回调;如果有一个通信错误,当连接返回后一个连接监听事件触发。　　　　

最后,注意DataMonitor过程观察事件: 
 public void process(WatchedEvent event) {String path = event.getPath();if (event.getType() == Event.EventType.None) {// We are are being told that the state of the// connection has changedswitch (event.getState()) {case SyncConnected:// In this particular example we don't need to do anything// here - watches are automatically re-registered with // server and any watches triggered while the client was // disconnected will be delivered (in order of course)break;case Expired:// It's all overdead = true;listener.closing(KeeperException.Code.SessionExpired);break;}} else {if (path != null && path.equals(znode)) {// Something has changed on the node, let's find outzk.exists(znode, true, this, null);}}if (chainedWatcher != null) {chainedWatcher.process(event);}}

如果客户端Zookpeer库可以重建通信通道(SyncConnected事件)会话过期前Zookpeer(过期事件)的所有会话的监听会自动重新建立。

完整源代码清单

/*** A simple example program to use DataMonitor to start and* stop executables based on a znode. The program watches the* specified znode and saves the data that corresponds to the* znode in the filesystem. It also starts the specified program* with the specified arguments when the znode exists and kills* the program if the znode goes away.*/
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;public class Executorimplements Watcher, Runnable, DataMonitor.DataMonitorListener
{String znode;DataMonitor dm;ZooKeeper zk;String filename;String exec[];Process child;public Executor(String hostPort, String znode, String filename,String exec[]) throws KeeperException, IOException {this.filename = filename;this.exec = exec;zk = new ZooKeeper(hostPort, 3000, this);dm = new DataMonitor(zk, znode, null, this);}/*** @param args*/public static void main(String[] args) {if (args.length < 4) {System.err.println("USAGE: Executor hostPort znode filename program [args ...]");System.exit(2);}String hostPort = args[0];String znode = args[1];String filename = args[2];String exec[] = new String[args.length - 3];System.arraycopy(args, 3, exec, 0, exec.length);try {new Executor(hostPort, znode, filename, exec).run();} catch (Exception e) {e.printStackTrace();}}/**************************************************************************** We do process any events ourselves, we just need to forward them on.** @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.proto.WatcherEvent)*/public void process(WatchedEvent event) {dm.process(event);}public void run() {try {synchronized (this) {while (!dm.dead) {wait();}}} catch (InterruptedException e) {}}public void closing(int rc) {synchronized (this) {notifyAll();}}static class StreamWriter extends Thread {OutputStream os;InputStream is;StreamWriter(InputStream is, OutputStream os) {this.is = is;this.os = os;start();}public void run() {byte b[] = new byte[80];int rc;try {while ((rc = is.read(b)) > 0) {os.write(b, 0, rc);}} catch (IOException e) {}}}public void exists(byte[] data) {if (data == null) {if (child != null) {System.out.println("Killing process");child.destroy();try {child.waitFor();} catch (InterruptedException e) {}}child = null;} else {if (child != null) {System.out.println("Stopping child");child.destroy();try {child.waitFor();} catch (InterruptedException e) {e.printStackTrace();}}try {FileOutputStream fos = new FileOutputStream(filename);fos.write(data);fos.close();} catch (IOException e) {e.printStackTrace();}try {System.out.println("Starting child");child = Runtime.getRuntime().exec(exec);new StreamWriter(child.getInputStream(), System.out);new StreamWriter(child.getErrorStream(), System.err);} catch (IOException e) {e.printStackTrace();}}}
}

/*** A simple class that monitors the data and existence of a ZooKeeper* node. It uses asynchronous ZooKeeper APIs.*/
import java.util.Arrays;import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;public class DataMonitor implements Watcher, StatCallback {ZooKeeper zk;String znode;Watcher chainedWatcher;boolean dead;DataMonitorListener listener;byte prevData[];public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,DataMonitorListener listener) {this.zk = zk;this.znode = znode;this.chainedWatcher = chainedWatcher;this.listener = listener;// Get things started by checking if the node exists. We are going// to be completely event drivenzk.exists(znode, true, this, null);}/*** Other classes use the DataMonitor by implementing this method*/public interface DataMonitorListener {/*** The existence status of the node has changed.*/void exists(byte data[]);/*** The ZooKeeper session is no longer valid.** @param rc*                the ZooKeeper reason code*/void closing(int rc);}public void process(WatchedEvent event) {String path = event.getPath();if (event.getType() == Event.EventType.None) {// We are are being told that the state of the// connection has changedswitch (event.getState()) {case SyncConnected:// In this particular example we don't need to do anything// here - watches are automatically re-registered with // server and any watches triggered while the client was // disconnected will be delivered (in order of course)break;case Expired:// It's all overdead = true;listener.closing(KeeperException.Code.SessionExpired);break;}} else {if (path != null && path.equals(znode)) {// Something has changed on the node, let's find outzk.exists(znode, true, this, null);}}if (chainedWatcher != null) {chainedWatcher.process(event);}}public void processResult(int rc, String path, Object ctx, Stat stat) {boolean exists;switch (rc) {case Code.Ok:exists = true;break;case Code.NoNode:exists = false;break;case Code.SessionExpired:case Code.NoAuth:dead = true;listener.closing(rc);return;default:// Retry errorszk.exists(znode, true, this, null);return;}byte b[] = null;if (exists) {try {b = zk.getData(znode, false, null);} catch (KeeperException e) {// We don't need to worry about recovering now. The watch// callbacks will kick off any exception handlinge.printStackTrace();} catch (InterruptedException e) {return;}}if ((b == null && b != prevData)|| (b != null && !Arrays.equals(prevData, b))) {listener.exists(b);prevData = b;}}
}


这篇关于ZooKeeper Java Example的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1001608。

相关文章

JVM 的类初始化机制

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时，有没有考虑过 JVM 是如何把静态的字节码（byte code）转化为运行时对象的呢，这个问题看似简单，但清楚的同学相信也不会太多，这篇文章首先介绍 JVM 类初始化的机制，然后给出几个易出错的实例来分析，帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段，分别是：loading 、Linking、initialization
阅读更多...
Spring Security 基于表达式的权限控制

Spring Security 基于表达式的权限控制

前言 spring security 3.0已经可以使用spring el表达式来控制授权，允许在表达式中使用复杂的布尔逻辑来控制访问的权限。 常见的表达式 Spring Security可用表达式对象的基类是SecurityExpressionRoot。 表达式描述hasRole([role])用户拥有制定的角色时返回true （Spring security默认会带有ROLE_前缀）,去
阅读更多...
浅析Spring Security认证过程

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程，特意画了如下的类图，包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口，接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti
阅读更多...
Spring Security--Architecture Overview

Spring Security--Architecture Overview

1 核心组件 这一节主要介绍一些在Spring Security中常见且核心的Java类，它们之间的依赖，构建起了整个框架。想要理解整个架构，最起码得对这些类眼熟。 1.1 SecurityContextHolder SecurityContextHolder用于存储安全上下文（security context）的信息。当前操作的用户是谁，该用户是否已经被认证，他拥有哪些角色权限…这些都被保
阅读更多...
Spring Security基于数据库验证流程详解

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明（认真看哦） AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证，则会调用 #attemptAuthentica
阅读更多...
Spring Security 从入门到进阶系列教程

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门（一）：登录与退出》 《Spring-Security-入门（二）：基于数据库验证》 《Spring-Security-入门（三）：密码加密》 《Spring-Security-入门（四）：自定义-Filter》 《Spring-Security-入门（五）：在 Sprin
阅读更多...
Java架构师知识体认识

Java架构师知识体认识

源码分析 常用设计模式 Proxy代理模式Factory工厂模式Singleton单例模式Delegate委派模式Strategy策略模式Prototype原型模式Template模板模式 Spring5 beans 接口实例化代理Bean操作 Context Ioc容器设计原理及高级特性Aop设计原理Factorybean与Beanfactory Transaction 声明式事物
阅读更多...
Zookeeper安装和配置说明

Zookeeper安装和配置说明

一、Zookeeper的搭建方式 Zookeeper安装方式有三种，单机模式和集群模式以及伪集群模式。 ■ 单机模式：Zookeeper只运行在一台服务器上，适合测试环境； ■ 伪集群模式：就是在一台物理机上运行多个Zookeeper 实例； ■ 集群模式：Zookeeper运行于一个集群上，适合生产环境，这个计算机集群被称为一个“集合体”（ensemble） Zookeeper通过复制来实现
阅读更多...
Java进阶13讲__第12讲_1/2

Java进阶13讲__第12讲_1/2

多线程、线程池 1.  线程概念 1.1  什么是线程 1.2  线程的好处 2.   创建线程的三种方式 注意事项 2.1  继承Thread类 2.1.1 认识  2.1.2  编码实现  package cn.hdc.oop10.Thread;import org.slf4j.Logger;import org.slf4j.LoggerFactory
阅读更多...
JAVA智听未来一站式有声阅读平台听书系统小程序源码

JAVA智听未来一站式有声阅读平台听书系统小程序源码

智听未来，一站式有声阅读平台听书系统 🌟&nbsp;开篇：遇见未来，从“智听”开始 在这个快节奏的时代，你是否渴望在忙碌的间隙，找到一片属于自己的宁静角落？是否梦想着能随时随地，沉浸在知识的海洋，或是故事的奇幻世界里？今天，就让我带你一起探索“智听未来”——这一站式有声阅读平台听书系统，它正悄悄改变着我们的阅读方式，让未来触手可及！ 📚&nbsp;第一站：海量资源，应有尽有 走进“智听
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2