1. 服务端

服务启动：
jetbrains://idea/navigate/reference?project=tuling-zookeeper&fqn=org.apache.zookeeper.server.quorum.QuorumPeerMain

  1. public static void main(String[] args) {
  2.     QuorumPeerMain main = new QuorumPeerMain();
  3.     try {
  4.         // 主要函数（进入）
  5.         main.initializeAndRun(args);
  6.     } catch (IllegalArgumentException e) {
  7.         LOG.error("Invalid arguments, exiting abnormally", e);
  8.         LOG.info(USAGE);
  9.         System.err.println(USAGE);
  10.         System.exit(2);
  11.     } catch (ConfigException e) {
  12.         LOG.error("Invalid config, exiting abnormally", e);
  13.         System.err.println("Invalid config, exiting abnormally");
  14.         System.exit(2);
  15.     } catch (Exception e) {
  16.         LOG.error("Unexpected exception, exiting abnormally", e);
  17.         System.exit(1);
  18.     }
  19.     LOG.info("Exiting normally");
  20.     System.exit(0);
  21. }

org.apache.zookeeper.server.quorum.QuorumPeerMain#initializeAndRun

  1. protected void initializeAndRun(String[] args)throws ConfigException, IOException
  2. {
  3.     // 读取zoo.cfg配置参数
  4.     QuorumPeerConfig config = new QuorumPeerConfig();
  5.     if (args.length == 1) {
  6.         config.parse(args[0]);
  7.     }
  9.     // Start and schedule the the purge task
  10.     // 启动日志清楚任务
  11.     DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config
  12.             .getDataDir(), config.getDataLogDir(), config
  13.             .getSnapRetainCount(), config.getPurgeInterval());
  14.     purgeMgr.start();
  16.     if (args.length == 1 && config.servers.size() > 0) {
  17.         // 读取配置开始搞事情（进入）
  18.         runFromConfig(config);
  19.     } else {
  20.         LOG.warn("Either no config or no quorum defined in config, running "
  21.                 + " in standalone mode");
  22.         // there is only server in the quorum -- run as standalone
  23.         ZooKeeperServerMain.main(args);
  24.     }
  25. }

org.apache.zookeeper.server.quorum.QuorumPeerMain#runFromConfig

  1.  public void runFromConfig(QuorumPeerConfig config) throws IOException {
  2.   try {
  3.       ManagedUtil.registerLog4jMBeans();
  4.   } catch (JMException e) {
  5.       LOG.warn("Unable to register log4j JMX control", e);
  6.   }
  8.   LOG.info("Starting quorum peer");
  9.   try {
  10.       ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
  11.       // 创建服务端的 socket 实例，有两个实现：
  12.       // 1. NIOServerCnxnFactory; 2. NetyServerCnxnFactory 
  13.       cnxnFactory.configure(config.getClientPortAddress(),
  14.                             config.getMaxClientCnxns());
  15.       // confg读取到的zoo.cfg赋值
  16.       quorumPeer = new QuorumPeer();
  17.       quorumPeer.setClientPortAddress(config.getClientPortAddress());
  18.       quorumPeer.setTxnFactory(new FileTxnSnapLog(
  19.                   new File(config.getDataLogDir()),
  20.                   new File(config.getDataDir())));
  21.       quorumPeer.setQuorumPeers(config.getServers());
  22.       quorumPeer.setElectionType(config.getElectionAlg());
  23.       quorumPeer.setMyid(config.getServerId());
  24.       quorumPeer.setTickTime(config.getTickTime());
  25.       quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
  26.       quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
  27.       quorumPeer.setInitLimit(config.getInitLimit());
  28.       quorumPeer.setSyncLimit(config.getSyncLimit());
  29.       quorumPeer.setQuorumVerifier(config.getQuorumVerifier());
  30.       quorumPeer.setCnxnFactory(cnxnFactory);
  31.       quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
  32.       quorumPeer.setLearnerType(config.getPeerType());
  33.       quorumPeer.setSyncEnabled(config.getSyncEnabled());
  34.       quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());
  36.       // 调用start方法 注意这不是调用线程的start方法(进入)
  37.       quorumPeer.start();
  38.       quorumPeer.join();
  39.   } catch (InterruptedException e) {
  40.       // warn, but generally this is ok
  41.       LOG.warn("Quorum Peer interrupted", e);
  42.   }
  43. }

