Zookeeper start class position in the org. Apache. Zookeeper. Server ZooKeeperServerMain, yes, find it, and run the Main method, can start the zookeeper server.
Note that only one ZooKeeper server was started in my environment, so it is not a clustered environment.
Load the configuration
The first step is to load the configuration file, so let’s look at the initializeAndRun method.
protected void initializeAndRun(String[] args)throws ConfigException, IOException{
ServerConfig config = new ServerConfig();
if (args.length == 1) {
config.parse(args[0]);
} else {
config.parse(conf);
}
runFromConfig(config);
}
Copy the codeThe main thing here is to load the zoo.cfg configuration into the ServerConfig object. The process is relatively simple and will not be described here. Let’s start with a few simple configuration items.
| configuration | meaning |
|---|---|
| clientPort | External service port, generally 2181 |
| dataDir | The directory where snapshot files are stored, as well as transaction log files by default |
| tickTime | A unit of time in ZK. All times in ZK are configured in integer multiples based on this time unit |
| minSessionTimeout maxSessionTimeout | Session timeout duration. The default value is 2tickTime ~ 20Between the tickTime |
| preAllocSize | Preset disk space for subsequent writing to the transaction log, default 64 MB |
| snapCount | After each snapCount transaction log output, a snapshot is triggered. The default value is 100,000 |
| maxClientCnxns | The maximum number of concurrent clients is 60 by default |
2. Start the service
Let’s move on to the runFromConfig method.
public void runFromConfig(ServerConfig config) throws IOException {
LOG.info("Starting server"); FileTxnSnapLog txnLog = null; try { final ZooKeeperServer zkServer = new ZooKeeperServer(); final CountDownLatch shutdownLatch = new CountDownLatch(1); / / register server shutdown event zkServer registerServerShutdownHandler (new ZooKeeperServerShutdownHandler (shutdownLatch)); TxnLog = new FileTxnSnapLog(new File(config.datalogdir), new File(config.datadir)); txnLog.setServerStats(zkServer.serverStats()); / / set the configuration properties zkServer. SetTxnLogFactory (txnLog); zkServer.setTickTime(config.tickTime); zkServer.setMinSessionTimeout(config.minSessionTimeout); zkServer.setMaxSessionTimeout(config.maxSessionTimeout); / / instantiate ServerCnxnFactory abstract class cnxnFactory = ServerCnxnFactory. CreateFactory (); cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns()); cnxnFactory.startup(zkServer); shutdownLatch.await(); shutdown(); cnxnFactory.join();if (zkServer.canShutdown()) {
zkServer.shutdown(true);
}
} catch (InterruptedException e) {
LOG.warn("Server interrupted", e);
} finally {
if(txnLog ! = null) { txnLog.close(); }}}Copy the codeThis is the zooKeeper server startup and shutdown process. Let’s break it down.
1. Service shutdown event
We see a server shutdown handler class registered with zkServer.
final ZooKeeperServer zkServer = new ZooKeeperServer();
final CountDownLatch shutdownLatch = new CountDownLatch(1);
zkServer.registerServerShutdownHandler(
new ZooKeeperServerShutdownHandler(shutdownLatch));
Copy the codeFirst, we should know that the ZooKeeper server is stateful.
protected enum State {
INITIAL, RUNNING, SHUTDOWN, ERROR;
}
Copy the codeThen, when the state changes, the setState method is called.
Public class ZooKeeperServer{// Call this method protected void when the zooKeeper server status changessetState(State state) {
this.state = state;
if(zkShutdownHandler ! = null) { zkShutdownHandler.handle(state); }else {
LOG.debug("ZKShutdownHandler is not registered, so ZooKeeper server "
+ "won't take any action on ERROR or SHUTDOWN server state changes"); }}}Copy the codeThis is where the registered handler is called. In the processor, the Shutdownlatch.await method is awakened if the status is found to be incorrect.
class ZooKeeperServerShutdownHandler {
void handle(State state) {
if(state == State.ERROR || state == State.SHUTDOWN) { shutdownLatch.countDown(); }}}Copy the codeWhen it is awakened, things are easy. Close and clean up various resources.
2. Log files
The operation of transaction log files and snapshot files, respectively, corresponds to two implementation classes, in this case to create file paths and class instances.
public FileTxnSnapLog(File dataDir, File snapDir) throws IOException {
LOG.debug("Opening datadir:{} snapDir:{}", dataDir, snapDir);
this.dataDir = new File(dataDir, version + VERSION);
this.snapDir = new File(snapDir, version + VERSION);
if(! this.dataDir.exists()) {if(! this.dataDir.mkdirs()) { throw new IOException("Unable to create data directory "+ this.dataDir); }}if(! this.dataDir.canWrite()) { throw new IOException("Cannot write to data directory " + this.dataDir);
}
if(! this.snapDir.exists()) {if(! this.snapDir.mkdirs()) { throw new IOException("Unable to create snap directory "+ this.snapDir); }}if(! this.snapDir.canWrite()) { throw new IOException("Cannot write to snap directory " + this.snapDir);
}
if(! this.dataDir.getPath().equals(this.snapDir.getPath())){ checkLogDir(); checkSnapDir(); } txnLog = new FileTxnLog(this.dataDir); snapLog = new FileSnap(this.snapDir); }Copy the codeCreate a file if it doesn’t exist and check if you have write permission.
If two files have different paths, call checkLogDir and checkSnapDir to check. What to check? They just don’t fit together.
The transaction log file directory cannot contain snapshot files. The snapshot file directory cannot contain transaction log files.
Finally, you initialize the two implementation classes and tell them about the file object that was created.
3. Start the service
The startup of the server corresponds to two implementations: NIO server and Netty server. So you start by calling createFactory to select an implementation class to instantiate.
static public ServerCnxnFactory createFactory() throws IOException {
String serverCnxnFactoryName =
System.getProperty("zookeeper.serverCnxnFactory");
if (serverCnxnFactoryName == null) {
serverCnxnFactoryName = NIOServerCnxnFactory.class.getName();
}
try {
ServerCnxnFactory serverCnxnFactory = Class.forName(serverCnxnFactoryName)
.getDeclaredConstructor().newInstance();
return serverCnxnFactory;
} catch (Exception e) {
IOException ioe = new IOException("Couldn't instantiate "+ serverCnxnFactoryName); ioe.initCause(e); throw ioe; }}Copy the codeObtains a zookeeper. ServerCnxnFactory attribute value, if it is empty, the default is created NIOServerCnxnFactory instance.
So, if we want to use Netty start, can such Settings: System. SetProperty (” zookeeper. ServerCnxnFactory, “NettyServerCnxnFactory. Class. GetName ());
Finally, their constructors are retrieved by reflection and instantiated. Then call their methods to bind the ports and start the service. There is not much difference, so let’s take Netty as an example.
- The constructor
NettyServerCnxnFactory() {
bootstrap = new ServerBootstrap(
new NioServerSocketChannelFactory(
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
bootstrap.setOption("reuseAddress".true);
bootstrap.setOption("child.tcpNoDelay".true);
bootstrap.setOption("child.soLinger", 1); bootstrap.getPipeline().addLast("servercnxnfactory", channelHandler);
}
Copy the codeIn the constructor, initialize the ServerBootstrap object and set the TCP parameters. We’ll focus on its event handler, channelHandler.
- Event handler
The channelHandler here is an inner class that inherits from SimpleChannelHandler. It’s labeled @sharable, and it’s a shared processor.
@sharable Class CnxnChannelHandler extends SimpleChannelHandler {// The client connection is closed public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)throws Exception{// Remove the corresponding Channel allchannels.remove (ctx.getChannel()); } // Client connection public void channelConnected(ChannelHandlerContext CTX,ChannelStateEvent e) throws Exception{ allChannels.add(ctx.getChannel()); NettyServerCnxn cnxn = new NettyServerCnxn(ctx.getChannel(), zkServer, NettyServerCnxnFactory.this); ctx.setAttachment(cnxn); addCnxn(cnxn); Public void channelDisconnected(ChannelHandlerContext CTX,ChannelStateEvent e) throws Exception{ NettyServerCnxn cnxn = (NettyServerCnxn) ctx.getAttachment();if(cnxn ! = null) { cnxn.close(); }} public void exceptionCaught(ChannelHandlerContext CTX, ExceptionEvent e)throws Exception{ NettyServerCnxn cnxn = (NettyServerCnxn) ctx.getAttachment();if(cnxn ! = null) {if (LOG.isDebugEnabled()) {
LOG.debug("Closing "+ cnxn); } cnxn.close(); }} public void messageReceived(ChannelHandlerContext CTX, Throws Exception{try {// Find the corresponding NettyServerCnxn, Call method handling request information NettyServerCnxn CNXN = (NettyServerCnxn)ctx.getAttachment(); synchronized(cnxn) { processMessage(e, cnxn); } } catch(Exception ex) { LOG.error("Unexpected exception in receive", ex); throw ex; Private void processMessage(MessageEvent e, NettyServerCnxn CNXN) {.... Omit}}Copy the codeThis is where you handle all kinds of IO events. Such as client connection, disconnection, readable message…
Let’s look at the messageReceived method. This method is called when there is a message request. It finds the NettyServerCnxn object corresponding to the current Channel and calls its receiveMessage method to complete the processing of the specific request.
- Binding port
After the initialization, use bootstrap.bind to bind the port to provide external services.
public class NettyServerCnxnFactory extends ServerCnxnFactory {
public void start() {
LOG.info("binding to port " + localAddress);
parentChannel = bootstrap.bind(localAddress); }}Copy the codeWe called the start method to start the Netty service, but the whole ZooKeeper startup process is not complete.
public void startup(ZooKeeperServer zks) throws IOException,InterruptedException {
start();
setZooKeeperServer(zks);
zks.startdata();
zks.startup();
}
Copy the codeLoad data
Next we look at zks.startData (); It loads data from the ZooKeeper database.
Some students can not help but have questions, what, ZK actually has a database? Take your time. Let’s take our time.
public class ZooKeeperServer implements SessionExpirer, Serverstats. Provider {// Load data public void startData ()throws IOException, InterruptedException {// zkDb is null when the server is started.if(zkDb == null) { zkDb = new ZKDatabase(this.txnLogFactory); } // Load dataif(! zkDb.isInitialized()) { loadData(); }}}Copy the codeIn the above code, zkDb is empty at startup, so the first condition is checked and the constructor is called to initialize zkDb. The loadData method is then called to load the data.
1, ZKDatabase
In fact, ZooKeeper doesn’t have a database, just a class called ZKDatabase, or in-memory database. Let’s look at what properties it has.
Public class ZKDatabase {// protected DataTree DataTree; //Session timeout Session protected ConcurrentHashMap<Long, Integer> sessionsWithTimeouts; Log protected FileTxnSnapLog snapLog; Protected long minCommittedLog, maxCommittedLog; public static final int commitLogCount = 500; protected static int commitLogBuffer = 700; // Protected LinkedList<Proposal> committedLog = new LinkedList<Proposal>(); protected ReentrantReadWriteLocklogLock = new ReentrantReadWriteLock(); Volatile private Boolean initialized = volatile private Boolean initialized =false;
}
Copy the codeThese include sessions, data trees, and commit logs. All data is stored in a DataTree, a DataTree, which holds all node data.
Public class DataTree {// Hash tables provide quick lookup of data nodes private Final ConcurrentHashMap<String, DataNode> nodes = new ConcurrentHashMap<String, DataNode>(); //Watcher related private final WatchManager dataWatches = new WatchManager(); private final WatchManager childWatches = new WatchManager(); // Default zooKeeper node private static Final String rootZookeeper ="/";
private static final String procZookeeper = "/zookeeper";
private static final String procChildZookeeper = procZookeeper.substring(1);
private static final String quotaZookeeper = "/zookeeper/quota";
private static final String quotaChildZookeeper = quotaZookeeper
.substring(procZookeeper.length() + 1);
}
Copy the codeWhen we query node data from ZooKeeper, it is through DataTree method to obtain. To be specific, go to the nodes hash table through the node name to query. Such as:
public byte[] getData(String path, Stat stat, Watcher watcher){
DataNode n = nodes.get(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
n.copyStat(stat);
if(watcher ! = null) { dataWatches.addWatch(path, watcher); }returnn.data; }}Copy the codeThen we probably already know that datanodes are the real carriers of data.
Public class DataNode implements Record {// DataNode parent; Byte data[]; // Permission information Long acl; Public StatPersisted node statisticsstat; Private Set<String> children = null; Private static final Set<String> EMPTY_SET = collections.emptySet (); }Copy the codeEach node in ZooKeeper corresponds to a DataNode object. It contains a collection of parent and child nodes, permission information, node data content, and statistics, all represented in this class.
2. Instantiate the object
So let’s go back and look at the code. If zkDb is empty, instantiate it.
public ZKDatabase(FileTxnSnapLog snapLog) {
dataTree = new DataTree();
sessionsWithTimeouts = new ConcurrentHashMap<Long, Integer>();
this.snapLog = snapLog;
}
Copy the codeHere we instantiate the DataTree object, initialize the Map of the timeout session, and assign the snapLog object.
In the DataTree constructor, the default ZooKeeper node is initialized by adding DataNode objects to the Nodes hash table.
public DataTree() {
nodes.put("", root);
nodes.put(rootZookeeper, root);
root.addChild(procChildZookeeper);
nodes.put(procZookeeper, procDataNode);
procDataNode.addChild(quotaChildZookeeper);
nodes.put(quotaZookeeper, quotaDataNode);
}
Copy the code3. Load data
If zkDb has not been initialized, load the database, set it to initialized, and clean up expired sessions.
public class ZooKeeperServer{
public void loadData() throws IOException, InterruptedException {
if(zkDb.isInitialized()){
setZxid(zkDb.getDataTreeLastProcessedZxid());
}
else {
setZxid(zkDb.loadDataBase()); } // deadSessions <Long> deadSessions = new LinkedList<Long>();for (Long session : zkDb.getSessions()) {
if (zkDb.getSessionWithTimeOuts().get(session) == null) {
deadSessions.add(session);
}
}
zkDb.setDataTreeInit(true);
for (long session : deadSessions) {
killSession(session, zkDb.getDataTreeLastProcessedZxid()); }}}Copy the codeLet’s look at the zkdb.loadDatabase () method. It will load the database from a disk file.
Public class ZKDatabase {// Load the database from a disk file, Public Long loadDataBase() throws IOException {long zxID = snaplog. restore(dataTree, s essionsWithTimeouts, commitProposalPlaybackListener); initialized =true;
returnzxid; }}Copy the codeSince they are disk files, they must be snapshot files and transaction log files. Snaplog. restore will confirm this.
public class FileTxnSnapLog { public long restore(DataTree dt, Map<Long, Integer> sessions, PlayBackListener Listener) throws IOException {// Load data from the snapshot file snapLog. Deserialize (dt, sessions); Long fastForwardFromEdits = fastForwardFromEdits(dt, sessions, listener);returnfastForwardFromEdits; }}Copy the codeThe process of loading the data may seem complicated, but at its core, it reads the data from the file stream, converts it into a DataTree object, and puts it into zkDb. Here, let’s not look at the process of parsing the file, just to see what is stored in the file?
The snapshot file
We find the org. Apache. Zookeeper. Server. SnapshotFormatter, it can help us to output the snapshot file content. In the main method, set the path to the snapshot file and run it.
Public class SnapshotFormatter {public static void main(String[] args) throws Exception {// Sets the snapshot file path args = new String[1]; args[0] ="E:\\zookeeper-data\\version-2\\snapshot.6";
if(args.length ! = 1) { System.err.println("USAGE: SnapshotFormatter snapshot_file"); System.exit(2); } new SnapshotFormatter().run(args[0]); }}Copy the codeRunning the main method outputs the contents of the snapshot file on the console.
ZNode Details (count=8): ---- / cZxid = 0x00000000000000 ctime = Thu Jan 01 08:00:00 CST 1970 mZxid = 0x00000000000000 mtime = Thu Jan 01 08:00:00 CST 1970 pZxid = 0x00000000000002 cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x00000000000000 dataLength = 0 ---- /zookeeper cZxid = 0x00000000000000 ctime = Thu Jan 01 08:00:00 CST 1970 mZxid = 0x00000000000000 mtime = Thu Jan 01 08:00:00 CST 1970 pZxid = 0x00000000000000 cversion = 0 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x00000000000000 dataLength = 0 ---- /zookeeper/quota cZxid = 0x00000000000000 ctime = Thu Jan 01 08:00:00 CST 1970 mZxid = 0x00000000000000 mtime = Thu Jan 01 08:00:00 CST 1970 pZxid = 0x00000000000000 cversion = 0 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x00000000000000 dataLength = 0 ---- /testcZxid = 0x00000000000002 ctime = Sat Feb 23 19:57:43 CST 2019 mZxid = 0x00000000000002 mtime = Sat Feb 23 19:57:43 CST 2019 pZxid = 0x00000000000005 cversion = 3 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x00000000000000 dataLength = 4 - /test/t1
cZxid = 0x00000000000003
ctime = Sat Feb 23 19:57:53 CST 2019
mZxid = 0x00000000000003
mtime = Sat Feb 23 19:57:53 CST 2019
pZxid = 0x00000000000003
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x00000000000000
dataLength = 4
----
/test/t2
cZxid = 0x00000000000004
ctime = Sat Feb 23 19:57:56 CST 2019
mZxid = 0x00000000000004
mtime = Sat Feb 23 19:57:56 CST 2019
pZxid = 0x00000000000004
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x00000000000000
dataLength = 4
----
/test/t3
cZxid = 0x00000000000005
ctime = Sat Feb 23 19:57:58 CST 2019
mZxid = 0x00000000000005
mtime = Sat Feb 23 19:57:58 CST 2019
pZxid = 0x00000000000005
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x00000000000000
dataLength = 4
----
Session Details (sid, timeout, ephemeralCount):
0x10013d3939a0000, 99999, 0
0x10013d1adcb0000, 99999, 0
Copy the codeYou can see that each line in the snapshot file after formatting is a DataNode except for the count information at the beginning and Session information at the end. From the node name you extrude the parent node and child node, and the other is the statistics object from the node, StatPersisted.
Transaction log file
We find the org. Apache. Zookeeper. Server. LogFormatter this class, set up the transaction log file path in the main method, and then run it. Every transaction in ZooKeeper is recorded.
19-2-23 7:57.32sec session 0x10013D1AdCB0000 CXID 0x0 zxID 0x1 createSession 99999 19-2-23 7:57.43sec session 0x10013d1adcb0000 cxid 0x2 zxid 0x2 create'/test,#31323334,v{s{31,s{'world,Session 0x10013D1ADCB0000 CXID 0x3 zxID 0x3 create '/test/t1,#31323334,v{s{31,s{'world,'anyone}}},F,119-2-23 7:57pm 56sec Session 0x10013D1ADCB0000 CXID 0x4 zxID 0x4 Create'/test/t2,#31323334,v{s{31,s{'world,Session 0x10013D1ADCB0000 CXID 0x5 zxID 0x5 create '/test/t3,#31323334,v{s{31,s{'world,'anyone}}},F,319-2-23 7:58:51 PM Session 0x10013D3939A0000 CXID 0x0 zxID 0x6 createSession 99999 19-2-23 7:59:07 PM Session 0x10013d3939a0000 cxid 0x4 zxid 0x7 create'/test/t4,#31323334,v{s{31,s{'world,'anyone}}},F,4
Copy the codeAs you can see, each transaction corresponds to a row of records. The operation time, sessionId, transaction ID, operation type, node name, and permission information are included. It is important to note that only change operations are recorded in the transaction log. So, we don’t see any read requests here.
Session manager
Sessions are an important abstraction in Zookeeper. Ensure that requests are ordered, temporary ZNode nodes, and listening points are closely related to the session. An important task of the Zookeeper server is to track and maintain these sessions.
In ZooKeeper, the server is responsible for cleaning up expired sessions, while the client can only keep itself active with heartbeat information or a new read/write request.
The management of expired sessions relies on the bucket policy. The details are as follows:
- Zookeeper sets an expiration time for each session, which we call nextExpirationTime
- 2. Put this expiration time and the corresponding Session collection into the Map
- Open the thread, rotate the Map, extract the Session collection of the current expiration point nextExpirationTime, and close them
- 4. An inactive Session is closed. An active Session recalculates its expiration time and changes its expiration time nextExpirationTime to ensure that it cannot be scanned by threads
In short, an active Session keeps resetting its nextExpirationTime so it can’t be scanned by a thread and shut down.
Zks.startup (); Method, how to do it.
public class ZooKeeperServer
public synchronized void startup() {
if (sessionTracker == null) {
createSessionTracker();
}
startSessionTracker();
setupRequestProcessors();
registerJMX();
setState(State.RUNNING); notifyAll(); }}Copy the codeWe will focus only on the createSessionTracker and startSessionTracker methods, which are session specific.
1. Create a session tracker
Create a session tracker, here is an instance of the SessionTrackerImpl object.
protected void createSessionTracker() {
sessionTracker = new SessionTrackerImpl(this, zkDb.getSessionWithTimeOuts(),
tickTime, 1, getZooKeeperServerListener());
}
Copy the codeIn the constructor, we do some parameter initialization.
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());
this.nextSessionId = initializeNextSession(sid);
for(Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) { addSession(e.getKey(), e.getValue()); }}Copy the codeLet’s focus on how the nextExpirationTime nextExpirationTime is calculated. Let’s look at the roundToInterval method.
private long roundToInterval(long time) {
return (time / expirationInterval + 1) * expirationInterval;
}
Copy the codeWhere, time is a timestamp based on the current time; ExpirationInterval is the tickTime in our configuration file. If we assume time=10 and expirationInterval=2, then the next expiration time calculated above is (10/2+1)*2=12
This means that the current Session will be allocated to bucket 12. Let’s go ahead and watch this process. In the addSession method, first check whether there is a session Id SessionImpl, if not, create and save.
synchronized public void addSession(long id, int sessionTimeout) { sessionsWithTimeout.put(id, sessionTimeout); // Query the Impl class corresponding to the SessionIdif (sessionsById.get(id) == null) {
SessionImpl s = new SessionImpl(id, sessionTimeout, 0);
sessionsById.put(id, s);
} else {
if (LOG.isTraceEnabled()) {
ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
"SessionTrackerImpl --- Existing session 0x"
+ Long.toHexString(id) + "" + sessionTimeout);
}
}
touchSession(id, sessionTimeout);
}
Copy the codeFinally, touchSession is called to activate the session. Note that this method is called for every request in ZooKeeper. It calculates the next expiration of an active Session and migrates it to a different bucket.
We talk about “buckets” all the time, and maybe it’s hard to understand what a bucket is. In code, it’s actually a HashSet object.
Public class SessionTrackerImpl{public class SessionTrackerImpl{public class SessionTrackerImpl{public class SessionTrackerImpl{public class SessionTrackerImpl{ SessionSet>(); Sessions = new HashSet<SessionImpl>(); } synchronized public boolean touchSession(long sessionId, int timeout) { SessionImpl s = sessionsById.get(sessionId); // If the session is deleted or is already marked as closedif (s == null || s.isClosing()) {
return false; } // Calculate the next expiration Time long expireTime = roundToInterval(time.currentelapsedTime () + timeout);if (s.tickTime >= expireTime) {
return true; } // Get the current Session expiration time SessionSetset = sessionSets.get(s.tickTime);
if (set! = null) {// Delete set.sessions. Remove (s); } // Set a new expiration time and add it to the Session set.set = sessionSets.get(s.tickTime);
if (set == null) {
set = new SessionSet();
sessionSets.put(expireTime, set);
}
set.sessions.add(s);
return true; }}Copy the codeGoing back to the formula above, if the first Session request is evaluated with an expiration time of 12. So, the corresponding mapping Session are as follows: 12 = org. Apache. Zookeeper. Server SessionTrackerImpl $SessionSet @ 25143 a5e second request, calculation after the expiration time is 15. Becomes: 15 = org. Apache. Zookeeper. Server SessionTrackerImpl $SessionSet @ 3045314 d
At the same time, records with an expiration time of 12 are deleted. In this way, the location of this Session is continuously migrated through expiration time changes. We can assume that if the request does not reach the server for a long time due to network reasons or the client is feigned dead, the expiration time of the corresponding Session will not change. Times change; if you do not change, you will be abandoned. ** also applies to zooKeeper sessions.
Let’s move on to startSessionTracker();
protected void startSessionTracker() {
((SessionTrackerImpl)sessionTracker).start();
}
Copy the codeSessionTrackerImpl inherits from ZooKeeperCriticalThread, so it is itself a thread class. After calling the start method to start the thread, let’s look at the run method.
synchronized public void run() {
try {
while (running) {
currentTime = Time.currentElapsedTime();
if (nextExpirationTime > currentTime) {
this.wait(nextExpirationTime - currentTime);
continue;
}
SessionSet set; // Get the Session set corresponding to the expiration timeset= sessionSets.remove(nextExpirationTime); // Loop sessions and close themif (set! = null) {for (SessionImpl s : set.sessions) {
setSessionClosing(s.sessionId);
expirer.expire(s);
}
}
nextExpirationTime += expirationInterval;
}
} catch (InterruptedException e) {
handleException(this.getName(), e);
}
LOG.info("SessionTrackerImpl exited loop!");
}
Copy the codeThis method continuously retrieves the Session collection corresponding to the expiration time in an infinite loop. It’s a case of catch and bust. This explains why an active Session must constantly change its expiration date, because someone is watching.
The last step is to register JMX and set up the running state of the server.
Five, the summary
This article mainly analyzes the specific process of zooKeeper server startup. Let’s review it again.
- Configure the zoo. CFG file and run the Main method
- Register zK server shutdown events to clean up resources
- Select a NIO or Netty server bound port to enable the service
- Initialize zkDB and load disk files into memory
- Create a session manager to monitor expired sessions and delete them
- Register JMX and set the ZK service status to RUNNING