Zookeeper 源码(三)Zookeeper 客户端源码
Zookeeper 客户端由以下几个核心组件组成:
类 | 说明 |
---|---|
Zookeeper | Zookeeper 客户端入口 |
ClientWatchManager | 客户端 Watcher 管理器 |
HostProvider | 客户端地址列表管理器 |
ClientCnxn | 客户端核心线程,其内部又包含两个线程,即 SendThread 和 EventThread。前者是一个 IO 线程,主要负责 ZooKeeper 客户端和服务端之间的网络通信;后者是一个事件线程,主要负责对服务端事件进行处理。 |
ClientCnxnSocketNetty | 最底层的通信 netty |
客户端整体结构如下图:
一、Zookeeper
客户端在构造阶段创建 ClientCnxn 与服务端连接,后续命令都是通过 ClientCnxn 发送给服务端。ClientCnxn 是客户端与服务端通信的底层接口,它和 ClientCnxnSocketNetty 一起工作提供网络通信服务。
服务端是 ZookeeperServer 类,收到 ClientCnxn 的请求处理后再通过 ClientCnxn 返回到客户端。
ClientCnxn 连接时可以同时指定多台服务器地址,根据指定的算法连接一台服务器,当某个服务器发生故障无法连接时,会自动连接到其他的服务器。实现这一机制的是 StaticHostProvider 类。
(1) 客户端使用:
ZooKeeper zooKeeper = new ZooKeeper("127.0.0.1", 5000, new Watcher() {
@Override
public void process(WatchedEvent event) {
if (event.getState() == Event.KeeperState.SyncConnected) {
if (type == Event.EventType.None) {
// 如果连接建立成功才能继续执行
countDownLatch.countDown();
}
}
}
});
countDownLatch.await();
zooKeeper.create(
"/testRoot", // 节点路径,不允许递归创建节点
"testRoot".getBytes(), // 节点内容
ZooDefs.Ids.OPEN_ACL_UNSAFE, // 节点权限,一般情况下不用关注
CreateMode.PERSISTENT); // 节点类型
}
(2) ZooKeeper 创建
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
long sessionId, byte[] sessionPasswd, boolean canBeReadOnly,
HostProvider aHostProvider) throws IOException {
// 1. watcher 保存在 ZKWatchManager 的 defaultWatcher 中,作为整个会话的默认 watcher
watchManager = defaultWatchManager();
watchManager.defaultWatcher = watcher;
// 2. 解析 server 获取 IP 以及 PORT
ConnectStringParser connectStringParser = new ConnectStringParser(
connectString);
hostProvider = aHostProvider;
// 3. 创建 ClientCnxn 实例
cnxn = new ClientCnxn(connectStringParser.getChrootPath(),
hostProvider, sessionTimeout, this, watchManager,
getClientCnxnSocket(), sessionId, sessionPasswd, canBeReadOnly);
cnxn.seenRwServerBefore = true; // since user has provided sessionId
// 4. 启动 SendThread 和 EventThread 线程,这两个线程均为守护线程
cnxn.start();
}
创建底层通信 ClientCnxnSocketNIO 或 ClientCnxnSocketNetty
public static final String ZOOKEEPER_CLIENT_CNXN_SOCKET = "zookeeper.clientCnxnSocket";
private static ClientCnxnSocket getClientCnxnSocket() throws IOException {
String clientCnxnSocketName = System
.getProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET);
if (clientCnxnSocketName == null) {
clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();
}
try {
return (ClientCnxnSocket) Class.forName(clientCnxnSocketName)
.newInstance();
} catch (Exception e) {
IOException ioe = new IOException("Couldn't instantiate "
+ clientCnxnSocketName);
ioe.initCause(e);
throw ioe;
}
}
(3) ClientCnxn 创建
Packet | 所有的请求都会封装成 packet
outgoingQueue | 即将发送的请求 packets
pendingQueue | 已经发送等待响应的 packets
ClientCnxn 创建时创建了两个线程 SendThread 和 EventThread,这两个线程都是守护线程,主线程结束时即关闭线程。
public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,
long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {
this.zooKeeper = zooKeeper;
this.watcher = watcher;
this.sessionId = sessionId;
this.sessionPasswd = sessionPasswd;
this.sessionTimeout = sessionTimeout;
this.hostProvider = hostProvider;
this.chrootPath = chrootPath;
connectTimeout = sessionTimeout / hostProvider.size();
readTimeout = sessionTimeout * 2 / 3;
readOnly = canBeReadOnly;
sendThread = new SendThread(clientCnxnSocket);
eventThread = new EventThread();
}
SendThread(ClientCnxnSocket clientCnxnSocket) {
super(makeThreadName("-SendThread()"));
state = States.CONNECTING;
this.clientCnxnSocket = clientCnxnSocket;
setDaemon(true);
}
EventThread() {
super(makeThreadName("-EventThread"));
setDaemon(true);
}