org.apache.zookeeper.server.quorum.QuorumPeer#start

  1. @Override
  2. public synchronized void start() {
  3.     loadDataBase(); // 先从内存中恢复数据写到文件中
  4.     cnxnFactory.start(); // 启动服务器端 Socket 实现(进入)      
  5.     startLeaderElection(); // 开始选举
  6.     super.start(); // 这才真正调用线程的start方法也就会执行run方法
  7. }

org.apache.zookeeper.server.NIOServerCnxnFactory#run 服务端建立链接

  1. public void run() {
  2.     while (!ss.socket().isClosed()) {
  3.         try {
  4.             selector.select(1000);
  5.             Set<SelectionKey> selected;
  6.             synchronized (this) {
  7.                 selected = selector.selectedKeys();
  8.             }
  9.             ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>(
  10.                     selected);
  11.             Collections.shuffle(selectedList); // 乱序
  12.             for (SelectionKey k : selectedList) {
  13.                 if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
  14.                     SocketChannel sc = ((ServerSocketChannel) k
  15.                             .channel()).accept();
  16.                     InetAddress ia = sc.socket().getInetAddress();
  17.                     int cnxncount = getClientCnxnCount(ia);
  18.                     // 调用zoo.cfg配置的客户端连接数是否超过了
  19.                     if (maxClientCnxns > 0 && cnxncount >= maxClientCnxns){
  20.                         LOG.warn("Too many connections from " + ia
  21.                                  + " - max is " + maxClientCnxns );
  22.                         sc.close();
  23.                     } else {
  24.                         LOG.info("Accepted socket connection from "
  25.                                  + sc.socket().getRemoteSocketAddress());
  26.                         sc.configureBlocking(false);
  27.                         // 监听read事件
  28.                         SelectionKey sk = sc.register(selector,
  29.                                 SelectionKey.OP_READ);
  30.                         NIOServerCnxn cnxn = createConnection(sc, sk);
  31.                         sk.attach(cnxn);
  32.                         addCnxn(cnxn);
  33.                     }
  34.                 } else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
  35.                     NIOServerCnxn c = (NIOServerCnxn) k.attachment();
  36.                     c.doIO(k);
  37.                 } else {
  38.                     if (LOG.isDebugEnabled()) {
  39.                         LOG.debug("Unexpected ops in select "
  40.                                   + k.readyOps());
  41.                     }
  42.                 }
  43.             }
  44.             // 清除，下次之需
  45.             selected.clear();
  46.         } catch (RuntimeException e) {
  47.             LOG.warn("Ignoring unexpected runtime exception", e);
  48.         } catch (Exception e) {
  49.             LOG.warn("Ignoring exception", e);
  50.         }
  51.     }
  52.     closeAll();
  53.     LOG.info("NIOServerCnxn factory exited run method");
  54. }

org.apache.zookeeper.server.quorum.QuorumPeer#startLeaderElection 选举开始

  1. synchronized public void startLeaderElection() {
  2.     try {
  3.         // 投票给自己
  4.         currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
  5.     } catch(IOException e) {
  6.         RuntimeException re = new RuntimeException(e.getMessage());
  7.         re.setStackTrace(e.getStackTrace());
  8.         throw re;
  9.     }
  10.     // 从配置中拿出自己的选举地址
  11.     for (QuorumServer p : getView().values()) {
  12.         if (p.id == myid) {
  13.             myQuorumAddr = p.addr;
  14.             break;
  15.         }
  16.     }
  17.     if (myQuorumAddr == null) {
  18.         throw new RuntimeException("My id " + myid + " not in the peer list");
  19.     }
  20.     if (electionType == 0) {
  21.         try {
  22.             udpSocket = new DatagramSocket(myQuorumAddr.getPort());
  23.             responder = new ResponderThread();
  24.             responder.start();
  25.         } catch (SocketException e) {
  26.             throw new RuntimeException(e);
  27.         }
  28.     }
  29.     // 选举开始
  30.     this.electionAlg = createElectionAlgorithm(electionType);
  31. }

