传统直播颠覆性变化，直播云服务助力行业变革
游戏直播、电商直播、综艺直播……各类型直播无处不在。遍地都是直播的时代，直播行业的变革也变得愈发重要。
在很多时候，我们都可以在各种框架应用中看到ZooKeeper的身影，比如Kafka中间件，Dubbo框架，Hadoop等等。为什么到处都看到ZooKeeper？
一、什么是ZooKeeper
ZooKeeper是一个分布式服务协调框架，提供了分布式数据一致性的解决方案，基于ZooKeeper的数据结构，Watcher，选举机制等特点，可以实现数据的发布/订阅，软负载均衡，命名服务，统一配置管理，分布式锁，集群管理等等。
二、为什么使用ZooKeeper
ZooKeeper能保证：
更新请求顺序进行。来自同一个client的更新请求按其发送顺序依次执行
数据更新原子性。一次数据更新要么成功，要么失败
全局唯一数据视图。client无论连接到哪个server，数据视图都是一致的
实时性。在一定时间范围内，client读到的数据是最新的
三、数据结构
ZooKeeper的数据结构和Unix文件系统很类似，总体上可以看做是一棵树，每一个节点称之为一个ZNode，每一个ZNode默认能存储1M的数据。每一个ZNode可通过唯一的路径标识。如下图所示：
数据结构示意图
创建ZNode时，可以指定以下四种类型，包括：
PERSISTENT，持久性ZNode。创建后，即使客户端与服务端断开连接也不会删除，只有客户端主动删除才会消失。
PERSISTENT_SEQUENTIAL，持久性顺序编号ZNode。和持久性节点一样不会因为断开连接后而删除，并且ZNode的编号会自动增加。
EPHEMERAL，临时性ZNode。客户端与服务端断开连接，该ZNode会被删除。
EPEMERAL_SEQUENTIAL，临时性顺序编号ZNode。和临时性节点一样，断开连接会被删除，并且ZNode的编号会自动增加。
四、监听通知机制
Watcher是基于观察者模式实现的一种机制。如果我们需要实现当某个ZNode节点发生变化时收到通知，就可以使用Watcher监听器。
客户端通过设置监视点（watcher）向 ZooKeeper 注册需要接收通知的 znode，在 znode 发生变化时 ZooKeeper 就会向客户端发送消息。
这种通知机制是一次性的。一旦watcher被触发，ZooKeeper就会从相应的存储中删除。如果需要不断监听ZNode的变化，可以在收到通知后再设置新的watcher注册到ZooKeeper。
监视点的类型有很多，如监控ZNode数据变化、监控ZNode子节点变化、监控ZNode 创建或删除。
五、选举机制
ZooKeeper是一个高可用的应用框架，因为ZooKeeper是支持集群的。ZooKeeper在集群状态下，配置文件是不会指定Master和Slave，而是在ZooKeeper服务器初始化时就在内部进行选举，产生一台做为Leader，多台做为Follower，并且遵守半数可用原则。
由于遵守半数可用原则，所以5台服务器和6台服务器，实际上最大允许宕机数量都是3台，所以为了节约成本，集群的服务器数量一般设置为奇数。
如果在运行时，如果长时间无法和Leader保持连接的话，则会再次进行选举，产生新的Leader，以保证服务的可用。
选举机制示意图
六、初の体验
首先在官网下载ZooKeeper，我这里用的是3.3.6版本。
然后解压，复制一下/conf目录下的zoo_sample.cfg文件，重命名为zoo.cfg。
修改zoo.cfg中dataDir的值，并创建对应的目录：
最后到/bin目录下启动，我用的是window系统，所以启动zkServer.cmd，双击即可：
ZooKeeper目录结构
启动成功的话就可以看到这个对话框：
ZooKeeper启动成功
可视化界面的话，我推荐使用ZooInspector，操作比较简便：
ZooInspector
6.1 使用java连接ZooKeeper
首先引入Maven依赖：
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.6</version></dependency>
接着我们写一个Main方法，进行操作：
//连接地址及端口号
private static final String SERVER_HOST = “127.0.0.1:2181”;
//会话超时时间
private static final int SESSION_TIME_OUT = 2000;
public static void main(String[] args) throws Exception {
//参数一：服务端地址及端口号
//参数二：超时时间
//参数三：监听器
ZooKeeper zooKeeper = new ZooKeeper(SERVER_HOST, SESSION_TIME_OUT, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
//获取事件的状态
Event.KeeperState state = watchedEvent.getState();
//判断是否是连接事件
if (Event.KeeperState.SyncConnected == state) {
Event.EventType type = watchedEvent.getType();
if (Event.EventType.None == type) {
System.out.println(“zk客户端已连接…”);
}
}
}
});
zooKeeper.create(“/java”, “Hello World”.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println(“新增ZNode成功”);
zooKeeper.close();
}
创建一个持久性ZNode，路径是/java，值为”Hello World”：
Hello world
七、API概述
7.1 创建
public String create(final String path, byte data[], List<ACL> acl, CreateMode createMode)
参数解释：
path ZNode路径
data ZNode存储的数据
acl ACL权限控制
createMode ZNode类型
ACL权限控制，有三个是ZooKeeper定义的常用权限，在ZooDefs.Ids类中：
/**
* This is a completely open ACL.
* 完全开放的ACL，任何连接的客户端都可以操作该属性znode
*/public final ArrayList<ACL> OPEN_ACL_UNSAFE = new ArrayList<ACL>(Collections.singletonList(new ACL(Perms.ALL, ANYONE_ID_UNSAFE)));
/**
* This ACL gives the creators authentication id’s all permissions.
* 只有创建者才有ACL权限
*/public final ArrayList<ACL> CREATOR_ALL_ACL = new ArrayList<ACL>(Collections.singletonList(new ACL(Perms.ALL, AUTH_IDS)));
/**
* This ACL gives the world the ability to read.
* 只能读取ACL
*/public final ArrayList<ACL> READ_ACL_UNSAFE = new ArrayList<ACL>(Collections.singletonList(new ACL(Perms.READ, ANYONE_ID_UNSAFE)));
createMode就是前面讲过的四种ZNode类型：
public enum CreateMode {
/**
* 持久性ZNode
*/
PERSISTENT (0, false, false),
/**
* 持久性自动增加顺序号ZNode
*/
PERSISTENT_SEQUENTIAL (2, false, true),
/**
* 临时性ZNode
*/
EPHEMERAL (1, true, false),
/**
* 临时性自动增加顺序号ZNode
*/
EPHEMERAL_SEQUENTIAL (3, true, true);
}
7.2 查询
//同步获取节点数据public byte[] getData(String path, boolean watch, Stat stat){
…
}
//异步获取节点数据public void getData(final String path, Watcher watcher, DataCallback cb, Object ctx){
…
}
同步getData()方法中的stat参数是用于接收返回的节点描述信息：
public byte[] getData(final String path, Watcher watcher, Stat stat){
//省略…
GetDataResponse response = new GetDataResponse();
//发送请求到ZooKeeper服务器，获取到response
ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
if (stat != null) {
//把response的Stat赋值到传入的stat中
DataTree.copyStat(response.getStat(), stat);
}
}
使用同步getData()获取数据：
//数据的描述信息，包括版本号，ACL权限，子节点信息等等
Stat stat = new Stat();
//返回结果是byte[]数据，getData()方法底层会把描述信息复制到stat对象中
byte[] bytes = zooKeeper.getData(“/java”, false, stat);
//打印结果
System.out.println(“ZNode的数据data:” + new String(bytes));//Hello World
System.out.println(“获取到dataVersion版本号:” + stat.getVersion());//默认数据版本号是0
验证信息
7.3 更新
public Stat setData(final String path, byte data[], int version){
…
}
值得注意的是第三个参数version，使用CAS机制，这是为了防止多个客户端同时更新节点数据，所以需要在更新时传入版本号，每次更新都会使版本号+1，如果服务端接收到版本号，对比发现不一致的话，则会抛出异常。
所以，在更新前需要先查询获取到版本号，否则你不知道当前版本号是多少，就没法更新：
//获取节点描述信息
Stat stat = new Stat();
zooKeeper.getData(“/java”, false, stat);
System.out.println(“更新ZNode数据…”);
//更新操作，传入路径，更新值，版本号三个参数,返回结果是新的描述信息
Stat setData = zooKeeper.setData(“/java”, “fly!!!”.getBytes(), stat.getVersion());
System.out.println(“更新后的版本号为：” + setData.getVersion());//更新后的版本号为：1
更新后，版本号增加了：
如果传入的版本参数是”-1″，就是告诉zookeeper服务器，客户端需要基于数据的最新版本进行更新操作。但是-1并不是一个合法的版本号，而是一个标识符。
7.4 删除
public void delete(final String path, int version){
…
}
path 删除节点的路径
version 版本号
这里也需要传入版本号，调用getData()方法即可获取到版本号，很简单：
Stat stat = new Stat();
zooKeeper.getData(“/java”, false, stat);//删除ZNode
zooKeeper.delete(“/java”, stat.getVersion());
7.5 watcher机制
在上面第三点提到，ZooKeeper是可以使用通知监听机制，当ZNode发生变化会收到通知消息，进行处理。基于watcher机制，ZooKeeper能玩出很多花样。怎么使用？
ZooKeeper的通知监听机制，总的来说可以分为三个过程：
①客户端注册 Watcher ②服务器处理 Watcher ③客户端回调 Watcher客户端。
注册 watcher 有 4 种方法，new ZooKeeper()、getData()、exists()、getChildren()。下面演示一下使用exists()方法注册watcher：
首先需要实现Watcher接口，新建一个监听器：
public class MyWatcher implements Watcher {
@Override
public void process(WatchedEvent event) {
//获取事件类型
Event.EventType eventType = event.getType();
//通知状态
Event.KeeperState eventState = event.getState();
//节点路径
String eventPath = event.getPath();
System.out.println(“监听到的事件类型:” + eventType.name());
System.out.println(“监听到的通知状态:” + eventState.name());
System.out.println(“监听到的ZNode路径:” + eventPath);
}
}
然后调用exists()方法，注册监听器：
zooKeeper.exists(“/java”, new MyWatcher());//对ZNode进行更新数据的操作，触发监听器
zooKeeper.setData(“/java”, “fly”.getBytes(), -1);
然后在控制台就可以看到打印的信息：
事件监听
这里我们可以看到Event.EventType对象就是事件类型，我们可以对事件类型进行判断，再配合Event.KeeperState通知状态，做相关的业务处理，事件类型有哪些？
打开EventType、KeeperState的源码查看：
//事件类型是一个枚举public enum EventType {
None (-1),//无
NodeCreated (1),//Watcher监听的数据节点被创建
NodeDeleted (2),//Watcher监听的数据节点被删除
NodeDataChanged (3),//Watcher监听的数据节点内容发生变更
NodeChildrenChanged (4);//Watcher监听的数据节点的子节点列表发生变更
}
//通知状态也是一个枚举public enum KeeperState {
Unknown (-1),//属性过期
Disconnected (0),//客户端与服务端断开连接
NoSyncConnected (1),//属性过期
SyncConnected (3),//客户端与服务端正常连接
AuthFailed (4),//身份认证失败
ConnectedReadOnly (5),//返回这个状态给客户端，客户端只能处理读请求
SaslAuthenticated(6),//服务器采用SASL做校验时
Expired (-112);//会话session失效
}
7.5.1 watcher的特性
一次性。一旦watcher被触发，ZK都会从相应的存储中移除。
zooKeeper.exists(“/java”, new Watcher() {
@Override
public void process(WatchedEvent event) {
System.out.println(“我是exists()方法的监听器”);
}
});
//对ZNode进行更新数据的操作，触发监听器
zooKeeper.setData(“/java”, “fly”.getBytes(), -1);
//企图第二次触发监听器
zooKeeper.setData(“/java”, “spring”.getBytes(), -1);
触发一次监听器
串行执行。客户端Watcher回调的过程是一个串行同步的过程，这是为了保证顺序。
zooKeeper.exists(“/java”, new Watcher() {
@Override
public void process(WatchedEvent event) {
System.out.println(“我是exists()方法的监听器”);
}
});
Stat stat = new Stat();
zooKeeper.getData(“/java”, new Watcher() {
@Override
public void process(WatchedEvent event) {
System.out.println(“我是getData()方法的监听器”);
}
}, stat);
//对ZNode进行更新数据的操作，触发监听器
zooKeeper.setData(“/java”, “fly”.getBytes(), stat.getVersion());
打印结果，说明先调用exists()方法的监听器，再调用getData()方法的监听器。因为exists()方法先注册了。
exists()方法先注册
轻量级。WatchedEvent是ZK整个Watcher通知机制的最小通知单元。WatchedEvent包含三部分：通知状态，事件类型，节点路径。Watcher通知仅仅告诉客户端发生了什么事情，而不会说明事件的具体内容。
转载请注明：小猪云服务器租用推荐 » ZooKeeper使用初级使用篇，一文搞定