Zookeeper官网Java示例代码解读(一)

2024-08-30 07:20

本文主要是介绍Zookeeper官网Java示例代码解读(一),希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

2024-08-22

1. 基本信息

  • 官网地址:
    https://zookeeper.apache.org/doc/r3.8.4/javaExample.html

  • 示例设计思路

Conventionally, ZooKeeper applications are broken into two units, one which maintains the connection, and the other which monitors data. In this application, the class called the Executor maintains the ZooKeeper connection, and the class called the DataMonitor monitors the data in the ZooKeeper tree. Also, Executor contains the main thread and contains the execution logic. It is responsible for what little user interaction there is, as well as interaction with the executable program you pass in as an argument and which the sample (per the requirements) shuts down and restarts, according to the state of the znode.

  • Demo的功能
    借助Zookeeper实现分布式环境中的配置文件实时更新

2. 环境准备

  • 准备一台虚拟机(也可以在本机启动ZooKeeper)
  • 安装ZooKeeper、JDK
  • 启动ZooKeeper Server
  • 启动客户端,创建znode,用于测试

3. 示例代码

3.1 Executor

package com.agileluo.zookeeperdemo.simple_watch;  /**  * 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 Executor  implements Watcher, Runnable, DataMonitor.DataMonitorListener  
{  String znode;  DataMonitor dm;  ZooKeeper zk;  String filename;  String exec[];  Process child;  static{  System.setProperty("zookeeper.sasl.client", "false");  }  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) {  }  }  }  /**  * DataMonitor.DataMonitorListener 接口方法exists()的实现  * @param data  */  public void exists(byte[] data) {  if (data == null) { //zooKeeper客户端操作(delete /my_test)时触发  if (child != null) {  System.out.println("Killing process");  child.destroy();  try {  child.waitFor();  } catch (InterruptedException e) {  }  }  child = null;  } else {  //zooKeeper客户端操作(set /my_test test_data)时触发  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();  }  }  }  
}

3.2 DataMonitor

package com.agileluo.zookeeperdemo.simple_watch;  /**  * 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 driven        zk.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 changed            switch (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 over  dead = true;  listener.closing(KeeperException.Code.SessionExpired);  break;  }  } else {  if (path != null && path.equals(znode)) {  // Something has changed on the node, let's find out  zk.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 errors  zk.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 handling                e.printStackTrace();  } catch (InterruptedException e) {  return;  }  }  if ((b == null && b != prevData)  || (b != null && !Arrays.equals(prevData, b))) {  listener.exists(b);  prevData = b;  }  }  
}

4. 测试

运行Executor,参数传入: 192.168.206.100:2181 /my_test filename calc

其中192.168.206.100:2181为ZooKeeper的访问串;
/my_test 是预先创建的Znode
filename 是变动的Znode数据写入的文件,只保留最后的数据,
calc 指定执行完成后,此例为打开计算器(因为是在Windows下跑,所以可以有cmd,run,calc可以用来做测试)

5 注意点

5.1 防火墙

查看防火墙的状态
systemctl status firewalld.service

 firewalld.service - firewalld - dynamic firewall daemonLoaded: loaded (/usr/lib/systemd/system/firewalld.service; enabled; vendor preset: enabled)Active: active (running) since Tue 2024-08-27 19:41:00 PDT; 2s agoDocs: man:firewalld(1)Main PID: 2967 (firewalld)Tasks: 2CGroup: /system.slice/firewalld.service└─2967 /usr/bin/python2 -Es /usr/sbin/firewalld --nofork --nopid

关闭/开启VM的防火墙
systemctl stop|start firewalld.service

5.2 关闭SASL安全验证

Executor类中增加代码:

static{  System.setProperty("zookeeper.sasl.client", "false");  
}

这篇关于Zookeeper官网Java示例代码解读(一)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1120157

相关文章

springboot集成easypoi导出word换行处理过程

《springboot集成easypoi导出word换行处理过程》SpringBoot集成Easypoi导出Word时,换行符n失效显示为空格,解决方法包括生成段落或替换模板中n为回车,同时需确... 目录项目场景问题描述解决方案第一种:生成段落的方式第二种:替换模板的情况,换行符替换成回车总结项目场景s

SpringBoot集成redisson实现延时队列教程

《SpringBoot集成redisson实现延时队列教程》文章介绍了使用Redisson实现延迟队列的完整步骤,包括依赖导入、Redis配置、工具类封装、业务枚举定义、执行器实现、Bean创建、消费... 目录1、先给项目导入Redisson依赖2、配置redis3、创建 RedissonConfig 配

SpringBoot中@Value注入静态变量方式

《SpringBoot中@Value注入静态变量方式》SpringBoot中静态变量无法直接用@Value注入,需通过setter方法,@Value(${})从属性文件获取值,@Value(#{})用... 目录项目场景解决方案注解说明1、@Value("${}")使用示例2、@Value("#{}"php

SpringBoot分段处理List集合多线程批量插入数据方式

《SpringBoot分段处理List集合多线程批量插入数据方式》文章介绍如何处理大数据量List批量插入数据库的优化方案:通过拆分List并分配独立线程处理,结合Spring线程池与异步方法提升效率... 目录项目场景解决方案1.实体类2.Mapper3.spring容器注入线程池bejsan对象4.创建

线上Java OOM问题定位与解决方案超详细解析

《线上JavaOOM问题定位与解决方案超详细解析》OOM是JVM抛出的错误,表示内存分配失败,:本文主要介绍线上JavaOOM问题定位与解决方案的相关资料,文中通过代码介绍的非常详细,需要的朋... 目录一、OOM问题核心认知1.1 OOM定义与技术定位1.2 OOM常见类型及技术特征二、OOM问题定位工具

基于 Cursor 开发 Spring Boot 项目详细攻略

《基于Cursor开发SpringBoot项目详细攻略》Cursor是集成GPT4、Claude3.5等LLM的VSCode类AI编程工具,支持SpringBoot项目开发全流程,涵盖环境配... 目录cursor是什么?基于 Cursor 开发 Spring Boot 项目完整指南1. 环境准备2. 创建

C#实现千万数据秒级导入的代码

《C#实现千万数据秒级导入的代码》在实际开发中excel导入很常见,现代社会中很容易遇到大数据处理业务,所以本文我就给大家分享一下千万数据秒级导入怎么实现,文中有详细的代码示例供大家参考,需要的朋友可... 目录前言一、数据存储二、处理逻辑优化前代码处理逻辑优化后的代码总结前言在实际开发中excel导入很

Spring Security简介、使用与最佳实践

《SpringSecurity简介、使用与最佳实践》SpringSecurity是一个能够为基于Spring的企业应用系统提供声明式的安全访问控制解决方案的安全框架,本文给大家介绍SpringSec... 目录一、如何理解 Spring Security?—— 核心思想二、如何在 Java 项目中使用?——

SpringBoot+RustFS 实现文件切片极速上传的实例代码

《SpringBoot+RustFS实现文件切片极速上传的实例代码》本文介绍利用SpringBoot和RustFS构建高性能文件切片上传系统,实现大文件秒传、断点续传和分片上传等功能,具有一定的参考... 目录一、为什么选择 RustFS + SpringBoot?二、环境准备与部署2.1 安装 RustF

springboot中使用okhttp3的小结

《springboot中使用okhttp3的小结》OkHttp3是一个JavaHTTP客户端,可以处理各种请求类型,比如GET、POST、PUT等,并且支持高效的HTTP连接池、请求和响应缓存、以及异... 在 Spring Boot 项目中使用 OkHttp3 进行 HTTP 请求是一个高效且流行的方式。