org.apache.zookeeper.server.quorum.FastLeaderElection#starter 选举初始化

  1. private void starter(QuorumPeer self, QuorumCnxManager manager) {
  2.     this.self = self;
  3.     proposedLeader = -1;
  4.     proposedZxid = -1;
  6.     sendqueue = new LinkedBlockingQueue<ToSend>();
  7.     recvqueue = new LinkedBlockingQueue<Notification>();
  8.     this.messenger = new Messenger(manager);
  9. }

org.apache.zookeeper.server.quorum.QuorumPeer#run选举开始
这就不贴代码了

2. 客户端

image.png

  1. public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
  2.         boolean canBeReadOnly)
  3.     throws IOException
  4. {
  5.     LOG.info("Initiating client connection, connectString=" + connectString
  6.             + " sessionTimeout=" + sessionTimeout + " watcher=" + watcher);
  8.     watchManager.defaultWatcher = watcher;
  10.     ConnectStringParser connectStringParser = new ConnectStringParser(
  11.             connectString);
  12.     HostProvider hostProvider = new StaticHostProvider(
  13.             connectStringParser.getServerAddresses());//拿到ip端口号
  14.     cnxn = new ClientCnxn(connectStringParser.getChrootPath(),
  15.             hostProvider, sessionTimeout, this, watchManager,
  16.             getClientCnxnSocket(), canBeReadOnly);//创建ClientCnxn对象
  17.     cnxn.start();//非thread线程启动
  18. }

org.apache.zookeeper.ClientCnxn#ClientCnxn初始化 启动了两个线程 send和event

  1. public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
  2.         ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,
  3.         long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {
  4.     this.zooKeeper = zooKeeper;
  5.     this.watcher = watcher;
  6.     this.sessionId = sessionId;
  7.     this.sessionPasswd = sessionPasswd;
  8.     this.sessionTimeout = sessionTimeout;
  9.     this.hostProvider = hostProvider;
  10.     this.chrootPath = chrootPath;
  12.     connectTimeout = sessionTimeout / hostProvider.size();
  13.     readTimeout = sessionTimeout * 2 / 3;
  14.     readOnly = canBeReadOnly;
  16.     sendThread = new SendThread(clientCnxnSocket);
  17.     eventThread = new EventThread();
  19. }

org.apache.zookeeper.ClientCnxn#start连个线程start>run方法

  1. public void start() {
  2.     sendThread.start();
  3.     eventThread.start();
  4. }

