精选30+云产品，助力企业轻松上云！>>>

客户端连接源码分析

ZKClient 客户端，Curator 客户端

先下结论：
Client 要创建一个连接，其首先会在本地创建一个 ZooKeeper 对象，用于表示其所连接上的 Server。
连接成功后，该连接的各种临时性数据会被初始化到 zk 对象中。
连接关闭后，这个代表 Server 的 zk 对象会被删除。

我们知道常用的ZK客户端技术有ZKClient 客户端，Curator 客户端，而客户端在连接ZK Server的时候，会配置集群信息，而连接集群中具体哪一台服务器，采用轮询的方式，先将集群配置信息打散，打散以后再轮询（默认情况，当然可以指定重连策略）

ZKClient源码分析：
下面是ZKClient使用的DEMO：

public class ZKClientTest {
// 指定 zk 集群
private static final String CLUSTER = "zkOS:2181";
// 指定节点名称
private static final String PATH = "/mylog";
public static void main(String[] args) {
// ---------------- 创建会话 -----------
// 创建 zkClient
ZkClient zkClient = new ZkClient(CLUSTER);
// 为 zkClient 指定序列化器
zkClient.setZkSerializer(new SerializableSerializer());
// ---------------- 创建节点 -----------
// 指定创建持久节点
CreateMode mode = CreateMode.PERSISTENT;
// 指定节点数据内容
String data = "first log";
// 创建节点
String nodeName = zkClient.create(PATH, data, mode);
...

追踪ZKClient源码，看下是如何连接的，从ZkClient 构造开始：

public class ZkClient implements Watcher {
...
public ZkClient(String serverstring) {
this(serverstring, Integer.MAX_VALUE);
}
public ZkClient(String zkServers, int connectionTimeout) {
//关键点 看到创建了ZkConnection对象
this(new ZkConnection(zkServers), connectionTimeout);
}
...
//构造一直走，会走到下面该方法
public ZkClient(final IZkConnection zkConnection, final int connectionTimeout, final ZkSerializer zkSerializer, final long operationRetryTimeout) {
if (zkConnection == null) {
throw new NullPointerException("Zookeeper connection is null!");
}
//将创建的ZkConnection，赋值到_connection 成员变量上
_connection = zkConnection;
_zkSerializer = zkSerializer;
_operationRetryTimeoutInMillis = operationRetryTimeout;
_isZkSaslEnabled = isZkSaslEnabled();
connect(connectionTimeout, this);
}

public void connect(final long maxMsToWaitUntilConnected, Watcher watcher) throws ZkInterruptedException, ZkTimeoutException, IllegalStateException {
boolean started = false;
acquireEventLock();
try {
setShutdownTrigger(false);
_eventThread = new ZkEventThread(_connection.getServers());
_eventThread.start();
//调用ZkConnection.connect进行连接
_connection.connect(watcher);

LOG.debug("Awaiting connection to Zookeeper server");
boolean waitSuccessful = waitUntilConnected(maxMsToWaitUntilConnected, TimeUnit.MILLISECONDS);
if (!waitSuccessful) {
throw new ZkTimeoutException("Unable to connect to zookeeper server '" + _connection.getServers() + "' with timeout of " + maxMsToWaitUntilConnected + " ms");
}
started = true;
} finally {
getEventLock().unlock();

// we should close the zookeeper instance, otherwise it would keep
// on trying to connect
if (!started) {
close();
}
}
}
}

通过上面源码追踪，看到ZKClient连接实际上是通过ZkConnection.connect方法进行连接的，我们继续追踪ZkConnection

public class ZkConnection implements IZkConnection {
...
//关键对象ZooKeeper
private ZooKeeper _zk = null;
...
public ZkConnection(String zkServers, int sessionTimeOut) {
_servers = zkServers;
_sessionTimeOut = sessionTimeOut;
}

@Override
public void connect(Watcher watcher) {
_zookeeperLock.lock();
try {
if (_zk != null) {
throw new IllegalStateException("zk client has already been started");
}
try {
LOG.debug("Creating new ZookKeeper instance to connect to " + _servers + ".");
//！！！可以看到实际上ZKCLient与服务端连接，靠的就是ZooKeeper对象
_zk = new ZooKeeper(_servers, _sessionTimeOut, watcher);
} catch (IOException e) {
throw new ZkException("Unable to connect to " + _servers, e);
}
} finally {
_zookeeperLock.unlock();
}
}
}

Curator 源码分析：
下面是Curator使用的DEMO：

public class FluentTest {
public static void main(String[] args) throws Exception {
// ---------------- 创建会话 -----------
// 创建重试策略对象：重试间隔时间是1秒，最多重试 3 次
ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
// 创建客户端
CuratorFramework client = CuratorFrameworkFactory
.builder()
.connectString("zkOS:2181")
.sessionTimeoutMs(15000)
.connectionTimeoutMs(13000)
.retryPolicy(retryPolicy)
//namespace：根路径，所有操作都是基于该路径之上
.namespace("logs")
.build();
// 开启客户端
client.start();
...

追踪Curator源码，看下是如何连接的，从client.start()开始：

public class CuratorFrameworkImpl implements CuratorFramework{
...
@Override
public void start(){
log.info("Starting");
if ( !state.compareAndSet(CuratorFrameworkState.LATENT, CuratorFrameworkState.STARTED) ){
throw new IllegalStateException("Cannot be started more than once");
}

try{
...
this.getConnectionStateListenable().addListener(listener);

client.start();
...
}catch ( Exception e ){
ThreadUtils.checkInterrupted(e);
handleBackgroundOperationException(null, e);
}
}

}

关注client.start();这个方法：

public class CuratorZookeeperClient implements Closeable{
...
public void start() throws Exception
{
log.debug("Starting");

if ( !started.compareAndSet(false, true) )
{
throw new IllegalStateException("Already started");
}

state.start();
}
...
}

继续追踪state.start();

class ConnectionState implements Watcher, Closeable{
...
void start() throws Exception{
log.debug("Starting");
ensembleProvider.start();
reset();
}

synchronized void reset() throws Exception{
log.debug("reset");

instanceIndex.incrementAndGet();

isConnected.set(false);
connectionStartMs = System.currentTimeMillis();
handleHolder.closeAndReset();
handleHolder.getZooKeeper();   // initiate connection
}
```
}

关键点看handleHolder.getZooKeeper()方法：

class HandleHolder{
...
ZooKeeper getZooKeeper() throws Exception{
return (helper != null) ? helper.getZooKeeper() : null;
}
...
}

class Helper{
private final Data data;
...
ZooKeeper getZooKeeper() throws Exception{
return data.zooKeeperHandle;
}
...
}

直接从data里面取了，Hepler是什么时候创建的呢？回到org.apache.curator.ConnectionState#reset，看handleHolder.closeAndReset()方法：

class HandleHolder{
...

void closeAndReset() throws Exception{
internalClose(0);

Helper.Data data = new Helper.Data();
helper = new Helper(data){
@Override
ZooKeeper getZooKeeper() throws Exception{
synchronized(this){
if ( data.zooKeeperHandle == null ){
resetConnectionString(ensembleProvider.getConnectionString());
data.zooKeeperHandle = zookeeperFactory.newZooKeeper(data.connectionString, sessionTimeout, watcher, canBeReadOnly);
}

helper = new Helper(data);

return super.getZooKeeper();
}
}
};
}
...
}

我们看下data.zooKeeperHandle到底是怎么创建的：

public class NonAdminZookeeperFactory implements ZookeeperFactory{
@Override
public ZooKeeper newZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly) throws Exception{
return new ZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly);
}
}

可以看到，无论哪种客户端技术，最终都会在本地创建一个ZooKeeper对象，接下来我们分析ZK源码中的ZooKeeper对象

ZK源码中客户端对象ZooKeeper

我们找到ZK源码中的ZooKeeper对象代码（下面是构造代码）：

// 跟构造
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
throws IOException
{
this(connectString, sessionTimeout, watcher, false);
}

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
boolean canBeReadOnly)
throws IOException
{
LOG.info("Initiating client connection, connectString=" + connectString
+ " sessionTimeout=" + sessionTimeout + " watcher=" + watcher);

watchManager.defaultWatcher = watcher;

// 创建一个zk集群字符串解析器，将解析出的ip与port构建为一个地址实例，放入到缓存集合
ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
// 将缓存集合中的地址打散
HostProvider hostProvider = new StaticHostProvider(connectStringParser.getServerAddresses());
// 创建一个连接实例
cnxn = new ClientCnxn(connectStringParser.getChrootPath(),
hostProvider, sessionTimeout, this, watchManager,
getClientCnxnSocket(), canBeReadOnly);
// 启动连接
cnxn.start();
}

ConnectStringParser connectStringParser = new ConnectStringParser(connectString)

• 创建一个zk集群字符串解析器，将解析出的ip与port构建为一个地址实例，放入到缓存集合

  public ConnectStringParser(String connectString) {
  // parse out chroot, if any
  // 解析chroot，如果有的话（其他的一种写法，不用关注）
  int off = connectString.indexOf('/');
  if (off >= 0) {
  String chrootPath = connectString.substring(off);
  // ignore "/" chroot spec, same as null
  if (chrootPath.length() == 1) {
  this.chrootPath = null;
  } else {
  PathUtils.validatePath(chrootPath);
  this.chrootPath = chrootPath;
  }
  connectString = connectString.substring(0, off);
  } else {
  this.chrootPath = null;
  }
  //逗号分割
  String hostsList[] = connectString.split(",");
  for (String host : hostsList) {
  int port = DEFAULT_PORT;
  int pidx = host.lastIndexOf(':');
  if (pidx >= 0) {
  // otherwise : is at the end of the string, ignore
  if (pidx < host.length() - 1) {
  port = Integer.parseInt(host.substring(pidx + 1));
  }
  host = host.substring(0, pidx);
  }
  //解析出主机和端口后生成地址并添加到serverAddresses
  serverAddresses.add(InetSocketAddress.createUnresolved(host, port));
  }
  }

HostProvider hostProvider = new StaticHostProvider(connectStringParser.getServerAddresses());

• 创建主机提供者，把将缓存集合中的地址打散

  public StaticHostProvider(Collection<InetSocketAddress> serverAddresses) {
  // 实例化一个地址解析器
  this.resolver = new Resolver() {
  // 根据主机名获取其对应的所有ip
  @Override
  public InetAddress[] getAllByName(String name) throws UnknownHostException {
  return InetAddress.getAllByName(name);
  }
  };
  // 将地址打散
  init(serverAddresses);
  }
  
  private void init(Collection<InetSocketAddress> serverAddresses) {
  if (serverAddresses.isEmpty()) {
  throw new IllegalArgumentException(
  "A HostProvider may not be empty!");
  }
  // 初始化本地的sererAddresses
  this.serverAddresses.addAll(serverAddresses);
  // 将serverAddresses中的元素打散
  Collections.shuffle(this.serverAddresses);
  }

  打散的目的在于负载均衡，不然每个客户端轮询都会连上第一个

cnxn = new ClientCnxn(…);

• 创建一个连接实例

cnxn.start();

• 启动连接

  public void start() {
  // 启动连接线程
  sendThread.start();
  // 启动事件线程
  eventThread.start();
  }

查看启动连接线程sendThread的run方法

public void run() {
clientCnxnSocket.introduce(this,sessionId);
clientCnxnSocket.updateNow();
clientCnxnSocket.updateLastSendAndHeard();
int to;
long lastPingRwServer = Time.currentElapsedTime();
final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds
InetSocketAddress serverAddress = null;
// 若当前连接对象处于激活状态
while (state.isAlive()) {
try {
// 若当前连接没有连接着server
if (!clientCnxnSocket.isConnected()) {
// 若不是第一次连接，则随机sleep一会儿，小于1秒
// 休息一会儿减少服务器资源消耗
if(!isFirstConnect){
try {
Thread.sleep(r.nextInt(1000));
} catch (InterruptedException e) {
LOG.warn("Unexpected exception", e);
}
}
// don't re-establish connection if we are closing
// 若sleep过后，当前连接被关闭了，或不可用了，则直接结束
if (closing || !state.isAlive()) {
break;
}
// rwServerAddress是一个动态配置的地址，当前Client会先连接该地址
// 若连接失败，则再连接其它地址
if (rwServerAddress != null) {
serverAddress = rwServerAddress;
rwServerAddress = null;
} else {
// 获取要连接的zkServer的地址
serverAddress = hostProvider.next(1000);
}
// 开启连接尝试，有可能连接不上，连接不上会循环获取下一个地址继续尝试连接
startConnect(serverAddress);
clientCnxnSocket.updateLastSendAndHeard();
}

if (state.isConnected()) {
// determine whether we need to send an AuthFailed event.
if (zooKeeperSaslClient != null) {
boolean sendAuthEvent = false;
if (zooKeeperSaslClient.getSaslState() == ZooKeeperSaslClient.SaslState.INITIAL) {
try {
zooKeeperSaslClient.initialize(ClientCnxn.this);
} catch (SaslException e) {
LOG.error("SASL authentication with Zookeeper Quorum member failed: " + e);
state = States.AUTH_FAILED;
sendAuthEvent = true;
}
}
KeeperState authState = zooKeeperSaslClient.getKeeperState();
if (authState != null) {
if (authState == KeeperState.AuthFailed) {
// An authentication error occurred during authentication with the Zookeeper Server.
state = States.AUTH_FAILED;
sendAuthEvent = true;
} else {
if (authState == KeeperState.SaslAuthenticated) {
sendAuthEvent = true;
}
}
}

if (sendAuthEvent == true) {
eventThread.queueEvent(new WatchedEvent(
Watcher.Event.EventType.None,
authState,null));
}
}
to = readTimeout - clientCnxnSocket.getIdleRecv();
} else {
to = connectTimeout - clientCnxnSocket.getIdleRecv();
}

if (to <= 0) {
String warnInfo;
warnInfo = "Client session timed out, have not heard from server in "
+ clientCnxnSocket.getIdleRecv()
+ "ms"
+ " for sessionid 0x"
+ Long.toHexString(sessionId);
LOG.warn(warnInfo);
throw new SessionTimeoutException(warnInfo);
}
if (state.isConnected()) {
//1000(1 second) is to prevent race condition missing to send the second ping
//also make sure not to send too many pings when readTimeout is small
int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() -
((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);
//send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL
if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {
sendPing();
clientCnxnSocket.updateLastSend();
} else {
if (timeToNextPing < to) {
to = timeToNextPing;
}
}
}

// If we are in read-only mode, seek for read/write server
if (state == States.CONNECTEDREADONLY) {
long now = Time.currentElapsedTime();
int idlePingRwServer = (int) (now - lastPingRwServer);
if (idlePingRwServer >= pingRwTimeout) {
lastPingRwServer = now;
idlePingRwServer = 0;
pingRwTimeout =
Math.min(2*pingRwTimeout, maxPingRwTimeout);
pingRwServer();
}
to = Math.min(to, pingRwTimeout - idlePingRwServer);
}

clientCnxnSocket.doTransport(to, pendingQueue, outgoingQueue, ClientCnxn.this);
} catch (Throwable e) {
if (closing) {
if (LOG.isDebugEnabled()) {
// closing so this is expected
LOG.debug("An exception was thrown while closing send thread for session 0x"
+ Long.toHexString(getSessionId())
+ " : " + e.getMessage());
}
break;
} else {
// this is ugly, you have a better way speak up
if (e instanceof SessionExpiredException) {
LOG.info(e.getMessage() + ", closing socket connection");
} else if (e instanceof SessionTimeoutException) {
LOG.info(e.getMessage() + RETRY_CONN_MSG);
} else if (e instanceof EndOfStreamException) {
LOG.info(e.getMessage() + RETRY_CONN_MSG);
} else if (e instanceof RWServerFoundException) {
LOG.info(e.getMessage());
} else if (e instanceof SocketException) {
LOG.info("Socket error occurred: {}: {}", serverAddress, e.getMessage());
} else {
LOG.warn("Session 0x{} for server {}, unexpected error{}",
Long.toHexString(getSessionId()),
serverAddress,
RETRY_CONN_MSG,
e);
}
cleanup();
if (state.isAlive()) {
eventThread.queueEvent(new WatchedEvent(
Event.EventType.None,
Event.KeeperState.Disconnected,
null));
}
clientCnxnSocket.updateNow();
clientCnxnSocket.updateLastSendAndHeard();
}
}
}
cleanup();
clientCnxnSocket.close();
if (state.isAlive()) {
eventThread.queueEvent(new WatchedEvent(Event.EventType.None,
Event.KeeperState.Disconnected, null));
}
ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),
"SendThread exited loop for session: 0x"
+ Long.toHexString(getSessionId()));
}

• while (state.isAlive()) {

  判断当前连接对象是否处于激活状态

  // 一共有这么多状态
  public enum States {
  CONNECTING, ASSOCIATING, CONNECTED, CONNECTEDREADONLY,
  CLOSED, AUTH_FAILED, NOT_CONNECTED;
  
  public boolean isAlive() {
  return this != CLOSED && this != AUTH_FAILED;
  }
  
  /**
  * Returns whether we are connected to a server (which
  * could possibly be read-only, if this client is allowed
  * to go to read-only mode)
  * */
  public boolean isConnected() {
  return this == CONNECTED || this == CONNECTEDREADONLY;
  }
  }

// 启动的会执行该方法
public synchronized void startup() {
if (sessionTracker == null) {
// 创建一个sessionTracker线程
createSessionTracker();
}
// 开启sessionTracker线程
startSessionTracker();
setupRequestProcessors();

registerJMX();

setState(State.RUNNING);
notifyAll();
}

protected void createSessionTracker() {
sessionTracker = new SessionTrackerImpl(this, zkDb.getSessionWithTimeOuts(),
tickTime, 1, getZooKeeperServerListener());
}

//SessionTrackerImpl调用的构造
public SessionTrackerImpl(SessionExpirer expirer,
ConcurrentHashMap<Long, Integer> sessionsWithTimeout, int tickTime,
long sid, ZooKeeperServerListener listener)
{
super("SessionTracker", listener);
this.expirer = expirer;
// 桶的大小，即过期时间间隔
this.expirationInterval = tickTime;
this.sessionsWithTimeout = sessionsWithTimeout;
// 计算当前时间所在的会话桶
nextExpirationTime = roundToInterval(Time.currentElapsedTime());
// 初始化下一个会话一但创建所用的sessionId
this.nextSessionId = initializeNextSession(sid);
//从内存中取出之前保存的会话数据，重新加载到sessionTracker（应该是选举后重启之类的场景）
for (Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) {
// 会话不存在则创建，并将会话放入相应桶中
addSession(e.getKey(), e.getValue());
}
}

// 计算指定时间所在的桶
private long roundToInterval(long time) {
// We give a one interval grace period
// 即计算当前时间已经走过的时间桶数量+1，在乘以时间桶大小，得到我的桶的过期时间点
return (time / expirationInterval + 1) * expirationInterval;
}

// 创建当前会话，并将会话放入相应桶中
synchronized public void addSession(long id, int sessionTimeout) {
sessionsWithTimeout.put(id, sessionTimeout);
if (sessionsById.get(id) == null) {
// 创建会话实例，刚创建的会话不在任何桶中
SessionImpl s = new SessionImpl(id, sessionTimeout, 0);
// 将会话放入到缓存map，注意这个map不是会话桶
sessionsById.put(id, s);
if (LOG.isTraceEnabled()) {
ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
"SessionTrackerImpl --- Adding session 0x"
+ Long.toHexString(id) + " " + sessionTimeout);
}
} else {
if (LOG.isTraceEnabled()) {
ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
"SessionTrackerImpl --- Existing session 0x"
+ Long.toHexString(id) + " " + sessionTimeout);
}
}
// 将会话放入到桶中
touchSession(id, sessionTimeout);
}

//先看下创建会话做了什么
public static class SessionImpl implements Session {
SessionImpl(long sessionId, int timeout, long expireTime) {
this.sessionId = sessionId;
this.timeout = timeout;
// 注意，这个tickTime用于记录的是当前会话所在的桶
// 这个初值为0，表示当前会话不在任何桶中
this.tickTime = expireTime;
isClosing = false;
}
...
}

// 将会话放入到桶中，该方法的返回值表示当前会话是否有效
synchronized public boolean touchSession(long sessionId, int timeout) {
if (LOG.isTraceEnabled()) {
ZooTrace.logTraceMessage(LOG,
ZooTrace.CLIENT_PING_TRACE_MASK,
"SessionTrackerImpl --- Touch session: 0x"
+ Long.toHexString(sessionId) + " with timeout " + timeout);
}
// 根据session从缓存map中获取相应Session
SessionImpl s = sessionsById.get(sessionId);

// Return false, if the session doesn't exists or marked as closing
// 若会话不存在，或会话关闭了，则返回false
if (s == null || s.isClosing()) {
return false;
}
// 计算本次会话交互的下次超时时间点所在的会话桶
long expireTime = roundToInterval(Time.currentElapsedTime() + timeout);
// s.tickTime用于记录当前会话所在的桶
// 若s.tickTime < expireTime，说明应该换桶了
// 若s.tickTime == expireTime，说明不用换桶
// 注意不可能出现s.tickTime > expireTime的情况
if (s.tickTime >= expireTime) {
// Nothing needs to be done
// 不需要换桶，不做任何处理
return true;
}

// 代码走到这里，说明s.tickTime < expireTime成立
// 下面要进行换桶了
// sessionSets是桶集合，key为桶的边界时间点，value为桶实例SessionSet
SessionSet set = sessionSets.get(s.tickTime);
// 若该桶不空，则从该桶中将会话移除
if (set != null) {
set.sessions.remove(s);
}
// 更新当前会话所在桶的id，即桶的边界桶
s.tickTime = expireTime;
// 从桶集合中查找该桶
set = sessionSets.get(s.tickTime);
// 若该桶不存在，则创建一个桶，然后将该新建的桶放入桶集合
if (set == null) {
set = new SessionSet();
sessionSets.put(expireTime, set);
}
// 将会话放到新桶
set.sessions.add(s);
return true;
}

//SessionSet就是桶实例，里面放了Session
static class SessionSet {
HashSet<SessionImpl> sessions = new HashSet<SessionImpl>();
}

@Override
synchronized public void run() {
try {
// 只要当前server处理运行态，则该while就不会停止
while (running) {
currentTime = Time.currentElapsedTime();
// nextExpirationTime是此时还在使用的那个桶的边界时间
// 刚启动的时候会为nextExpirationTime赋值
// 如果当前时间还没超过桶的边界时间，说明桶还没过期，不需要处理
if (nextExpirationTime > currentTime) {
this.wait(nextExpirationTime - currentTime);
continue;
}
// 代码走到这里，说明nextExpirationTime <= currentTime成立
// 说明nextExpirationTime桶已经过期，需要将该桶从桶集合中清除，
// 需要将该桶中的会话关闭
SessionSet set;
// 清除当前桶
set = sessionSets.remove(nextExpirationTime);
// 关闭该桶中的所有会话
if (set != null) {
for (SessionImpl s : set.sessions) {
setSessionClosing(s.sessionId);
expirer.expire(s);
}
}
// 获取下一个桶id
nextExpirationTime += expirationInterval;
}
} catch (InterruptedException e) {
handleException(this.getName(), e);
}
LOG.info("SessionTrackerImpl exited loop!");
}