Hadoop学习（三）——高可用性web架构
本文高可用性的架构主要是对三个常用的组件进行介绍，组合使用的过程后面会详细介绍，本文以了解入门为主：
本文的高可用性架构主要是指nginx、keepalived和zookeeper三种的应用。
1、nginx
1.1. nginx简介
Nginx是一个自由、开源、高性能及轻量级的HTTP服务器及反转代理服务器。Nginx以其高性能、稳定、功能丰富、配置简单及占用系统资源少而著称。
Nginx 超越 Apache 的高性能和稳定性，使得国内使用 Nginx 作为 Web 服务器的网站也越来越多.
1.2. 基础功能
反向代理加速，简单的负载均衡和容错；
1.3. 优势
1、Nginx专为性能优化而开发，性能是其最重要的考量, 实现上非常注重效率 。有报告表明能支持高达 50,000 个并发连接数。
2、Nginx具有很高的稳定性。其它HTTP服务器，当遇到访问的峰值，或者有人恶意发起慢速连接时，也很可能会导致服务器物理内存耗尽频繁交换，失去响应，只能重启服务器。
例如当前apache一旦上到200个以上进程，web响应速度就明显非常缓慢了。而Nginx采取了分阶段资源分配技术，使得它的CPU与内存占用率非常低。
3、nginx官方表示保持10,000个没有活动的连接，它只占2.5M内存，就稳定性而言, nginx比其他代理服务器更胜一筹。
4、Nginx支持热部署。它的启动特别容易, 并且几乎可以做到7*24不间断运行，即使运行数个月也不需要重新启动。你还能够在不间断服务的情况下，对软件版本进行进行升级。
5、Nginx采用C进行编写, 不论是系统资源开销还是CPU使用效率都高很多。
1.4. 安装
见安装文档，此处不涉及。
2、keepalived
2.1. 简介
Keepalived的作用是检测web服务器的状态，如果有一台web服务器死机，或工作出现故障，Keepalived将检测到，并将有故障的web服务器从系统中剔除，当web服务器工作正常后Keepalived自动将web服务器加入到服务器群中，这些工作全部自动完成，不需要人工干涉，需要人工做的只是修复故障的web服务器。
两台物理服务器和一个虚拟服务器(vip):
master：redhat2.6.18-53.el5 192.168.8.4
backup: redhat 2.6.18-53.el5 192.168.8.6
vip： 192.168.8.100
测试环境的网络topology图如下:
在这种模式下，虚拟IP在某时刻只能属于某一个节点，另一个节点作为备用节点存在。
当主节点不可用时，备用节点接管虚拟IP（即虚拟IP漂移至节点B），提供正常服务。
keepalived的原理可以这样简单理解:
keepalived安装在两台物理服务器上，并相互监控对方是否在正常运行。
当节点A正常的时候:节点A上的keepalived会将下面的信息广播出去:
192.168.8.100 这个IP对应的MAC地址为节点A网卡的MAC地址
图中的其它电脑如客户端和NodeB会更新自己的ARP表，对应192.168.8.100的MAC地址=节点A网卡的MAC地址。
当节点A发生故障的时候，节点B上的keepalived会检测到，并且将下面的信息广播出去:
192.168.8.100 这个IP对应的MAC地址为节点B网卡的MAC地址
图中的其它电脑如客户端会更新自己的ARP表，对应192.168.8.100的MAC地址=节点B网卡的MAC地址。
主节点的配置如下:
[html] view plain copy
1. global_defs {
2. router_id NodeA
3. }
4. vrrp_instance VI_1 {
5. state MASTER #设置为主服务器
6. interface eth0 #监测网络接口
7. virtual_router_id 51 #主、备必须一样
8. priority 100 #(主、备机取不同的优先级，主机值较大，备份机值较小,值越大优先级越高)
9. advert_int 1 #VRRP Multicast广播周期秒数
10. authentication {
11. auth_type PASS #VRRP认证方式，主备必须一致
12. auth_pass 1111 #(密码)
13. }
14. virtual_ipaddress {
15. 192.168.8.100/24 #VRRP HA虚拟地址
16. }
备用节点的配置如下:
[html] view plain copy
1. global_defs {
2. router_id NodeB
3. }
4. vrrp_instance VI_1 {
5. state BACKUP #设置为主服务器
6. interface eth0 #监测网络接口
7. virtual_router_id 51 #主、备必须一样
8. priority 90 #(主、备机取不同的优先级，主机值较大，备份机值较小,值越大优先级越高)
9. advert_int 1 #VRRP Multicast广播周期秒数
10. authentication {
11. auth_type PASS #VRRP认证方式，主备必须一致
12. auth_pass 1111 #(密码)
13. }
14. virtual_ipaddress {
15. 192.168.8.100/24 #VRRP HA虚拟地址
16. }
2.2. 作用
主要用作RealServer的健康状态检查以及LoadBalance主机和BackUP主机之间failover的实现。
3、Zookeeper：
zk的负载均衡是可以调控，nginx只是能调权重，其他需要可控的都需要自己写插件，nginx最主要的是反向代理，而zk最主要的工作是做分布式协调。
Nginx是著名的反向代理服务器，也被广泛的作为负载均衡服务器
ZooKeeper是分布式协调服务框架，有时也被用来做负载均衡
那么他们的区别是什么？如何选择呢？
下面从实际场景看下他们的关系
Nginx的负载均衡配置非常简单，把多个web server配置到nginx中，用户访问Nginx时，就会自动被分配到某个web server
upstream backend {
server 192.168.1.10;
server 192.168.1.11;}
当网站规模变大，通常会进行服务拆分，各个服务独立部署，通过远程调用方式协同工作。
为了保证稳定性，每个服务不会只使用一台服务器，也会作为一个集群存在，那么这个子集群同样需要一个负载均衡器，可以使用Nginx。
到这里还是没有感觉有使用ZooKeeper的必要，因为使用Nginx完全没问题。
但随着整个系统的演进， 服务 的数量会 增加 、每个服务集群中的 服务器 数量会增加。
这时就会有一些小麻烦，例如：
（1）配置维护的成本变高，因为节点太多
（2）单点故障的风险增加了，因为热点服务的访问量很高，如果这个服务集群内的负载均衡服务出现问题，这个服务将失效
第一个问题，可以通过自己开发程序解决，但只是降低复杂度，并没有实际解决
第二个问题，可以通过双机高可用部署方案，使用另一台nginx负载均衡服务器随时待命，只是成本较高
为了解决这些问题，就有人提出了使用ZooKeeper负载均衡的方案，之前就看到淘宝介绍过此类方案
ZooKeeper负载均衡的实现思路：
把ZooKeeper作为一个服务的注册中心，在其中登记每个服务，每台服务器知道自己是属于哪个服务，在服务器启动时，自己向所属服务进行登记，这样，一个树形的服务结构就呈现出来了。
服务的调用者到注册中心里面查找：能提供所需服务的服务器列表，然后自己根据负载均衡算法，从中选取一台服务器进行连接
调用者取到服务器列表后，就可以缓存到自己内部，省得下次再取，当服务器列表发生变化，例如某台服务器宕机下线，或者新加了服务器，ZooKeeper会自动通知调用者重新获取服务器列表
由于ZooKeeper并没有内置负载均衡策略，需要调用者自己实现，这个方案只是利用了ZooKeeper的树形数据结构、watcher机制等特性，把ZooKeeper作为服务的注册和变更通知中心，解决了Nginx负载均衡方案带来的问题
3.1. Zookeeper概念简介：
Zookeeper是一个分布式协调服务；就是为用户的分布式应用程序提供协调服务。
上侧为外侧服务器，主要是收集外部的数据，而内侧即为收集数据的部分，内侧的服务器将从外侧服务器上收集数据，以方便内侧使用。
如果内侧有台服务器宕机了，则对应的外侧服务器的数据就会丢失，如果用keepalived的虚拟IP来设置相对比较复杂，此时需要有一个对应的方法，即zookeeper作为第三方，每个服务器启动的时候都会到zookeeper中注册一下，则zookeeper中就会有所有服务器的状态信息，并对服务器进行监控，当其中一台服务器宕机后，第三方会自动选举出一个新的服务器重新运行，一般是最小IP接管。
另外情况是高可用场景，即主从选举场景：
外界服务访问服务器时需要先到zookeeper中确定一个master，然后访问对应的master。
第三种场景，配置管理：
即将共同的配置信息放到zookeeper中，实现配置信息的统一管理。
A、zookeeper是为别的分布式程序服务的
B、Zookeeper本身就是一个分布式程序（只要有半数以上节点存活，zk就能正常服务）
C、Zookeeper所提供的服务涵盖：主从协调、服务器节点动态上下线、统一配置管理、分布式共享锁、统一名称服务……
D、虽然说可以提供各种服务，但是zookeeper在底层其实只提供了两个功能：
管理(存储，读取)用户程序提交的数据；
并为用户程序提供数据节点监听服务；
Zookeeper集群的角色： Leader 和 follower （Observer）
只要集群中有半数以上节点存活，集群就能提供服务
3.2. zookeeper集群机制
半数机制：集群中半数以上机器存活，集群可用。
zookeeper适合装在奇数台机器上！！！
安装
3.2. 安装
3.2.1. 机器部署
安装到3台虚拟机上，zookeeper的安装需要安装奇数台。
在每台服务器上都需要安装zookeeper，zookeeper通过竞选来得到master，其他为follower，一旦产生master，则master需要同步各个follower的信息。
则配置文件中需要配置两样东西，一个是名称，一个是服务器，选举时自发形成的。
[[email protected] bin]#./zkServer.sh status
JMX enabled by default Using config:
/usr/apps/zookeeper/zookeeper-3.4.5/bin/../conf/zoo.cfg
Error contactingservice. It is probably not running.
此时应该重启多台服务器，其下就不会产生error，而是产生mode：leader等。
./zkCli.sh为启动zk的客户端。
需要使用zk的客户端进行使用。
Connect mini2:8080 :连接到mini2服务器的8080端口；
其他命令如下图：
Create /app1 “this isapp1 servers parent”
Get出的是文件中的内容，其他为文件的元数据。
Get /app1 watch :即监听功能，当app1数据发生变化时，服务器自动通知。
Ls /app1 watch:即子节点的监听功能。
Jps：查看zk的进程；
进程命令介绍：https://www.cnblogs.com/senlinyang/p/7833669.html
安装好JDK
解压 jdk-8u131-linux-x64.tar.gz安装包
[[email protected] software]# mkdir -p /usr/lib/jvm
[[email protected] software]# tar -zxvf jdk-8u131-linux-x64.tar.gz -C/usr/lib/jvm
5、设置环境变量
[[email protected] software]# vim /etc/profile
在最前面添加：
exportJAVA_HOME=/usr/lib/jvm/jdk1.8.0_131
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH
6、执行profile文件
[[email protected] software]# source /etc/profile
这样可以使配置不用重启即可立即生效。
7、检查新安装的jdk
[[email protected] software]# java -version
显示：
java version "1.8.0_131"
Java(TM) SE Runtime Environment (build1.8.0_131-b11)
Java HotSpot(TM) 64-Bit Server VM (build25.131-b11, mixed mode)
到此为止，整个安装过程结束。
3.2.2. 上传
上传用工具。
3.2.3. 解压
su – hadoop（切换到hadoop用户）
tar -zxvfzookeeper-3.4.5.tar.gz（解压）
3.2.4. 重命名
mv zookeeper-3.4.5 zookeeper（重命名文件夹zookeeper-3.4.5为zookeeper）
3.2.5. 修改环境变量
1、su – root(切换用户到root)
2、vi /etc/profile(修改文件)
3、添加内容：
|
export ZOOKEEPER_HOME=/home/hadoop/zookeeper export PATH=$PATH:$ZOOKEEPER_HOME/bin |
4、重新编译文件：
source /etc/profile
5、注意：3台zookeeper都需要修改
6、修改完成后切换回hadoop用户：
su - hadoop
3.2.6. 修改配置文件
1、用hadoop用户操作
cd zookeeper/conf
cp zoo_sample.cfg zoo.cfg
2、vi zoo.cfg
3、添加内容：
|
dataDir=/home/hadoop/zookeeper/data dataLogDir=/home/hadoop/zookeeper/log server.1=slave1:2888:3888 (主机名, 心跳端口、数据端口) server.2=slave2:2888:3888 server.3=slave3:2888:3888 |
4、创建文件夹：
cd /home/hadoop/zookeeper/
mkdir -m 755 data
mkdir -m 755 log
5、在data文件夹下新建myid文件，myid的文件内容为：
cd data
vi myid
添加内容：
1 |
3.2.7. 将集群下发到其他机器上
scp -r /home/hadoop/[email protected]:/home/hadoop/
scp -r /home/hadoop/[email protected]:/home/hadoop/
3.2.8. 修改其他机器的配置文件
到slave2上：修改myid为：2
到slave3上：修改myid为：3
3.2.9. 启动（每台机器）
./zkServer.sh start
如何远程启动zk：
为什么其中有一个source，首先要了解一下expect的作用域：
.sh的命令会重新启动一个进程来执行其中的命令，如果命令中有赋值操作，则不会显示在当前的界面上；通过ssh远程控制另一台服务器时如果需要写脚本。
如果直接如下图执行，则不会产生预期效果：
因为zk在启动的时候需要java_home，但是ssh会话远程后是没有java_home的，需要手动的添加一个java_home在。
如果在本线程中添加a=1，然后再在s1.sh中执行，此时会报错，执行的结果是没有返回值。
因为在子进程中看不到父进程的环境变量。
此时需要使用export a=1即可。
另外：
Export a=1
s1.sh：
改为：
s2.sh：
此时，s2中的b是打印不出来的，需要在s1中添加exportb=2，s2中就能够打印出b了，但是shell父进程中没有b的值，即export只能影响本进程以及子进程，但是不会影响父进程。
如果需要在父进程中也有效，即需要使用source s1.sh，此时程序会将变量赋值给父进程（即当前运行界面），作用域变大。
那么为什么ssh会话到另一台服务器，为什么对应的.sh命令没有生效。
因为会话的进程，与另一台服务器的shell进程是同级的，对应的sh命令不会在shell进程中执行，需要用到source来使命令在shell进程中执行。
非远程登录时，会默认启动一下对应的环境变量，但是远程时没有，需要手动的将环境变量加入到ssh进程中，需要用到source来使命令在shell进程中执行。
实现免密操作：
3.2.10. 查看集群状态
1、 jps（查看进程）
2、 zkServer.sh status（查看集群状态，主从信息）
4、zookeeper结构和命令
4.1. zookeeper特性
1、Zookeeper：一个leader，多个follower组成的集群
2、全局数据一致：每个server保存一份相同的数据副本，client无论连接到哪个server，数据都是一致的
3、分布式读写，更新请求转发，由leader实施
4、更新请求顺序进行，来自同一个client的更新请求按其发送顺序依次执行
5、数据更新原子性，一次数据更新要么成功，要么失败
6、实时性，在一定时间范围内，client能读到最新数据
4.2. zookeeper数据结构
1、层次化的目录结构，命名符合常规文件系统规范(见下图)
2、每个节点在zookeeper中叫做znode,并且其有一个唯一的路径标识
3、节点Znode可以包含数据和子节点（但是EPHEMERAL类型的节点不能有子节点，下一页详细讲解）
4、客户端应用可以在节点上设置监视器（后续详细讲解）
4.3. 数据结构的图
4.4. 节点类型
1、Znode有两种类型：
短暂（ephemeral）（断开连接自己删除）
Create -e /app1 8888
持久（persistent）（断开连接不删除）
Create -p /app2 8888
3、 Znode有四种形式的目录节点（默认是persistent ）
Create -s /test/aa888: 在父目录下生成的aa会形成一个序号。
PERSISTENT
PERSISTENT_SEQUENTIAL（持久序列/test0000000019 ）
EPHEMERAL
EPHEMERAL_SEQUENTIAL
3、创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护
4、在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序
4.5. zookeeper命令行操作
运行 zkCli.sh –server <ip>进入命令行工具
1、使用 ls 命令来查看当前 ZooKeeper 中所包含的内容：
[zk: 202.115.36.251:2181(CONNECTED) 1] ls /
2、创建一个新的 znode ，使用 create /zk myData 。这个命令创建了一个新的 znode 节点“ zk ”以及与它关联的字符串：
[zk: 202.115.36.251:2181(CONNECTED) 2] create /zk "myData“
3、我们运行 get 命令来确认 znode 是否包含我们所创建的字符串：
[zk: 202.115.36.251:2181(CONNECTED) 3] get /zk
#监听这个节点的变化,当另外一个客户端改变/zk时,它会打出下面的
#WATCHER::
#WatchedEventstate:SyncConnected type:NodeDataChanged path:/zk
[zk:localhost:2181(CONNECTED) 4] get /zk watch
4、下面我们通过 set 命令来对 zk 所关联的字符串进行设置：
[zk: 202.115.36.251:2181(CONNECTED) 4] set /zk "zsl“
5、下面我们将刚才创建的 znode 删除：
[zk: 202.115.36.251:2181(CONNECTED) 5] delete /zk
6、删除节点：rmr
[zk: 202.115.36.251:2181(CONNECTED) 5] rmr /zk
4.6. zookeeper-api应用
4.6.1. 基本使用
org.apache.zookeeper.Zookeeper是客户端入口主类，负责建立与server的会话它提供了表 1 所示几类主要方法 ：
功能 | 描述 |
create | 在本地目录树中创建一个节点 |
delete | 删除一个节点 |
exists | 测试本地是否存在目标节点 |
get/set data | 从目标节点上读取 / 写数据 |
get/set ACL | 获取 / 设置目标节点访问控制列表信息 |
get children | 检索一个子节点上的列表 |
sync | 等待要被传送的数据 |
表1：ZooKeeper API描述（demo增删改查）
package zookeeper.zk;
import java.util.List;
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;
import org.junit.Before;
import org.junit.Test;
publicclass SimpleZkClient {
privatestaticfinal String connectString="192.168.120.135:2181";
privatestaticfinalintsessionTimeout=2000;
ZooKeeper zkClient = null;
@Before
publicvoid init() throws Exception{
zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher(){
// zk发生变化时会执行process方法，用来通知。
@Override
publicvoid process(WatchedEvent event) {
// TODO Auto-generated method stub
System.out.println(event.getType()+"----"+event.getPath());
try {
zkClient.getChildren("/", true);
}catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
publicvoid testCreate() throwsKeeperException, InterruptedException{
/*数据的增删改查：*/
/*参数1：要创建的节点的路径；参数2：节点的大数据；参数3：节点的权限；参数4：节点的类型*/
zkClient.create("/zk2", "hellozk".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
@Test
publicvoid getChildren() throwsKeeperException, InterruptedException{
List<String>children = zkClient.getChildren("/", true);
for(String child:children){
System.out.println(child);
}
Thread.sleep(Long.MAX_VALUE);
}
}
这一代码的原理为：
客户端的main线程（主线程）创建一个对象zkclient，此时程序启动两个线程，一个是客户端，用来与zookeeper进行通信；
另一个线程是listener，他监听一个端口，同时调zkclient中的process方法，zookeeper集群中保留client的ip以及需要监听的/path路径，一旦path发生变化，集群就会向客户端的main方法发送通知，此时listener接收到这一通知，马上执行process方法。
Zk动态感知服务器变化：
需求：客户端能动态感知到服务器上下线的动态变化。
首先：各台服务器一启动就到zookeeper中注册，比如server_01 mini1_2181，server_02 mini2_2181，server_03 mini3_2181以及业务功能的端口。这些节点必须是临时节点，这样当服务器挂掉之后才会发生变化，产生通知，当然也可以添加序号节点，此时当服务器变化时就会产生通知。
其次：对于客户端，启动就去getchildren，获取到当前的在线服务器列表，然后通过策略分析使用哪台服务器。
第三：如果有机器下线了，客户端如何知道？客户端在getchildren时会同时注册监听，此时，如果某一个服务器宕机了，zk中的某一个节点就会消失，客户端就会接收到变化的通知，知道某一台服务器宕机。客户端调用process方法，在方法中可以重新再次获取服务器列表，并再次监听。
上图为主要的逻辑。
此时的程序在服务器和客户端都要加代码，服务端一启动就注册一个数据到zk中，客户端中一启动获取列表，并进行监听，如果收到变化的通知就再次获取列表并再次监听。
1、server：
packagezookeeper.zkser;
importjava.io.IOException;
importorg.apache.zookeeper.CreateMode;
importorg.apache.zookeeper.KeeperException;
importorg.apache.zookeeper.WatchedEvent;
importorg.apache.zookeeper.Watcher;
importorg.apache.zookeeper.ZooDefs.Ids;
importorg.apache.zookeeper.ZooKeeper;
public classDistributedServer {
private static final StringconnectString="192.168.120.135:2181";
private static final intsessionTimeout=2000;
private static final StringparentNode="/servers";
private ZooKeeper zk =null;
public void getConnect() throws Exception{
zk = new ZooKeeper(connectString,sessionTimeout, new Watcher(){
//zk发生变化时会执行process方法，用来通知。
@Override
public void process(WatchedEventevent) {
// TODO Auto-generated methodstub
System.out.println(event.getType()+"----"+event.getPath());
try {
zk.getChildren("/",true);
} catch (KeeperException e) {
// TODO Auto-generated catchblock
e.printStackTrace();
} catch (InterruptedException e){
// TODO Auto-generated catchblock
e.printStackTrace();
}
}
});
}
//向zk集群注册信息
public void registerServer(String hostname)throws Exception{
String create =zk.create(parentNode+"/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);
System.out.println(hostname+"isonline.."+create);
}
//业务功能：
public void handleBussiness(String hostname)throws Exception{
System.out.println(hostname+"startworking……………………");
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args)throws Exception {
//获取zk连接：
DistributedServer server = newDistributedServer();
server.getConnect();
//利用zk连接注册服务器信息
server.registerServer(args[0]);
//启动业务功能
server.handleBussiness(args[0]);
}
}
2、client：
package zookeeper.zkcli;
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;
publicclass DistributedClient {
privatestaticfinal String connectString="192.168.120.135:2181";
privatestaticfinalintsessionTimeout=2000;
privatestaticfinal String parentNode="/servers";
//加volatile的目的何在？主要是为了让getserverlist线程与业务操作线程一致。
privatevolatile List<String> serverList;
private ZooKeeper zk =null;
publicvoid getConnect() throwsException{
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher(){
//zk发生变化时会执行process方法，用来通知。
@Override
publicvoid process(WatchedEvent event) {
// TODO Auto-generated method stub
System.out.println(event.getType()+"----"+event.getPath());
try {
getServerList();
}catch (Exception e) {
}
}
});
}
//业务功能：
publicvoid handleBussiness() throwsException{
System.out.println("client start working……………………");
Thread.sleep(Long.MAX_VALUE);
}
//获取服务器信息列表。
publicvoid getServerList() throwsException{
//获取服务器子节点信息，并对父节点进行监听
List<String>children = zk.getChildren(parentNode, true);
List<String> servers = new ArrayList<String>();
for(String child:children){
byte[] data = zk.getData(parentNode+"/"+child, false, null);
servers.add(new String(data));
}
//把servers赋值给成员变量serverlist，已提供给各业务线程使用。
serverList = servers;
//打印服务器列表：
System.out.println(serverList);
}
publicstaticvoid main(String[] args) throws Exception {
//获取连接
DistributedClientclient = new DistributedClient();
client.getConnect();
//获取servers的子节点信息（并监听），从中获取服务器信息列表
client.getServerList();
//业务线程启动
client.handleBussiness();
}
}
这两段代码写完成后，需要进行打包：
1、运行一次main函数，然后打包成runnablejar
2、然后cmd到jar包所在的目录，通过cmd来运行所有的server和client；
3、通过观察即可以看到分布式服务器上线下线的变化。
当newzookeeper的时候，他会建一个connect同时建一个listener，listener会进行通信，获取服务器的内容。
在写客户端时需要用到volatile，其目的是：
堆内存是所有线程共享的空间，如果说某一个线程需要使用堆内存中的值，需要先把堆内存中的值复制到自己的空间中，处理完成后再返回值到堆内存中，此时如果其他线程再复制堆内存中的值时有可能会产生错误，volatile的作用是让线程直接在堆内存中操作，而不会复制到自己的工作空间中。
public classSimpleDemo {
// 会话超时时间，设置为与系统默认时间一致
private static final int SESSION_TIMEOUT =30000;
// 创建 ZooKeeper 实例
ZooKeeper zk;
// 创建 Watcher 实例
Watcher wh = new Watcher() {
public voidprocess(org.apache.zookeeper.WatchedEvent event)
{
System.out.println(event.toString());
}
};
// 初始化 ZooKeeper 实例
private void createZKInstance() throwsIOException
{
zk = newZooKeeper("weekend01:2181", SimpleDemo.SESSION_TIMEOUT,this.wh);
}
private void ZKOperations() throwsIOException, InterruptedException,KeeperException
{
System.out.println("/n1. 创建 ZooKeeper 节点 (znode ： zoo2, 数据： myData2 ，权限：OPEN_ACL_UNSAFE ，节点类型：Persistent");
zk.create("/zoo2","myData2".getBytes(), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
System.out.println("/n2. 查看是否创建成功：");
System.out.println(newString(zk.getData("/zoo2", false, null)));
System.out.println("/n3. 修改节点数据 ");
zk.setData("/zoo2","shenlan211314".getBytes(), -1);
System.out.println("/n4. 查看是否修改成功：");
System.out.println(newString(zk.getData("/zoo2", false, null)));
System.out.println("/n5. 删除节点 ");
zk.delete("/zoo2", -1);
System.out.println("/n6. 查看节点是否被删除：");
System.out.println(" 节点状态： [" +zk.exists("/zoo2", false) + "]");
}
private void ZKClose() throwsInterruptedException
{
zk.close();
}
public static void main(String[] args)throws IOException,
InterruptedException,KeeperException {
SimpleDemo dm = new SimpleDemo();
dm.createZKInstance();
dm.ZKOperations();
dm.ZKClose();
}
}
Zookeeper的监听器工作机制：
监听器是一个接口，我们的代码中可以实现Wather这个接口，实现其中的process方法，方法中即我们自己的业务逻辑
监听器的注册是在获取数据的操作中实现：
getData(path,watch?)监听的事件是：节点数据变化事件
getChildren(path,watch?)监听的事件是：节点下的子节点增减变化事件
4.6.2. 应用案例（分布式应用HA||分布式锁）
实现分布式应用的(主节点HA)及客户端动态更新主节点状态
某分布式系统中，主节点可以有多台，可以动态上下线
任意一台客户端都能实时感知到主节点服务器的上下线
A、客户端实现
public classAppClient {
private String groupNode ="sgroup";
private ZooKeeper zk;
private Stat stat = new Stat();
private volatile List<String>serverList;
/**
* 连接zookeeper
*/
public void connectZookeeper() throwsException {
zk= newZooKeeper("localhost:4180,localhost:4181,localhost:4182", 5000, newWatcher() {
public void process(WatchedEventevent) {
// 如果发生了"/sgroup"节点下的子节点变化事件,更新server列表, 并重新注册监听
if (event.getType() ==EventType.NodeChildrenChanged
&& ("/" +groupNode).equals(event.getPath())) {
try {
updateServerList();
} catch (Exception e) {
e.printStackTrace();
}
}
}
});
updateServerList();
}
/**
* 更新server列表
*/
private void updateServerList() throwsException {
List<String> newServerList = newArrayList<String>();
// 获取并监听groupNode的子节点变化
// watch参数为true, 表示监听子节点变化事件.
// 每次都需要重新注册监听, 因为一次注册, 只能监听一次事件, 如果还想继续保持监听, 必须重新注册
List<String> subList =zk.getChildren("/" + groupNode, true);
for (String subNode : subList) {
// 获取每个子节点下关联的server地址
byte[] data =zk.getData("/" + groupNode + "/" + subNode, false, stat);
newServerList.add(new String(data,"utf-8"));
}
// 替换server列表
serverList = newServerList;
System.out.println("server listupdated: " + serverList);
}
/**
*client的工作逻辑写在这个方法中
* 此处不做任何处理,只让clientsleep
*/
public void handle() throwsInterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args)throws Exception {
AppClient ac = new AppClient();
ac.connectZookeeper();
ac.handle();
}
}
B、服务器端实现
public classAppServer {
private String groupNode ="sgroup";
private String subNode = "sub";
/**
* 连接zookeeper
*@param address server的地址
*/
public void connectZookeeper(String address)throws Exception {
ZooKeeper zk = new ZooKeeper(
"localhost:4180,localhost:4181,localhost:4182",
5000, newWatcher() {
public void process(WatchedEventevent) {
// 不做处理
}
});
// 在"/sgroup"下创建子节点
// 子节点的类型设置为EPHEMERAL_SEQUENTIAL, 表明这是一个临时节点,且在子节点的名称后面加上一串数字后缀
// 将server的地址数据关联到新创建的子节点上
String createdPath =zk.create("/" + groupNode + "/" + subNode,address.getBytes("utf-8"),
Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("create: "+ createdPath);
}
/**
*server的工作逻辑写在这个方法中
* 此处不做任何处理,只让serversleep
*/
public void handle() throwsInterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args)throws Exception {
// 在参数中指定server的地址
if (args.length == 0) {
System.err.println("The firstargument must be server address");
System.exit(1);
}
AppServer as = new AppServer();
as.connectZookeeper(args[0]);
as.handle();
}
}
分布式共享锁的简单实现:
1) 客户端A
publicclass DistributedClient {
// 超时时间
private static final int SESSION_TIMEOUT =5000;
// zookeeper server列表
private String hosts ="localhost:4180,localhost:4181,localhost:4182";
private String groupNode ="locks";
private String subNode = "sub";
private ZooKeeper zk;
// 当前client创建的子节点
private String thisPath;
// 当前client等待的子节点
private String waitPath;
private CountDownLatch latch = newCountDownLatch(1);
/**
* 连接zookeeper
*/
public void connectZookeeper() throwsException {
zk = new ZooKeeper(hosts,SESSION_TIMEOUT, new Watcher() {
public void process(WatchedEventevent) {
try {
// 连接建立时, 打开latch, 唤醒wait在该latch上的线程
if (event.getState() ==KeeperState.SyncConnected) {
latch.countDown();
}
// 发生了waitPath的删除事件
if (event.getType() ==EventType.NodeDeleted &&event.getPath().equals(waitPath)){
doSomething();
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
// 等待连接建立
latch.await();
// 创建子节点
thisPath = zk.create("/" +groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// wait一小会, 让结果更清晰一些
Thread.sleep(10);
// 注意, 没有必要监听"/locks"的子节点的变化情况
List<String> childrenNodes =zk.getChildren("/" + groupNode, false);
// 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁
if (childrenNodes.size() == 1) {
doSomething();
} else {
String thisNode =thisPath.substring(("/" + groupNode + "/").length());
// 排序
Collections.sort(childrenNodes);
int index =childrenNodes.indexOf(thisNode);
if (index == -1) {
// never happened
} else if (index == 0) {
// inddx == 0, 说明thisNode在列表中最小, 当前client获得锁
doSomething();
} else {
// 获得排名比thisPath前1位的节点
this.waitPath = "/" +groupNode + "/" + childrenNodes.get(index - 1);
// 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法
zk.getData(waitPath, true, newStat());
}
}
}
private void doSomething() throws Exception{
try {
System.out.println("gain lock:" + thisPath);
Thread.sleep(2000);
// do something
} finally {
System.out.println("finished:" + thisPath);
// 将thisPath删除, 监听thisPath的client将获得通知
// 相当于释放锁
zk.delete(this.thisPath, -1);
}
}
public static void main(String[] args)throws Exception {
for (int i = 0; i < 10; i++) {
new Thread() {
public void run() {
try {
DistributedClient dl =new DistributedClient();
dl.connectZookeeper();
} catch (Exception e) {
e.printStackTrace();
}
}
}.start();
}
Thread.sleep(Long.MAX_VALUE);
}
}
2) 分布式多进程模式实现：
public classDistributedClientMy {
// 超时时间
private static final int SESSION_TIMEOUT =5000;
// zookeeper server列表
private String hosts ="spark01:2181,spark02:2181,spark03:2181";
private String groupNode ="locks";
private String subNode = "sub";
private boolean haveLock = false;
private ZooKeeper zk;
// 当前client创建的子节点
private volatile String thisPath;
/**
* 连接zookeeper
*/
public void connectZookeeper() throwsException {
zk = newZooKeeper("spark01:2181", SESSION_TIMEOUT, new Watcher() {
public void process(WatchedEventevent) {
try {
//子节点发生变化
if (event.getType() ==EventType.NodeChildrenChanged && event.getPath().equals("/" +groupNode)) {
// thisPath是否是列表中的最小节点
List<String>childrenNodes = zk.getChildren("/" + groupNode, true);
String thisNode =thisPath.substring(("/" + groupNode + "/").length());
// 排序
Collections.sort(childrenNodes);
if(childrenNodes.indexOf(thisNode) == 0) {
doSomething();
thisPath =zk.create("/" + groupNode + "/" + subNode,null,Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
// 创建子节点
thisPath = zk.create("/" +groupNode + "/" + subNode, null,Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
// wait一小会, 让结果更清晰一些
Thread.sleep(newRandom().nextInt(1000));
// 监听子节点的变化
List<String> childrenNodes =zk.getChildren("/" + groupNode, true);
// 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁
if (childrenNodes.size() == 1) {
doSomething();
thisPath = zk.create("/" +groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
}
}
/**
* 共享资源的访问逻辑写在这个方法中
*/
private void doSomething() throws Exception{
try {
System.out.println("gain lock:" + thisPath);
Thread.sleep(2000);
// do something
} finally {
System.out.println("finished:" + thisPath);
// 将thisPath删除, 监听thisPath的client将获得通知
// 相当于释放锁
zk.delete(this.thisPath, -1);
}
}
public static void main(String[] args)throws Exception {
DistributedClientMy dl = newDistributedClientMy();
dl.connectZookeeper();
Thread.sleep(Long.MAX_VALUE);
}
}
5. zookeeper原理
Zookeeper虽然在配置文件中并没有指定master和slave
但是，zookeeper工作时，是有一个节点为leader，其他则为follower
Leader是通过内部的选举机制临时产生的
5.1. zookeeper的选举机制（全新集群paxos）
以一个简单的例子来说明整个选举的过程.
假设有五台服务器组成的zookeeper集群,它们的id从1-5,同时它们都是最新启动的,也就是没有历史数据,在存放数据量这一点上,都是一样的.假设这些服务器依序启动,来看看会发生什么.
1) 服务器1启动,此时只有它一台服务器启动了,它发出去的报没有任何响应,所以它的选举状态一直是LOOKING状态
2) 服务器2启动,它与最开始启动的服务器1进行通信,互相交换自己的选举结果,由于两者都没有历史数据,所以id值较大的服务器2胜出,但是由于没有达到超过半数以上的服务器都同意选举它(这个例子中的半数以上是3),所以服务器1,2还是继续保持LOOKING状态.
3) 服务器3启动,根据前面的理论分析,服务器3成为服务器1,2,3中的老大,而与上面不同的是,此时有三台服务器选举了它,所以它成为了这次选举的leader.
4) 服务器4启动,根据前面的分析,理论上服务器4应该是服务器1,2,3,4中最大的,但是由于前面已经有半数以上的服务器选举了服务器3,所以它只能接收当小弟的命了.
5) 服务器5启动,同4一样,当小弟.
5.2. 非全新集群的选举机制(数据恢复)
那么，初始化的时候，是按照上述的说明进行选举的，但是当zookeeper运行了一段时间之后，有机器down掉，重新选举时，选举过程就相对复杂了。
需要加入数据id、leader id和逻辑时钟。
数据id：数据新的id就大，数据每次更新都会更新id。
Leader id：就是我们配置的myid中的值，每个机器一个。
逻辑时钟：这个值从0开始递增,每次选举对应一个值,也就是说: 如果在同一次选举中,那么这个值应该是一致的 ; 逻辑时钟值越大,说明这一次选举leader的进程更新.
选举的标准就变成：
1、逻辑时钟小的选举结果被忽略，重新投票
2、统一逻辑时钟后，数据id大的胜出
3、数据id相同的情况下，leader id大的胜出
根据这个规则选出leader。