Zookeeper操作工具类

package com.carelink.rpc.registry.client.util;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import com.carelink.rpc.registry.client.ZkClient;
import com.carelink.rpc.registry.client.ZkConfig;
import com.carelink.rpc.registry.client.factory.ZkClientServiceFactory;
import com.google.common.base.Charsets;
import com.google.common.base.Objects;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;

public class RpcRegisteryService {

	public static final int PROCESS = Runtime.getRuntime().availableProcessors();

	private RpcRegisteryService() {
	}

	private static class SingletonHolder {
		static final RpcRegisteryService instance = new RpcRegisteryService();
	}

	public static RpcRegisteryService instance() {
		return SingletonHolder.instance;
	}

	private CuratorFramework zkclient = null;

	public CuratorFramework getZkclient() {
		return zkclient;
	}

	private void setZkclient(CuratorFramework zkclient) {
		this.zkclient = zkclient;
	}

	/**
	 * 连接ZK 创建初始
	 *
	 * @param address
	 *            地址
	 * @param timeout
	 *            超时时间
	 * @param namespace
	 *            命名空间
	 * @param group
	 *            组
	 * @param groupVal
	 *            组节点值
	 * @param node
	 *            节点
	 * @param nodeVal
	 *            节点值
	 */
	public void connectZookeeper(
			String address,
			int timeout,
			String namespace,
			String group,
			String groupVal,
			String node,
			String nodeVal) {
		if (getZkclient() != null) {
			return;
		}
		CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
		builder.connectString(address).connectionTimeoutMs(timeout).sessionTimeoutMs(timeout)
				.retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 2000));
		if (!Strings.isNullOrEmpty(namespace)) {
			builder.namespace(namespace);
		}
		setZkclient(builder.build());
		RpcConnectionStateListener listener = new RpcConnectionStateListener(group, groupVal, node, nodeVal);
		getZkclient().getConnectionStateListenable().addListener(listener);
		getZkclient().start();
		// 注入
		startRegisterServer(group, groupVal, node, nodeVal);

	}

	private void startRegisterServer(String group, String groupVal, String node, String nodeVal) {
		registerGroup(group, groupVal);
		registerNode(group, node, nodeVal);
	}

	public boolean registerGroup(String group, String groupVal) {
		return createNode("/" + group, groupVal, CreateMode.PERSISTENT);// 创建持久的
	}

	public boolean registerNode(String group, String node, String nodeVal) {
		return createNode("/" + group + "/" + node, nodeVal, CreateMode.EPHEMERAL_SEQUENTIAL); // 创建临时的
	}

	// 设置路径更改监听
	public void listenerPathChildren(String group,PathChildrenCacheListener listener) throws Exception {
		ExecutorService pool = Executors.newFixedThreadPool(PROCESS * 2);
		@SuppressWarnings("resource")
		PathChildrenCache childrenCache = new PathChildrenCache(getZkclient(), "/" + group, true);
		childrenCache.start(StartMode.POST_INITIALIZED_EVENT);
		childrenCache.getListenable().addListener(listener, pool);
	}

	/**
	 * 创建节点
	 *
	 * @param nodeName
	 * @param value
	 * @param createMode
	 * @return
	 * @throws Exception
	 */
	public boolean createNode(String nodeName, String value, CreateMode createMode) {
		boolean suc = false;
		if (getZkclient() == null) {
			return suc;
		}
		try {
			Stat stat = getZkclient().checkExists().forPath(nodeName);
			if (stat == null) {
				String opResult = null;
				// 节点判断值为不为空
				if (Strings.isNullOrEmpty(value)) {
					opResult = getZkclient().create().creatingParentsIfNeeded().withMode(createMode).forPath(nodeName);
				} else {
					opResult = getZkclient().create().creatingParentsIfNeeded().withMode(createMode).forPath(nodeName,
							value.getBytes(Charsets.UTF_8));
				}
				suc = Objects.equal(nodeName, opResult);
			}
		} catch (Exception e) {
			System.out.println("create node fail! path: " + nodeName + "    value: " + value + "  CreateMode: "
					+ createMode.name());
			e.printStackTrace();
			return suc;
		}
		return suc;
	}

	public void destory() {
		if(getZkclient()==null){
			return;
		}
		getZkclient().close();
	}

	/**
	 * 删除节点
	 *
	 * @param node
	 * @return
	 */
	public boolean deleteNode(String node) {
		if (getZkclient() == null) {
			return false;
		}
		try {
			Stat stat = getZkclient().checkExists().forPath(node);
			if (stat != null) {
				getZkclient().delete().deletingChildrenIfNeeded().forPath(node);
			}
			return true;
		} catch (Exception e) {
			System.out.println("delete node fail! path: " + node);
			return false;
		}
	}

	/**
	 * 获取指定节点下的子节点路径和值
	 * @param node
	 * @return
	 */
	public Map<String, String> getChildrenDetail(String node) {
		if (getZkclient() == null) {
			return null;
		}
		Map<String, String> map = Maps.newHashMap();
		try {
			GetChildrenBuilder childrenBuilder = getZkclient().getChildren();
			List<String> children = childrenBuilder.forPath(node);
			GetDataBuilder dataBuilder = getZkclient().getData();
			if (children != null) {
				for (String child : children) {
					String propPath = ZKPaths.makePath(node, child);
					map.put(child, new String(dataBuilder.forPath(propPath), Charsets.UTF_8));
				}
			}
		} catch (Exception e) {
			System.out.println("get node chilren list fail! path: " + node);
			return null;
		}
		return map;
	}

	class RpcConnectionStateListener implements ConnectionStateListener{

		private String group;

		@SuppressWarnings("unused")
		private String groupVal;

		private String node;

		private String nodeVal;

		public RpcConnectionStateListener(String group, String groupVal,String node,String nodeVal) {
			this.groupVal = groupVal;
			this.group = group;
			this.node = node;
			this.nodeVal = nodeVal;
		}
		@Override
		public void stateChanged(CuratorFramework cf, ConnectionState state) {
			if(state == ConnectionState.LOST){
				//重新注册
				while(true){
					//只需要注册节点，组已经是持久的//
					if(registerNode(group, node, nodeVal)){
						break;
					}
				}
			}

		}
	}

	//获取本机IP
	public static String getLocalHost(String type){
		InetAddress addr = null;
		try {
			addr = InetAddress.getLocalHost();
			if("address".equals(type)){
				return addr.getHostName().toString();//获得本机名称
			}
			return addr.getHostAddress().toString();//获得本机IP
		} catch (UnknownHostException e) {
			e.printStackTrace();
			return "";
		}
	}

	public static void main(String[] args) throws Exception {
		String address = "192.168.200.34:2181,192.168.200.44:2181,192.168.200.64:2181";
		System.out.println(ZkConfig.forView());
		System.out.println(getLocalHost("ip"));

		ZkClient.startZkRegistery(address,5000);

		int i = 0;
		while(true){
			try {
				System.out.println( ZkClientServiceFactory.getRpcClientServcie(0).getServersByGroup(ZkConfig.getZkServerGroup()).size());
				System.out.println( ZkClientServiceFactory.getRpcClientServcie(0).getServersByGroup(ZkConfig.getZkServerGroup()));
				Thread.sleep(5000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			i++;
			if(i>10000){
				break;
			}
		}
	}

}

package com.carelink.rpc.registry.server.util;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import com.carelink.rpc.registry.server.ZkConfig;
import com.carelink.rpc.registry.server.ZkService;
import com.google.common.base.Charsets;
import com.google.common.base.Objects;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;

public class RpcRegisteryService {
	private RpcRegisteryService() {
	}

	private static class SingletonHolder {
		static final RpcRegisteryService instance = new RpcRegisteryService();
	}

	public static RpcRegisteryService instance() {
		return SingletonHolder.instance;
	}

	private CuratorFramework zkclient = null;

	private CuratorFramework getZkclient() {
		return zkclient;
	}

	private void setZkclient(CuratorFramework zkclient) {
		this.zkclient = zkclient;
	}

	/**
	 * 连接ZK 创建初始
	 *
	 * @param address
	 *            地址
	 * @param timeout
	 *            超时时间
	 * @param namespace
	 *            命名空间
	 * @param group
	 *            组
	 * @param groupVal
	 *            组节点值
	 * @param node
	 *            节点
	 * @param nodeVal
	 *            节点值
	 */
	public void connectZookeeper(
			String address,
			int timeout,
			String namespace,
			String group,
			String groupVal,
			String node,
			String nodeVal) {
		if (getZkclient() != null) {
			return;
		}
		CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
		// ExponentialBackoffRetry:重试指定的次数, 且每一次重试之间停顿的时间逐渐增加.
		// RetryNTimes:指定最大重试次数的重试策略
		// RetryOneTime:仅重试一次
		// RetryUntilElapsed:一直重试直到达到规定的时
		builder.connectString(address).connectionTimeoutMs(timeout).sessionTimeoutMs(timeout)
				.retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 2000));
		if (!Strings.isNullOrEmpty(namespace)) {
			builder.namespace(namespace);
		}
		setZkclient(builder.build());
		RpcConnectionStateListener listener = new RpcConnectionStateListener(group, groupVal, node, nodeVal);
		getZkclient().getConnectionStateListenable().addListener(listener);
		getZkclient().start();
		// 注入服务
		startRegisterServer(group, groupVal, node, nodeVal);

	}

	private void startRegisterServer(String group, String groupVal, String node, String nodeVal) {
		registerGroup(group, groupVal);
		registerNode(group, node, nodeVal);
	}

	public boolean registerGroup(String group, String groupVal) {
		return createNode("/" + group, groupVal, CreateMode.PERSISTENT);// 创建持久的
	}

	public boolean registerNode(String group, String node, String nodeVal) {
		return createNode("/" + group + "/" + node, nodeVal, CreateMode.EPHEMERAL_SEQUENTIAL); // 创建临时的
	}

	/**
	 * 创建节点
	 *
	 * @param nodeName
	 * @param value
	 * @param createMode
	 * @return
	 * @throws Exception
	 */
	public boolean createNode(String nodeName, String value, CreateMode createMode) {
		boolean suc = false;
		if (getZkclient() == null) {
			return suc;
		}
		try {
			Stat stat = getZkclient().checkExists().forPath(nodeName);
			if (stat == null) {
				String opResult = null;
				// 节点判断值为不为空
				if (Strings.isNullOrEmpty(value)) {
					opResult = getZkclient().create().creatingParentsIfNeeded().withMode(createMode).forPath(nodeName);
				} else {
					opResult = getZkclient().create().creatingParentsIfNeeded().withMode(createMode).forPath(nodeName,
							value.getBytes(Charsets.UTF_8));
				}
				suc = Objects.equal(nodeName, opResult);
			}
		} catch (Exception e) {
			System.out.println("create node fail! path: " + nodeName + "    value: " + value + "  CreateMode: "
					+ createMode.name());
			e.printStackTrace();
			return suc;
		}
		return suc;
	}

	public void destory() {
		if(getZkclient()==null){
			return;
		}
		getZkclient().close();
	}

	/**
	 * 删除节点
	 *
	 * @param node
	 * @return
	 */
	public boolean deleteNode(String node) {
		if (getZkclient() == null) {
			return false;
		}
		try {
			Stat stat = getZkclient().checkExists().forPath(node);
			if (stat != null) {
				getZkclient().delete().deletingChildrenIfNeeded().forPath(node);
			}
			return true;
		} catch (Exception e) {
			System.out.println("delete node fail! path: " + node);
			return false;
		}
	}

	/**
	 * 获取指定节点下的子节点路径和值
	 * @param node
	 * @return
	 */
	public Map<String, String> getChildrenDetail(String node) {
		if (getZkclient() == null) {
			return null;
		}
		Map<String, String> map = Maps.newHashMap();
		try {
			GetChildrenBuilder childrenBuilder = getZkclient().getChildren();
			List<String> children = childrenBuilder.forPath(node);
			GetDataBuilder dataBuilder = getZkclient().getData();
			if (children != null) {
				for (String child : children) {
					String propPath = ZKPaths.makePath(node, child);
					map.put(child, new String(dataBuilder.forPath(propPath), Charsets.UTF_8));
				}
			}
		} catch (Exception e) {
			System.out.println("get node chilren list fail! path: " + node);
			return null;
		}
		return map;
	}

	class RpcConnectionStateListener implements ConnectionStateListener{

		private String group;

		@SuppressWarnings("unused")
		private String groupVal;

		private String node;

		private String nodeVal;

		public RpcConnectionStateListener(String group, String groupVal,String node,String nodeVal) {
			this.groupVal = groupVal;
			this.group = group;
			this.node = node;
			this.nodeVal = nodeVal;
		}
		@Override
		public void stateChanged(CuratorFramework cf, ConnectionState state) {
			if(state == ConnectionState.LOST){
				//重新注册服务
				while(true){
					//只需要注册节点，组已经是持久的
					if(registerNode(group, node, nodeVal)){
						break;
					}
				}
			}
		}
	}

	public static String getLocalHost(String type){
		InetAddress addr = null;
		try {
			addr = InetAddress.getLocalHost();
			if("address".equals(type)){
				return addr.getHostName().toString();//获得本机名称
			}
			return addr.getHostAddress().toString();//获得本机IP
		} catch (UnknownHostException e) {
			e.printStackTrace();
			return "";
		}
	}

	public static void main(String[] args) {
		ZkConfig.readConfig();
		System.out.println(ZkConfig.forView());
		System.out.println(getLocalHost("ip"));

		ZkService.startZkRegistery();
		int i = 0;
		while(true){
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			i++;
			if(i>10){
				break;
			}
		}

		ZkService.stopZkRegistery();
	}

}
时间： 2024-10-22 10:40:21

