ZooKeeper zookeeper=new ZooKeeper(“192.168.11.152:2181”,4000,new Watcher(){

public void processor(WatchedEvent event){

?System.out.println(“event.type”);

}

})；

?

zookeeper.create(“/mic”,”0”.getByte(),ZooDefs.Ids.?OPEN_ACL_UNSAFE,CreateModel.?PERSISTENT); //創建節點

?

zookeeper.exists(“/mic”,true); //注冊監聽

?

zookeeper.setData(“/mic”, “1”.getByte(),-1) ;?//修改節點的值觸發監聽

?

ZooKeeper zookeeper=new ZooKeeper(“192.168.11.152:2181”,4000,new Watcher(){

public void processor(WatchedEvent event){

?System.out.println(“event.type”);

}

})；

????public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,

????????????boolean canBeReadOnly, HostProvider aHostProvider,

????????????ZKClientConfig clientConfig) throws IOException {

????????LOG.info("Initiating client connection, connectString=" + connectString

????????????????+ " sessionTimeout=" + sessionTimeout + " watcher=" + watcher);

?

????????if (clientConfig == null) {

????????????clientConfig = new ZKClientConfig();

????????}

????????this.clientConfig = clientConfig;

????????watchManager = defaultWatchManager();

????????watchManager.defaultWatcher = watcher; ?--在這里將watcher設置到ZKWatchManager

????????ConnectStringParser connectStringParser = new ConnectStringParser(

????????????????connectString);

????????hostProvider = aHostProvider;

????????--初始化了ClientCnxn，并且調用cnxn.start()方法

????????cnxn = new ClientCnxn(connectStringParser.getChrootPath(),

????????????????hostProvider, sessionTimeout, this, watchManager,

????????????????getClientCnxnSocket(), canBeReadOnly);

????????cnxn.start();

????}

??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); ?--初始化sendThread

????????eventThread = new EventThread(); ???????????????--初始化eventThread

????????this.clientConfig=zooKeeper.getClientConfig();

????}

?

????public void start() { --啟動兩個線程

????????sendThread.start();

????????eventThread.start();

????}

zookeeper.exists(“/mic”,true); //注冊監聽

????public Stat exists(final String path, Watcher watcher)

????????throws KeeperException, InterruptedException

????{

????????final String clientPath = path;

????????PathUtils.validatePath(clientPath);

?

????????// the watch contains the un-chroot path

????????WatchRegistration wcb = null;

????????if (watcher != null) {

????????????wcb = new ExistsWatchRegistration(watcher, clientPath);?//構建ExistWatchRegistration

????????}

?

????????final String serverPath = prependChroot(clientPath);

?

????????RequestHeader h = new RequestHeader();

????????h.setType(ZooDefs.OpCode.exists); ?//設置操作類型為exists

????????ExistsRequest request = new ExistsRequest();??//?構造ExistsRequest

????????request.setPath(serverPath);

????????request.setWatch(watcher != null);??//是否注冊監聽

????????SetDataResponse response = new SetDataResponse(); ?//設置服務端響應的接收類

//將封裝的RequestHeader、ExistsRequest、SetDataResponse、WatchRegistration添加到發送隊列

????????ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);?

????????if (r.getErr() != 0) {

????????????if (r.getErr() == KeeperException.Code.NONODE.intValue()) {

????????????????return null;

????????????}

????????????throw KeeperException.create(KeeperException.Code.get(r.getErr()),

????????????????????clientPath);

????????}

????????//返回exists得到的結果（Stat信息）

????????return response.getStat().getCzxid() == -1 ? null : response.getStat();

????}

public ReplyHeader submitRequest(RequestHeader h, Record request,

????????????Record response, WatchRegistration watchRegistration,

????????????WatchDeregistration watchDeregistration)

????????????throws InterruptedException {

????????ReplyHeader r = new ReplyHeader();

????????//將消息添加到隊列,并構造一個Packet傳輸對象

????????Packet packet = queuePacket(h, r, request, response, null, null, null, null, watchRegistration, watchDeregistration);

????????synchronized (packet) {

????????????while (!packet.finished) {?//在數據包沒有處理完成之前，一直阻塞

????????????????packet.wait();

????????????}

????????}

????????return r;

????}

????public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,

????????????Record response, AsyncCallback cb, String clientPath,

????????????String serverPath, Object ctx, WatchRegistration watchRegistration,

