Apache ZooKeeper 服务启动源码解释

[root@localhost zookeeper-3.4.7]# cd conf
[root@localhost conf]# ls -rlt
total 12
-rw-rw-r--. 1 1000 1000 922 Nov 10 22:32 zoo_sample.cfg
-rw-rw-r--. 1 1000 1000 2161 Nov 10 22:32 log4j.properties
-rw-rw-r--. 1 1000 1000 535 Nov 10 22:32 configuration.xsl
[root@localhost conf]# cat zoo_sample.cfg
# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1

[root@localhost bin]# ./zkServer.sh
ZooKeeper JMX enabled by default
Using config: /home/zhoumingyao/zookeeper/zookeeper-3.4.7/bin/../conf/zoo.cfg
Usage: ./zkServer.sh {start|start-foreground|stop|restart|status|upgrade|print-cmd}

[root@localhost bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /home/zhoumingyao/zookeeper/zookeeper-3.4.7/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

[root@localhost bin]# jps
2737 QuorumPeerMain
2751 Jps
[root@localhost bin]# ps -ef | grep zookeeper | grep -v grep | awk ‘{print $2}‘
2608

[root@localhost bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /home/zhoumingyao/zookeeper/zookeeper-3.4.7/bin/../conf/zoo.cfg
Mode: standalone

[root@localhost conf]# cat zoo1.cfg
# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=/var/lib/zookeeper/zoo1
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1
server.1=localhost:2666:3666
server.2=localhost:2667:3667
server.3=localhost:2668:3668

[root@localhost conf]# echo 1 > /var/lib/zookeeper/zoo1/myid
[root@localhost conf]# echo 2 > /var/lib/zookeeper/zoo2/myid
[root@localhost conf]# echo 3 > /var/lib/zookeeper/zoo3/myid

[root@localhost bin]# ./zkServer.sh start /home/zhoumingyao/zookeeper/zookeeper-3.4.7/conf/zoo1.cfg
ZooKeeper JMX enabled by default
Using config: /home/zhoumingyao/zookeeper/zookeeper-3.4.7/conf/zoo1.cfg
Starting zookeeper ... STARTED
[root@localhost bin]# ./zkServer.sh start /home/zhoumingyao/zookeeper/zookeeper-3.4.7/conf/zoo2.cfg
ZooKeeper JMX enabled by default
Using config: /home/zhoumingyao/zookeeper/zookeeper-3.4.7/conf/zoo2.cfg
Starting zookeeper ... STARTED
[root@localhost bin]# ./zkServer.sh start /home/zhoumingyao/zookeeper/zookeeper-3.4.7/conf/zoo3.cfg
ZooKeeper JMX enabled by default
Using config: /home/zhoumingyao/zookeeper/zookeeper-3.4.7/conf/zoo3.cfg
Starting zookeeper ... STARTED

[root@localhost bin]# jps
5537 QuorumPeerMain
5617 Jps
5585 QuorumPeerMain

server.1=node1:2889:3889
server.2=node2:2889:3889
server.3=node3:2889:3889

[root@node3 bin]# ./zkServer.sh status
JMX enabled by default
Using config: /usr/lib/zookeeper-3.4.6/bin/../conf/zoo.cfg
Mode: follower

[root@node2 bin]# ./zkServer.sh status
JMX enabled by default
Using config: /usr/lib/zookeeper-3.4.6/bin/../conf/zoo.cfg
Mode: leader

nohup "$JAVA" "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" -cp "$CLASSPATH" $JVMFLAGS $ZOOMAIN "$ZOOCFG" > "$_ZOO_DAEMON_OUT" 2>&1 < /dev/null &

if (purgeInterval <= 0) {
 LOG.info("Purge task is not scheduled.");
 return;
 }
//如果间隔大于 0，启动定时任务机制
timer = new Timer("PurgeTask", true);
 TimerTask task = new PurgeTask(dataLogDir, snapDir, snapRetainCount);
 timer.scheduleAtFixedRate(task, 0, TimeUnit.HOURS.toMillis(purgeInterval));

//首先根据配置创建对应 factory 的实例:NIOServerCnxnFactory 或者 NettyServerCnxnFactory
static public ServerCnxnFactory createFactory() throws IOException {
 String serverCnxnFactoryName =
 System.getProperty(ZOOKEEPER_SERVER_CNXN_FACTORY);
 if (serverCnxnFactoryName == null) {
 serverCnxnFactoryName = NIOServerCnxnFactory.class.getName();
 }
 try {
 return (ServerCnxnFactory) Class.forName(serverCnxnFactoryName)
 .newInstance();
 } catch (Exception e) {
 IOException ioe = new IOException("Couldn‘t instantiate "
 + serverCnxnFactoryName);
 ioe.initCause(e);
 throw ioe;
 }
}

quorumPeer = new QuorumPeer();
 quorumPeer.setClientPortAddress(config.getClientPortAddress());
 quorumPeer.setTxnFactory(new FileTxnSnapLog(
 new File(config.getDataLogDir()),
 new File(config.getDataDir())));
 quorumPeer.setQuorumPeers(config.getServers());

@Override
 public synchronized void start() {
 loadDataBase();
 cnxnFactory.start(); 
 startLeaderElection();
 super.start();
 }

public void addCommittedProposal(Request request) {
 WriteLock wl = logLock.writeLock();
 try {
 wl.lock();
//采用 LinkedList 作为提交日志的存储单元
 if (committedLog.size() > commitLogCount) {//数量限制在 500，如果超过 500，移除第一条
 committedLog.removeFirst();
 minCommittedLog = committedLog.getFirst().packet.getZxid();
 }
 if (committedLog.size() == 0) {
 minCommittedLog = request.zxid;
 maxCommittedLog = request.zxid;
 }

 ByteArrayOutputStream baos = new ByteArrayOutputStream();
 BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
 try {
 request.hdr.serialize(boa, "hdr");
 if (request.txn != null) {
 request.txn.serialize(boa, "txn");
 }
 baos.close();
 } catch (IOException e) {
 LOG.error("This really should be impossible", e);
 }
 QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid,
 baos.toByteArray(), null);
 Proposal p = new Proposal();
 p.packet = pp;
 p.request = request;
 committedLog.add(p);
 maxCommittedLog = p.packet.getZxid();
 } finally {
 wl.unlock();
 }
 }

//读取 EPOCH
long lastProcessedZxid = zkDb.getDataTree().lastProcessedZxid;
//从最新的 zxid 恢复 epoch 变量，zxid64 位，前 32 位是 epoch 值，后 32 位是 zxid 
long epochOfZxid = ZxidUtils.getEpochFromZxid(lastProcessedZxid);

currentEpoch = readLongFromFile(CURRENT_EPOCH_FILENAME);
 LOG.info("currentEpoch="+currentEpoch);
 if (epochOfZxid > currentEpoch && updating.exists()) {
 LOG.info("{} found. The server was terminated after " +
 "taking a snapshot but before updating current " +
 "epoch. Setting current epoch to {}.",
 UPDATING_EPOCH_FILENAME, epochOfZxid);
 setCurrentEpoch(epochOfZxid);
 if (!updating.delete()) {
 throw new IOException("Failed to delete " +
 updating.toString());
 }
 }

case 3:
 qcm = new QuorumCnxManager(this);
 QuorumCnxManager.Listener listener = qcm.listener;
 if(listener != null){
 listener.start();
 le = new FastLeaderElection(this, qcm);
 } else {
 LOG.error("Null listener when initializing cnx manager");
 }
 break;

while (running) {
 LOG.info("getPeerState()="+getPeerState());
 switch (getPeerState()) {
 case LOOKING:
 LOG.info("LOOKING");

 if (Boolean.getBoolean("readonlymode.enabled")) {
 LOG.info("Attempting to start ReadOnlyZooKeeperServer");

 // Create read-only server but don‘t start it immediately
 final ReadOnlyZooKeeperServer roZk = new ReadOnlyZooKeeperServer(
 logFactory, this,
 new ZooKeeperServer.BasicDataTreeBuilder(),
 this.zkDb);

while (!shutdown) {
 Socket client = ss.accept();
 setSockOpts(client);
 LOG.info("Received connection request "
 + client.getRemoteSocketAddress());
 receiveConnection(client);
 numRetries = 0;
 }

while (running && !shutdown && sock != null) {
 /**
 * Reads the first int to determine the length of the
 * message
 */
 int length = din.readInt();
 if (length <= 0 || length > PACKETMAXSIZE) {
 throw new IOException(
 "Received packet with invalid packet: "
 + length);
 }
 /**
 * Allocates a new ByteBuffer to receive the message
 */
 byte[] msgArray = new byte[length];
 din.readFully(msgArray, 0, length);
 ByteBuffer message = ByteBuffer.wrap(msgArray);
 addToRecvQueue(new Message(message.duplicate(), sid));
 }

Vote current = self.getCurrentVote();
 LOG.info("Vote current="+current);
 if(ackstate == QuorumPeer.ServerState.LOOKING){
 if(LOG.isDebugEnabled()){
 LOG.debug("Sending new notification. My id = " +
 self.getId() + " recipient=" +
 response.sid + " zxid=0x" +
 Long.toHexString(current.getZxid()) +
 " leader=" + current.getId());
 }
 
 ToSend notmsg;
 if(n.version > 0x0) {
 notmsg = new ToSend(
 ToSend.mType.notification,
 current.getId(),
 current.getZxid(),
 current.getElectionEpoch(),
 self.getPeerState(),
 response.sid,
 current.getPeerEpoch());
 
 } else {
 Vote bcVote = self.getBCVote();
 notmsg = new ToSend(
 ToSend.mType.notification,
 bcVote.getId(),
 bcVote.getZxid(),
 bcVote.getElectionEpoch(),
 self.getPeerState(),
 response.sid,
 bcVote.getPeerEpoch());
 }
 sendqueue.offer(notmsg);
 }