org.apache.zookeeper.ClientCnxn.SendThread#run

  1. @Override
  2. public void run() {
  3.     clientCnxnSocket.introduce(this,sessionId);
  4.     clientCnxnSocket.updateNow();
  5.     clientCnxnSocket.updateLastSendAndHeard(); //客户端和服务端链接的socket更新
  6.     int to;
  7.     long lastPingRwServer = System.currentTimeMillis();
  8.     final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds
  9.     while (state.isAlive()) {
  10.         try {
  11.             if (!clientCnxnSocket.isConnected()) {
  12.                 if(!isFirstConnect){
  13.                     try {
  14.                         Thread.sleep(r.nextInt(1000));
  15.                     } catch (InterruptedException e) {
  16.                         LOG.warn("Unexpected exception", e);
  17.                     }
  18.                 }
  19.                 // don't re-establish connection if we are closing
  20.                 if (closing || !state.isAlive()) {
  21.                     break;
  22.                 }
  23.                 startConnect();
  24.                 clientCnxnSocket.updateLastSendAndHeard();
  25.             }
  27.             if (state.isConnected()) {
  28.                 // determine whether we need to send an AuthFailed event.
  29.                 if (zooKeeperSaslClient != null) {
  30.                     boolean sendAuthEvent = false;
  31.                     if (zooKeeperSaslClient.getSaslState() == ZooKeeperSaslClient.SaslState.INITIAL) {
  32.                         try {
  33.                             zooKeeperSaslClient.initialize(ClientCnxn.this);
  34.                         } catch (SaslException e) {
  35.                            LOG.error("SASL authentication with Zookeeper Quorum member failed: " + e);
  36.                             state = States.AUTH_FAILED;
  37.                             sendAuthEvent = true;
  38.                         }
  39.                     }
  40.                     KeeperState authState = zooKeeperSaslClient.getKeeperState();
  41.                     if (authState != null) {
  42.                         if (authState == KeeperState.AuthFailed) {
  43.                             // An authentication error occurred during authentication with the Zookeeper Server.
  44.                             state = States.AUTH_FAILED;
  45.                             sendAuthEvent = true;
  46.                         } else {
  47.                             if (authState == KeeperState.SaslAuthenticated) {
  48.                                 sendAuthEvent = true;
  49.                             }
  50.                         }
  51.                     }
  53.                     if (sendAuthEvent == true) {
  54.                         eventThread.queueEvent(new WatchedEvent(
  55.                               Watcher.Event.EventType.None,
  56.                               authState,null));
  57.                     }
  58.                 }
  59.                 to = readTimeout - clientCnxnSocket.getIdleRecv();
  60.             } else {
  61.                 to = connectTimeout - clientCnxnSocket.getIdleRecv();
  62.             }
  64.             if (to <= 0) {
  65.                 String warnInfo;
  66.                 warnInfo = "Client session timed out, have not heard from server in "
  67.                     + clientCnxnSocket.getIdleRecv()
  68.                     + "ms"
  69.                     + " for sessionid 0x"
  70.                     + Long.toHexString(sessionId);
  71.                 LOG.warn(warnInfo);
  72.                 throw new SessionTimeoutException(warnInfo);
  73.             }
  74.             if (state.isConnected()) { 
  75.                //1000(1 second) is to prevent race condition missing to send the second ping
  76.                //also make sure not to send too many pings when readTimeout is small 
  77.                 int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() - 
  78.                       ((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);
  79.                 //send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL
  80.                 if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {
  81.                     sendPing();//发送心跳
  82.                     clientCnxnSocket.updateLastSend();
  83.                 } else {
  84.                     if (timeToNextPing < to) {
  85.                         to = timeToNextPing;
  86.                     }
  87.                 }
  88.             }
  90.             // If we are in read-only mode, seek for read/write server
  91.             if (state == States.CONNECTEDREADONLY) {
  92.                 long now = System.currentTimeMillis();
  93.                 int idlePingRwServer = (int) (now - lastPingRwServer);
  94.                 if (idlePingRwServer >= pingRwTimeout) {
  95.                     lastPingRwServer = now;
  96.                     idlePingRwServer = 0;
  97.                     pingRwTimeout =
  98.                         Math.min(2*pingRwTimeout, maxPingRwTimeout);
  99.                     pingRwServer();
  100.                 }
  101.                 to = Math.min(to, pingRwTimeout - idlePingRwServer);
  102.             }
  104.             clientCnxnSocket.doTransport(to, pendingQueue, outgoingQueue, ClientCnxn.this);
  105. //这个方法比较长 重点看这
  106.         } catch (Throwable e) {
  107.             if (closing) {
  108.                 if (LOG.isDebugEnabled()) {
  109.                     // closing so this is expected
  110.                     LOG.debug("An exception was thrown while closing send thread for session 0x"
  111.                             + Long.toHexString(getSessionId())
  112.                             + " : " + e.getMessage());
  113.                 }
  114.                 break;
  115.             } else {
  116.                 // this is ugly, you have a better way speak up
  117.                 if (e instanceof SessionExpiredException) {
  118.                     LOG.info(e.getMessage() + ", closing socket connection");
  119.                 } else if (e instanceof SessionTimeoutException) {
  120.                     LOG.info(e.getMessage() + RETRY_CONN_MSG);
  121.                 } else if (e instanceof EndOfStreamException) {
  122.                     LOG.info(e.getMessage() + RETRY_CONN_MSG);
  123.                 } else if (e instanceof RWServerFoundException) {
  124.                     LOG.info(e.getMessage());
  125.                 } else {
  126.                     LOG.warn(
  127.                             "Session 0x"
  128.                                     + Long.toHexString(getSessionId())
  129.                                     + " for server "
  130.                                     + clientCnxnSocket.getRemoteSocketAddress()
  131.                                     + ", unexpected error"
  132.                                     + RETRY_CONN_MSG, e);
  133.                 }
  134.                 cleanup();
  135.                 if (state.isAlive()) {
  136.                     eventThread.queueEvent(new WatchedEvent(
  137.                             Event.EventType.None,
  138.                             Event.KeeperState.Disconnected,
  139.                             null));
  140.                 }
  141.                 clientCnxnSocket.updateNow();
  142.                 clientCnxnSocket.updateLastSendAndHeard();
  143.             }
  144.         }
  145.     }
  146.     cleanup();
  147.     clientCnxnSocket.close();
  148.     if (state.isAlive()) {
  149.         eventThread.queueEvent(new WatchedEvent(Event.EventType.None,
  150.                 Event.KeeperState.Disconnected, null));
  151.     }
  152.     ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),
  153.             "SendThread exited loop for session: 0x"
  154.                    + Long.toHexString(getSessionId()));
  155. }