????????????WatchDeregistration watchDeregistration) {

????????//將相關傳輸對象轉化成Packet

????????Packet packet = null;

????????packet = new Packet(h, r, request, response, watchRegistration);

????????packet.cb = cb;

????????packet.ctx = ctx;

????????packet.clientPath = clientPath;

????????packet.serverPath = serverPath;

????????packet.watchDeregistration = watchDeregistration;

????????

????????synchronized (state) {

????????????if (!state.isAlive() || closing) {

????????????????conLossPacket(packet);

????????????} else {

????????????????if (h.getType() == OpCode.closeSession) {

????????????????????closing = true;

????????????????}

????????????????outgoingQueue.add(packet);?//添加到outgoingQueue

????????????}

????????}

????????sendThread.getClientCnxnSocket().packetAdded();//此處是多路復用機制，喚醒Selector，告訴他有數據包添加過來了

????????return packet;

????}

public void run() {

????????????clientCnxnSocket.introduce(this, sessionId, outgoingQueue);

????????????clientCnxnSocket.updateNow();

????????????clientCnxnSocket.updateLastSendAndHeard();

????????????int to;

????????????long lastPingRwServer = Time.currentElapsedTime();

????????????final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds

????????????while (state.isAlive()) {

????????????????try {

????????????????????if (!clientCnxnSocket.isConnected()) {// 如果沒有連接：發起連接

????????????????????????// don't re-establish connection if we are closing

????????????????????????if (closing) {

????????????????????????????break;

????????????????????????}

????????????????????????startConnect();?//發起連接

????????????????????????clientCnxnSocket.updateLastSendAndHeard();

????????????????????}

?

????????????????????if (state.isConnected()) {?//如果是連接狀態，則處理sasl的認證授權

????????????????????????// 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();

????????????????????}

????????????????????//to,表示客戶端距離timeout還剩多少時間，準備發起ping連接

????????????????????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()) {

????????????????????????//計算下一次ping請求的時間

????????????????????????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();?//發送ping請求

????????????????????????????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，發起傳輸

????????????????????其中 pendingQueue是一個用來存放已經發送、等待回應的Packet隊列，

clientCnxnSocket默認使用ClientCnxnSocketNIO（ps：還記得在哪里初始化嗎？在實例化zookeeper的時候）

????????????????????clientCnxnSocket.doTransport(to, pendingQueue, 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 {

????????????????????????????LOG.warn(

????????????????????????????????????"Session 0x"

????????????????????????????????????????????+ Long.toHexString(getSessionId())

????????????????????????????????????????????+ " for server "

????????????????????????????????????????????+ clientCnxnSocket.getRemoteSocketAddress()

????????????????????????????????????????????+ ", unexpected error"

????????????????????????????????????????????+ RETRY_CONN_MSG, e);

????????????????????????}

????????????????????????// At this point, there might still be new packets appended to outgoingQueue.

????????????????????????// they will be handled in next connection or cleared up if closed.

????????????????????????cleanup();

????????????????????????if (state.isAlive()) {

????????????????????????????eventThread.queueEvent(new WatchedEvent(

????????????????????????????????????Event.EventType.None,

????????????????????????????????????Event.KeeperState.Disconnected,

????????????????????????????????????null));

????????????????????????}

????????????????????????clientCnxnSocket.updateNow();

????????????????????????clientCnxnSocket.updateLastSendAndHeard();

????????????????????}

????????????????}

????????????}

????????????synchronized (state) {

????????????????// When it comes to this point, it guarantees that later queued

????????????????// packet to outgoingQueue will be notified of death.

????????????????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()));

????????}

@Override

