zookeeper3.4.6 使用研究

我的理解zookeeper是一个分布式服务发现中间件。提供实时的交互和通知功能。

Zookeeper原理

网上有很多这里就不说了，这里引用一篇。
原理篇-排版什么的都挺好 可以查看

Zookeeper主要API用法

这几个方法是最主要的使用api基本

//删除
String create(final String path, byte data[], List acl, CreateMode createMode)

void delete(final String path, int version)

Stat exists(final String path, Watcher watcher)

Stat exists(String path, boolean watch)

Stat setData(final String path, byte data[], int version)

byte[] getData(final String path, Watcher watcher, Stat stat)

List getChildren(final String path, Watcher watcher)

znode创建类型(CreateMode)

znode是节点的意思，说明该节点存货周期

PERSISTENT 持久化节点
PERSISTENT_SEQUENTIAL 顺序自动编号持久化节点，这种节点会根据当前已存在的节点数自动加 1 。 表现形式：

[zk: localhost:2181(CONNECTED) 13] ls /mytest/AAA
[INST0000000002]

EPHEMERAL 临时节点， 客户端session超时这类节点就会被自动删除
EPHEMERAL_SEQUENTIAL 临时自动编号节点

同上

服务发现使用场景

分布式生产者消费者，生产者和消费者场景大家都不陌生，我们关注的一点就是让生产者集中精力去生产，让消费者集中精力去消费，而不用关心谁生产谁消费，而且在生产规模增大的情况下我们可以动态的控制生产人员和消费者数量。 生产者（1）和消费者（*）的这个连接是面对面一对多服务的。

应用说明

服务发现分为发现端和注册端（你的服务提供端）。

注册端： 注册端为你的消费着，在注册之前你需要确保，你的消费服务已经启动。随时可以处理你的消息。
发现端：发现端为你的生产者，你需要获取到消费者的信息然后根据地址上门服务。如果这家不要你的产品，你需要重新获取其他消费者进行服务，直到找到为止。

该服务发现封装了Zookeeper API进行可以直接在工程使用，你也可已经进行修改满足你的功能需求。

程序代码

ServiceFinder.java

package testhbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

/**
* 服务发现
* stag: application / Service
* path: @basePath  / service_entity
*
* <b>版权信息 :</b> 2015<br/>
* <b>功能描述 :</b> <br/>
* <b>版本历史 :</b> <br/>
* @author lei| 2015年2月11日 下午3:31:30 | 创建
*/
public class ServiceFinder {

private String basePath= "/tmp" ;

private ZooKeeper zookeeper=null;

/**
* application service base path
* @param basePath
* @throws IOException
*/
public ServiceFinder(String zookeeperLinks,String basePath) throws IOException {
if( basePath != null){
this.basePath = basePath;
}

zookeeper = new ZooKeeper(zookeeperLinks, 3000, new Watcher() {

@Override
public void process(WatchedEvent event) {
System.out.println(event.getPath()+" - "+event.getState()+" - "+event.getType());
}
});
}

/**
* get all this applacation's services
* @return services name
* @author lei |2015年2月11日 下午3:26:56
* @version 0.1
* @throws InterruptedException
* @throws KeeperException
*/
public List<String> getServices() throws KeeperException, InterruptedException{
return zookeeper.getChildren(basePath, false);
}

/**
* 获取制定服务实例名称列表
* @param serviceName 服务名称
* @return
* @throws KeeperException
* @throws InterruptedException
* @author lei |2015年2月11日 下午5:56:55
* @version 0.1
*/
public List<String> getServicesEntity(String serviceName) throws KeeperException, InterruptedException{
return zookeeper.getChildren(Utils.joinpath(basePath , serviceName), false);
}
/**
* 获取指定服务实例名称数据信息
* @param serviceName
* @param oneName
* @return
* @throws KeeperException
* @throws InterruptedException
* @author lei |2015年2月11日 下午5:57:32
* @version 0.1
*/
public String getOneServiceData(String serviceName,String oneName) throws KeeperException, InterruptedException{
return String.valueOf(zookeeper.getData(Utils.joinpath(basePath , serviceName , oneName), false, null));
}

/**
* 随机抽取服务实例返回
* @param serviceName
* @return  如果没有可用的返回null
* @throws KeeperException
* @throws InterruptedException
* @author lei |2015年2月11日 下午6:04:16
* @version 0.1
*/
public String getRandomServiceData(String serviceName) throws KeeperException, InterruptedException{
List<String> list=getServiceData(serviceName);
if (list.size()>0){
return list.get(new Random().nextInt(list.size()));
}else{
return null;
}
}

/**
* 获取所有指定服务实例名称数据信息列表
* @param serviceName
* @return
* @throws KeeperException
* @throws InterruptedException
* @author lei |2015年2月11日 下午5:57:50
* @version 0.1
*/
public List<String> getServiceData(String serviceName) throws KeeperException, InterruptedException{
List<String> datas=new ArrayList<String>();

for(String sub:zookeeper.getChildren(Utils.joinpath(basePath , serviceName), false)){
datas.add(new String(zookeeper.getData(Utils.joinpath(basePath , serviceName ,  sub), false, null)));
}

return datas;
}

public void close() throws InterruptedException{
zookeeper.close();
}

public static void main(String[] args) {
try {

ServiceFinder f=new ServiceFinder("localhost:2181","/mytest");
System.out.println("----------------");
System.out.println(f.getServices());
System.out.println("----------------");
System.out.println(f.getServicesEntity("AAA"));
System.out.println("----------------");
System.out.println(f.getServiceData("AAA"));
System.out.println("----------------");
System.out.println("random:"+f.getRandomServiceData("AAA"));
f.close();
} catch (IOException | KeeperException | InterruptedException e) {
e.printStackTrace();
}finally{

}

}

}

ServiceRegister.java

package testhbase;

import java.io.IOException;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class ServiceRegister {

private String basePath= "/tmp" ;

private ZooKeeper zookeeper=null;

public ServiceRegister(String zookeeperLinks,String basePath) throws IOException {
if( basePath != null){
this.basePath = basePath;
}

zookeeper = new ZooKeeper(zookeeperLinks, 3000, new Watcher() {

@Override
public void process(WatchedEvent event) {
System.out.println(event.getPath()+" - "+event.getState()+" - "+event.getType());
}
});
}

/**
* 注册一个服务，这个服务会一直维护session，一旦服务停止，那本节点会在zookeeper自动删除
* @param ServiceName
* @param data your data
* @return
* @throws KeeperException
* @throws InterruptedException
* @author lei |2015年2月11日 下午4:05:01
* @version 0.1
*/
public String register(String serviceName,String data) throws KeeperException, InterruptedException{
String path=Utils.joinpath(basePath,serviceName);
if(zookeeper.exists(path, false) == null){
zookeeper.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}

return zookeeper.create(Utils.joinpath(path,"INST"), data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
}

public static void main(String[] args) {
ServiceRegister reg;
try {
reg = new ServiceRegister("localhost:2181","/mytest");
reg.register("AAA", "{'msg':'Json String You Want...'");
Thread.sleep(3000000);
} catch (IOException | KeeperException | InterruptedException e) {
e.printStackTrace();
}

}

}

Utils.java

package testhbase;

public class Utils {

private static final String sep="/";
public static String joinpath(String base, String ... subs){

if (base.endsWith(sep)){
base=base.substring(0,base.length()-1);
}

for(String sub:subs){
if(sub.startsWith(sep)){
sub=sub.substring(1);
}
base=base+sep+sub;
}
return base;
}

}