Zookeeper集群部署及分布式应用系统服务器上下线动态感知
2018-02-28 本文已影响17人
padluo
Zookeeper简介
Zookeeper是一个分布式协调服务，为用户的分布式应用程序提供协调服务。
- 为别的分布式程序服务
- 本身是一个分布式程序(只要有半数以上节点存活，zk就能正常服务)。
- 通过底层的两个功能提供各种服务，这两个功能分别是：管理(存储、读取)用户程序提交的数据，为用户程序提供数据节点监听服务。
Zookeeper集群部署
上传安装包
rpm -qa | grep lrzsz
yum install -y lrzsz
cd ~
mkdir apps
tar -xzvf zookeeper-3.4.5.tar.gz -C ~/apps
配置文件修改
cp zoo_sample.cfg zoo.cfg
vi zoo.cfg
# dataDir=/tmp/zookeeper
dataDir=/root/zkdata
# 主机名:心跳端口:数据端口
server.1=hadoopmaster:2888:3888
server.2=hadoopslave1:2888:3888
server.3=hadoopslave2:2888:3888
# myid记录到数据文件夹
# hadoopmaster
mkdir -p /root/zkdata
cd /root/zkdata
echo 1 > myid
cat myid
scp -r /root/apps/ root@hadoopslave1:/root
# hadoopslave1
echo 2 > myid
scp -r /root/apps/ root@hadoopslave2:/root
# hadoopslave2
echo 3 > myid
启动，
# 启动
cd /root/apps/zookeeper-3.4.5
bin/zkServer.sh start
# 查看集群状态和主从信息
bin/zkServer.sh status
# 查看进程
jps
export变量作用域，
export A=1，定义的变量，会对自己所在的shell进程及子进程生效。
B=1，定义的变量，只对自己所在的shell进程生效。
在script.sh中定义的变量，在当前登陆的shell进程中，source script.sh时，脚本中定义的变量也会进入当前登陆的进程。
要在父进程shell可见，可source一下定义export变量的脚本文件，让当前shell可见。
Zookeeper集群自动启动脚本，可先配置集群机器间的免密登录。
#!/bin/sh
# startzk.sh
echo "start zkServer..."
for i in hadoopmaster hadoopslave1 hadoopslave2
do
ssh $i "source /etc/profile;/root/apps/zookeeper-3.4.5/bin/zkServer.sh start"
done
分布式应用系统服务器上下线动态感知
客户端能实时洞察到服务器上下线的变化。
- 服务端启动时即去Zookeeper注册信息。
- 客户端启动就去getChildren，获取到当前在线服务器列表，并且注册监听，接收服务节点上下线通知。
- process()重新再去获取服务器列表，并注册监听。
服务端，
package cn.itcast.bigdata.zkdist;
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 DistributedServer {
private static final String connectString = "hadoopmaster:2181,hadoopslave1:2181,hadoopslave2:2181";
private static final int sessionTimeout = 2000;
private static final String ParentNode = "/servers";
private ZooKeeper zk = null;
/**
* 创建到zk的客户端连接
*
* @throws Exception
*/
public void getConnect() throws Exception {
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
// 收到事件通知后的回调函数(我们自己的事件处理逻辑)
System.out.println(event.getType() + "---" + event.getPath());
try {
zk.getChildren("/", true);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
/**
* 向zk集群注册服务器信息
*
* @param hostname
* @throws Exception
*/
public void registerServer(String hostname) throws Exception {
String create = zk.create(ParentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(hostname + " is online.." + create);
}
/**
* 业务功能
*
* @param hostname
* @throws InterruptedException
*/
public void handleBusiness(String hostname) throws InterruptedException {
System.out.println(hostname + " start working.....");
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
// 获取zk连接
DistributedServer server = new DistributedServer();
server.getConnect();
// 利用zk连接注册服务器信息
server.registerServer(args[0]);
// 启动业务功能
server.handleBusiness(args[0]);
}
}
客户端，
package cn.itcast.bigdata.zkdist;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
public class DistributedClient {
private static final String connectString = "hadoopmaster:2181,hadoopslave1:2181,hadoopslave2:2181";
private static final int sessionTimeout = 2000;
private static final String ParentNode = "/servers";
private ZooKeeper zk = null;
// 加volatile的意义的意义何在
private volatile List<String> serverList;
/**
* 创建到zk的客户端连接
*
* @throws Exception
*/
public void getConnect() throws Exception {
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
// 收到事件通知后的回调函数(我们自己的事件处理逻辑)
// System.out.println(event.getType() + "---" +
// event.getPath());
try {
// 重新更新服务器列表，并且注册了监听
getServerList();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
/**
* 获取服务器信息列表
*
* @throws Exception
*/
public void getServerList() throws Exception {
// 获取服务器子节点信息，并且对父节点进行监听
List<String> children = zk.getChildren(ParentNode, true);
// 先创建一个局部的list来存服务器信息
List<String> servers = new ArrayList<String>();
for (String child : children) {
// child只是子节点的节点名，不需要对子节点的数据进行监听，如做负载均衡，可在数据里记录连接数
byte[] data = zk.getData(ParentNode + "/" + child, false, null);
servers.add(new String(data));
}
// 把servers赋值给成员变量serverList，以提供给各业务线程使用
serverList = servers;
// 打印服务器列表
System.out.println(serverList);
}
/**
* 业务功能
*
* @throws InterruptedException
*/
public void handleBusiness() throws InterruptedException {
System.out.println("client start working.....");
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
// 获取zk连接
DistributedClient client = new DistributedClient();
client.getConnect();
// 获取servers的子节点信息(并监听)，从中获取服务器信息列表
client.getServerList();
// 业务线程启动
client.handleBusiness();
}
}
您可能还想看
Hadoop/CDH
Hadoop实战（1）_阿里云搭建Hadoop2.x的伪分布式环境
Hadoop实战（6）_搭建Apache Hadoop的Eclipse开发环境
Hadoop实战（7）_Apache Hadoop安装和配置Hue
Hadoop实战（8）_CDH添加Hive服务及Hive基础
Hadoop实战（10）_Sqoop import与抽取框架封装
微信公众号「数据分析」，分享数据科学家的自我修养，既然遇见，不如一起成长。
数据分析
读者交流电报群
知识星球交流群