org.apache.zookeeper.ClientCnxnSocketNIO#doTransport 真正干事的

  1. @Override
  2. void doTransport(int waitTimeOut, List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue,
  3.                  ClientCnxn cnxn)
  4.         throws IOException, InterruptedException {
  5.     selector.select(waitTimeOut);
  6.     Set<SelectionKey> selected;
  7.     synchronized (this) {
  8.         selected = selector.selectedKeys();
  9.     }
  10.     // Everything below and until we get back to the select is
  11.     // non blocking, so time is effectively a constant. That is
  12.     // Why we just have to do this once, here
  13.     updateNow();
  14.     for (SelectionKey k : selected) {
  15.         SocketChannel sc = ((SocketChannel) k.channel());
  16.         if ((k.readyOps() & SelectionKey.OP_CONNECT) != 0) {
  17.             if (sc.finishConnect()) {
  18.                 updateLastSendAndHeard();
  19.                 sendThread.primeConnection();
  20.             }
  21.         } else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
  22.             doIO(pendingQueue, outgoingQueue, cnxn);//这是处理客户端往服务端发送的数据 链接之后会处理读和写操作 这不往下跟代码了        }
  23.     }
  24.     if (sendThread.getZkState().isConnected()) {
  25.         synchronized(outgoingQueue) {
  26.             if (findSendablePacket(outgoingQueue,
  27.                     cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {
  28.                 enableWrite();
  29.             }
  30.         }
  31.     }
  32.     selected.clear();
  33. }

如果是回调函数怎么处理了 org.apache.zookeeper.ClientCnxn.SendThread#run
里面调用了org.apache.zookeeper.ClientCnxn.EventThread#queueEvent这个是往event队列放数据的。

org.apache.zookeeper.ClientCnxn.EventThread#run 这个就是从队列里面取数据了

  1. public void run() {
  2.    try {
  3.       isRunning = true;
  4.       while (true) {
  5.          Object event = waitingEvents.take();
  6.          if (event == eventOfDeath) {
  7.             wasKilled = true;
  8.          } else {
  9.             processEvent(event);
  10.          }
  11.          if (wasKilled)
  12.             synchronized (waitingEvents) {
  13.                if (waitingEvents.isEmpty()) {
  14.                   isRunning = false;
  15.                   break;
  16.                }
  17.             }
  18.       }
  19.    } catch (InterruptedException e) {
  20.       LOG.error("Event thread exiting due to interruption", e);
  21.    }
  23.     LOG.info("EventThread shut down for session: 0x{}",
  24.              Long.toHexString(getSessionId()));
  25. }