利用ZooKeeper开发服务器上下线感知程序
What is ZooKeeper
ZooKeeper是一个分布式的分布式应用程序协调服务。简单地来说，就是用于协调管理多个分布式应用程序的一个工具，扮演着一个第三方管理者的角色。
问题背景分析
假设现在有10个应用程序(App#0 - App#9)，运行在由10台服务器(Server#0 - Server#9)组成的集群上（假设平均分配，每台服务器上运行一个程序）。此时由于某个热门线上活动的开始（如抢票or低价秒杀等），突然间有数以百万计的用户访问服务器上的资源，等待服务器处理并应答（如下图所示）。
server.png
很不幸10台服务器中有K台受不住负载压力，导致服务器崩溃。在这种情况下，如果客户端无法感知服务器的状态（在线／离线），部分向已经崩溃的服务器发送请求的客户端将会有长时间无法获得应答，它们只能一直重复地向已经崩溃的服务器地址重发请求，无法切换至另外（10-K）台完好的服务器进行交互。
breakdown.png
其实在这种场景下，如果客户端能够及时地感知到集群中哪些节点已经崩溃，哪些节点仍然完好，是可以切换至完好的节点并向其发送请求的。理论上只要集群中仍有1个节点是完好的，它即能向客户端提供服务。
所以整个问题的症结就在于，如何让客户端感知到服务器上下线状态，以便切换请求发送的地址。
zk.png
重新参考ZooKeeper的功能描述，ZooKeeper可以用来协调管理多个分布式应用程序，那其实可以用于管理我们的分布式机器集群。如上图所示，在用户和服务器集群中间可设置ZooKeeper层，让ZooKeeper实时感知每一个节点的状态，然后客户端并不直接向具体节点发起请求，而应先向ZooKeeper询问当前仍然存活的服务器节点，然后再从中挑选一个负载较低的服务器节点进行交互。由于ZooKeeper本身的高可用性（本身也可拓展为分布式架构），所以就能大大地提高整个系统的可用性。
ZooKeeper数据结构
ZooKeeper数据结构采用了树状结构（在文件系统中被广泛使用），且不是简单的二叉树，而是多叉树。在ZooKeeper的树结构中，每一个节点被称为znode，可通过控制台命令或者Java的SDK对内部数据进行管理。
znode的类型有2*2=4种，分别是：
- PERSISTENT
- PERSISTENT_SEQUENTIAL
- EPHEMERAL
- EPHEMERAL_SEQUENTIAL
其中PERSISTENT和EPHEMERAL的区别正如其名，在无外力影响下PERSISTENT节点不会被改变和删除，而EPHEMERAL节点在创建节点的session结束后会自动从树中删除。至于SEQUENTIAL与非SEQUENTIAL则影响了节点id自增，SEQUENTIAL节点的id会自动遵循父节点下的自增规则进行命名。
zkds.png
如图所示，在本问题中我们可以把一台服务器看作树中的一个节点，我们可以利用EPHEMERAL节点的这一特性进行服务器状态的监听。服务器上线时创建与zk之间的session并向zk注册节点，只要服务器不崩溃，session便不会结束，即EPHEMERAL节点会一直存在，可被客户端感知；当服务器崩溃时，其与zk之间保持的session自然也会结束，EPHEMERAL节点会自动被删除，客户端查询服务器列表时绝对无法获得已删除的节点信息。
Demo程序
- Server.java (服务器端代码)
package my.bigdata.zk;
import org.apache.zookeeper.*;
public class Server {
private static final String HOST_ADDRESS = "localhost:2181";
private static final int DEFAULT_TIMEOUT = 2000;
private static final String DEFAULT_SERVER_PARENT = "/servers";
private ZooKeeper zkConnect = null;
/**
* 连接至ZooKeeper
* @throws Exception
*/
public void connect() throws Exception{
zkConnect = new ZooKeeper(HOST_ADDRESS, DEFAULT_TIMEOUT, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
System.out.println("Type:" + watchedEvent.getType()
+ " Path:" + watchedEvent.getPath());
}
});
}
/**
* 向ZooKeeper注册本服务器节点
* @param data 服务器信息
* @throws Exception
*/
public void register(String data) throws Exception{
String create = zkConnect.create(DEFAULT_SERVER_PARENT + "/server",
data.getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL); // 注册成ephemeral节点以便自动在zk上注销
System.out.println(create + " is registered!");
}
/**
* 通过sleep模拟服务器在线
*/
public void sleep() {
try {
Thread.sleep(20000);
} catch (Exception e) {
System.out.println(e.toString());
}
}
public static void main(String[] args) throws Exception {
//连接至zk
Server server = new Server();
server.connect();
//向zk注册服务器信息
String data = args[0];
server.register(data);
server.sleep();
}
}
服务器端的重点在于，程序启动时向ZooKeeper的指定节点下注册服务器信息，相当于通知ZooKeeper这个第三方：“服务器已上线”。其次，注册的节点类型必须是ephemeral节点，为了实现节点id自增(auto-increment)还可以使用ephemeral_sequential节点。
- Client.java (客户端代码)
package my.bigdata.zk;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class Client {
private static final String HOST_ADDRESS = "localhost:2181";
private static final int DEFAULT_TIMEOUT = 2000;
private static final String DEFAULT_SERVER_PARENT = "/servers";
private ZooKeeper zkConnect = null;
private List<String> availableServers;
/**
* 连接至ZooKeeper
* @throws Exception
*/
public void connect() throws Exception {
zkConnect = new ZooKeeper(HOST_ADDRESS, DEFAULT_TIMEOUT, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
try {
updateServerCondition(); // 重复注册
} catch (Exception e) {
System.out.println(e.toString());
}
}
});
}
/**
* 向zk查询服务器情况, 并update本地服务器列表
* @throws Exception
*/
public void updateServerCondition() throws Exception {
List<String> children = zkConnect.getChildren(DEFAULT_SERVER_PARENT, true);
List<String> servers = new ArrayList<>();
for(String child : children) {
byte[] data = zkConnect.getData(DEFAULT_SERVER_PARENT + "/" + child,
false,
null);
servers.add(new String(data));
}
availableServers = servers;
System.out.println(Arrays.toString(servers.toArray(new String[0])));
}
/**
* 通过sleep让客户端持续运行，模拟"监听"
*/
public void sleep() throws Exception{
System.out.println("client is working");
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
// 连接zk
Client client = new Client();
client.connect();
// 获取servers节点信息（并监听），从中获取服务器信息列表
client.updateServerCondition();
client.sleep();
}
}
客户端的重点在于，它不断地向ZooKeeper某个特定节点（此处是servers节点）注册了一个Watcher，那么一旦该节点下的结构发生改变，ZooKeeper会向注册了Watcher的客户端发送“状态变化”的消息，那么客户端即可动态地从ZooKeeper中获取最新的服务器节点信息，甚至无需“主动”询问。
当然，ZooKeeper的应用场景还有很多，考虑到它本身也可拓展为一个分布式应用，在这种高可用性保证下它简直就是多个分布式应用的万能管家和协调者😊。