Zookeeper操作工具类的相关文章

FileUtils.java 本地 文件操作工具类

package Http; import java.io.File;import java.io.FileOutputStream;import java.io.FileWriter;import java.io.IOException; /** * * 本地文件操作工具类 *保存文本 *保存图片 * Created by lxj-pc on 2017/6/27. */public class FileUtils { public static void saveText(String cont

Code片段 ： .properties属性文件操作工具类 &amp; JSON工具类

摘要: 原创出处:www.bysocket.com 泥瓦匠BYSocket 希望转载,保留摘要,谢谢! “贵专” — 泥瓦匠 一.java.util.Properties API & 案例 java.util.Properties 是一个属性集合.常见的api有如下: load(InputStream inStream)  从输入流中读取属性 getProperty(String key)  根据key,获取属性值 getOrDefault(Object key, V defaultValue)

[转载]C# FTP操作工具类

本文转载自<C# Ftp操作工具类>,仅对原文格式进行了整理. 介绍了几种FTP操作的函数,供后期编程时查阅. 参考一: using System; using System.Collections.Generic; using System.Text; using System.IO; using System.Net; using System.Globalization; namespace FtpTest1 { public class FtpWeb { string ftpServe

Java IO(文件操作工具类)

FileOperate实现的功能: 1. 返回文件夹中所有文件列表 2. 读取文本文件内容 3. 新建目录 4. 新建多级目录 5. 新建文件 6. 有编码方式的创建文件 7. 删除文件 8. 删除指定文件夹下所有文件 9. 复制单个文件 10. 复制整个文件夹的内容 11. 移动文件 12. 移动目录 13. 建立一个可以追加的bufferedwriter 14. 得到一个bufferedreader Java代码    package utils; import java.io.Buffer

Android工具类之日期操作工具类

/** * 日期操作工具类. */ public class DateUtil { /** * 英文简写如:2016 */ public static String FORMAT_Y = "yyyy"; /** * 英文简写如:12:01 */ public static String FORMAT_HM = "HH:mm"; /** * 英文简写如:1-12 12:01 */ public static String FORMAT_MDHM = "MM-

反射操作工具类

using System; using System.Collections.Generic; using System.Data; using System.Reflection; namespace Framework.Utility { /// <summary> /// 反射操作工具类 /// </summary> public class ReflectionUtil { #region 根据反射机制将dataTable中指定行的数据赋给obj对象 /// <sum

Android设备内存和SD卡操作工具类

package cc.c; import java.io.File; import java.util.List; import android.os.StatFs; import java.io.FileReader; import java.io.IOException; import java.io.BufferedReader; import android.os.Environment; import android.content.Context; import android.ap

文件操作工具类

package utils; import java.io.BufferedInputStream; import java.io.BufferedOutputStream; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.Outpu

拼音操作工具类 - PinyinUtil.java

拼音操作工具类,提供字符串转换成拼音数组.汉字转换成拼音.取汉字的首字母等方法. 源码如下:(点击下载 -PinyinUtil.java.pinyin4j-2.5.0.jar ) 1 import net.sourceforge.pinyin4j.PinyinHelper; 2 import net.sourceforge.pinyin4j.format.HanyuPinyinCaseType; 3 import net.sourceforge.pinyin4j.format.HanyuPiny