基于Zookeeper服务注册和发现
1、项目的目录结构2、基于zk服务注册和发现的架构图 3、服务端(像zk提供服务的访问地址)package cn.zk.distribute;import org.apache.zookeeper.CreateMode;import org.apache.zookeeper.KeeperException;import org.apache.zookeep
·
1、项目的目录结构
2、基于zk服务注册和发现的架构图
3、服务端(像zk提供服务的访问地址)
package cn.zk.distribute;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
public class DistributedSystemServer {
private ZooKeeper zk = null;
private void getZkClient() throws Exception {
// 服务器在需求中并不需要做任何监听
zk = new ZooKeeper(GlobalConstants.zkhosts,
GlobalConstants.sessionTimeout, null);
}
/**
* 向zookeeper中的/servers下创建子节点
*
* @throws InterruptedException
* @throws KeeperException
*/
private void connectZK(String serverName, String port) throws Exception {
// 先创建出父节点
if (zk.exists(GlobalConstants.parentZnodePath, false) == null) {
zk.create(GlobalConstants.parentZnodePath, null,
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
// 连接zk创建znode
zk.create(GlobalConstants.parentZnodePath + "/",
(serverName + ":" + port).getBytes(), Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("server " + serverName + " is online ......");
}
// 服务器的具体业务处理功能
private void handle(String serverName) throws Exception {
System.out.println("server " + serverName
+ " is waiting for task process......");
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
DistributedSystemServer server = new DistributedSystemServer();
// 获取与zookeeper通信的客户端连接
server.getZkClient();
// 一启动就去zookeeper上注册服务器信息,参数1: 服务器的主机名 参数2:服务器的监听端口
server.connectZK(args[0], args[1]);
// 进入业务逻辑处理流程
server.handle(args[0]);
}
}
2、客户端(获取zk提供的服务地址,并调用服务)
package cn.zk.distribute;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
public class DistributedSystemClient {
private volatile List<String> servers = null;
private ZooKeeper zk = null;
// 获取zk连接
private void getZkClient() throws Exception {
// 服务器在需求中并不需要做任何监听
zk = new ZooKeeper(GlobalConstants.zkhosts,
GlobalConstants.sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
if (event.getType() == EventType.None)
return;
try {
// 获取新的服务器列表,重新注册监听
updateServers();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
/**
* 从zk中获取在线服务器信息
*/
public void updateServers() throws Exception {
// 从servers父节点下获取到所有子节点,并注册监听
List<String> children = zk.getChildren(GlobalConstants.parentZnodePath,
true);
ArrayList<String> serverList = new ArrayList<String>();
for (String child : children) {
byte[] data = zk.getData(GlobalConstants.parentZnodePath + "/"
+ child, false, null);
serverList.add(new String(data));
}
// 如果客户端是一个多线程程序,而且各个线程都会竞争访问servers列表,所以,在成员中用volatile修饰了一个servers变量
// 而在更新服务器信息的这个方法中,是用一个临时List变量来进行更新
servers = serverList;
// 将更新之后的服务器列表信息打印在控制台观察一下
for (String server : serverList) {
System.out.println(server);
}
System.out.println("===================");
}
/**
* 业务逻辑
*
* @throws InterruptedException
*/
private void requestService() throws InterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
DistributedSystemClient client = new DistributedSystemClient();
// 先构造一个zk的连接
client.getZkClient();
// 获取服务器列表
client.updateServers();
// 客户端进入业务流程,请求服务器的服务
client.requestService();
}
}
3、用到的常量配置信息
package cn.zk.distribute;
public class GlobalConstants {
// zk服务器列表
public static final String zkhosts = "192.168.2.118:2181";
// 连接的超时时间
public static final int sessionTimeout = 2000;
// 服务在zk下的路径
public static final String parentZnodePath = "/servers";
}
4、将服务端,导出为可以运行的jar文件
jar的运行
[root@localhost Desktop]# java -jar server.jar 192.168.2.11 4567
5、将客户端,导出为可以运行的jar文件
步骤与上面的4相同,可以参照上面的步骤过程。
7、源码下载
链接:http://pan.baidu.com/s/1eRHHGBK 密码:9d90
更多推荐
已为社区贡献4条内容
所有评论(0)