????void doTransport(int waitTimeOut, List<Packet> pendingQueue, ClientCnxn cnxn) throws IOException, InterruptedException {

????????try {

????????????if (!firstConnect.await(waitTimeOut, TimeUnit.MILLISECONDS)) {

????????????????return;

????????????}

????????????Packet head = null;

????????????if (needSasl.get()) {

????????????????if (!waitSasl.tryAcquire(waitTimeOut, TimeUnit.MILLISECONDS)) {

????????????????????return;

????????????????}

????????????} else { ?

????????????????//判斷outgoingQueue是否存在待發送的數據包，不存在則直接返回

????????????????if ((head = outgoingQueue.poll(waitTimeOut, TimeUnit.MILLISECONDS)) == null) {

????????????????????return;

????????????????}

????????????}

????????????// check if being waken up on closing.

????????????if (!sendThread.getZkState().isAlive()) {

????????????????// adding back the patck to notify of failure in conLossPacket().

????????????????addBack(head);

????????????????return;

????????????}

????????????// channel disconnection happened

????????????if (disconnected.get()) {?//異常流程，channel關閉了，講當前的packet添加到addBack中

????????????????addBack(head);

????????????????throw new EndOfStreamException("channel for sessionid 0x"

????????????????????????+ Long.toHexString(sessionId)

????????????????????????+ " is lost");

????????????}

????????????if (head != null) {?//如果當前存在需要發送的數據包，則調用doWrite方法，pendingQueue表示處于已經發送過等待響應的packet隊列

????????????????doWrite(pendingQueue, head, cnxn);

????????????}

????????} finally {

????????????updateNow();

????????}

????}

????private void doWrite(List<Packet> pendingQueue, Packet p, ClientCnxn cnxn) {

????????updateNow();

????????while (true) {

????????????if (p != WakeupPacket.getInstance()) {

????????????????if ((p.requestHeader != null) &&?//判斷請求頭以及判斷當前請求類型不是ping或者auth操作

????????????????????????(p.requestHeader.getType() != ZooDefs.OpCode.ping) &&

????????????????????????(p.requestHeader.getType() != ZooDefs.OpCode.auth)) {

????????????????????p.requestHeader.setXid(cnxn.getXid()); ?//設置xid，這個xid用來區分請求類型

????????????????????synchronized (pendingQueue) {

????????????????????????pendingQueue.add(p);?//將當前的packet添加到pendingQueue隊列中

????????????????????}

????????????????}

????????????????sendPkt(p);?//將數據包發送出去

????????????}

????????????if (outgoingQueue.isEmpty()) {

????????????????break;

????????????}

????????????p = outgoingQueue.remove();

????????}

????}

???private void sendPkt(Packet p) {

????????// Assuming the packet will be sent out successfully. Because if it fails,

????????// the channel will close and clean up queues.

????????p.createBB();?//序列化請求數據

????????updateLastSend();?//更新最后一次發送updateLastSend

????????sentCount++;?//更新發送次數

????????channel.write(ChannelBuffers.wrappedBuffer(p.bb));?//通過nio?channel發送字節緩存到服務端

????}

public void createBB() {

????????????try {

????????????????ByteArrayOutputStream baos = new ByteArrayOutputStream();

????????????????BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);

????????????????boa.writeInt(-1, "len"); // We'll fill this in later

????????????????if (requestHeader != null) {

????????????????????requestHeader.serialize(boa, "header"); //序列化header頭(requestHeader)

????????????????}

????????????????if (request instanceof ConnectRequest) {

????????????????????request.serialize(boa, "connect");

????????????????????// append "am-I-allowed-to-be-readonly" flag

????????????????????boa.writeBool(readOnly, "readOnly");

????????????????} else if (request != null) {

????????????????????request.serialize(boa, "request"); //序列化request(request)

????????????????}

????????????????baos.close();

????????????????this.bb = ByteBuffer.wrap(baos.toByteArray());

????????????????this.bb.putInt(this.bb.capacity() - 4);

????????????????this.bb.rewind();

????????????} catch (IOException e) {

????????????????LOG.warn("Ignoring unexpected exception", e);

????????????}

????????}

cnxn = new ClientCnxn(connectStringParser.getChrootPath(),

????????????????hostProvider, sessionTimeout, this, watchManager,

????????????????getClientCnxnSocket(), canBeReadOnly);

private ClientCnxnSocket getClientCnxnSocket() throws IOException {

????????String clientCnxnSocketName = getClientConfig().getProperty(

????????????????ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET);

????????if (clientCnxnSocketName == null) {

????????????clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();

????????}

????????try {

????????????Constructor<?> clientCxnConstructor =

Class.forName(clientCnxnSocketName).getDeclaredConstructor(ZKClientConfig.class);

????????????ClientCnxnSocket clientCxnSocket = (ClientCnxnSocket) clientCxnConstructor.newInstance(getClientConfig());

????????????return clientCxnSocket;

????????} catch (Exception e) {

????????????IOException ioe = new IOException("Couldn't instantiate "

????????????????????+ clientCnxnSocketName);

????????????ioe.initCause(e);

????????????throw ioe;

????????}

????}