【Zookeeper源码分析】—服务器(二)之ZooKeeperServer

作者:leesf

出处:https://www.cnblogs.com/leesf456/p/6518040.html


一、前言

前面阐述了服务器的总体框架,下面来分析服务器的所有父类ZooKeeperServer。

二、ZooKeeperServer源码分析

2.1 类的继承关系 

 public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {}

说明:ZooKeeperServer是ZooKeeper中所有服务器的父类,其实现了Session.Expirer和ServerStats.Provider接口,SessionExpirer中定义了expire方法(表示会话过期)和getServerId方法(表示获取服务器ID),而Provider则主要定义了获取服务器某些数据的方法。

2.2 类的内部类

1. DataTreeBuilder类

     public interface DataTreeBuilder {
        // 构建DataTree
        public DataTree build();
    }

说明:其定义了构建树DataTree的接口。

2. BasicDataTreeBuilder类

     static public class BasicDataTreeBuilder implements DataTreeBuilder {
        public DataTree build() {
            return new DataTree();
        }
    }

说明:实现DataTreeBuilder接口,返回新创建的树DataTree。

3. MissingSessionException类


public static class MissingSessionException extends IOException { private static final long serialVersionUID = 7467414635467261007L; public MissingSessionException(String msg) { super(msg); } }

说明:表示会话缺失异常。

4. ChangeRecord类 


static class ChangeRecord { ChangeRecord(long zxid, String path, StatPersisted stat, int childCount, List<ACL> acl) { // 属性赋值 this.zxid = zxid; this.path = path; this.stat = stat; this.childCount = childCount; this.acl = acl; } // zxid long zxid; // 路径 String path; // 统计数据 StatPersisted stat; /* Make sure to create a new object when changing */ // 子节点个数 int childCount; // ACL列表 List<ACL> acl; /* Make sure to create a new object when changing */ @SuppressWarnings("unchecked") // 拷贝 ChangeRecord duplicate(long zxid) { StatPersisted stat = new StatPersisted(); if (this.stat != null) { DataTree.copyStatPersisted(this.stat, stat); } return new ChangeRecord(zxid, path, stat, childCount, acl == null ? new ArrayList<ACL>() : new ArrayList(acl)); } }

说明:ChangeRecord数据结构是用于方便PrepRequestProcessor和FinalRequestProcessor之间进行信息共享,其包含了一个拷贝方法duplicate,用于返回属性相同的ChangeRecord实例。

2.3 类的属性


public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider { // 日志器 protected static final Logger LOG; static { // 初始化日志器 LOG = LoggerFactory.getLogger(ZooKeeperServer.class); Environment.logEnv("Server environment:", LOG); } // JMX服务 protected ZooKeeperServerBean jmxServerBean; protected DataTreeBean jmxDataTreeBean; // 默认心跳频率 public static final int DEFAULT_TICK_TIME = 3000; protected int tickTime = DEFAULT_TICK_TIME; / value of -1 indicates unset, use default */ // 最小会话过期时间 protected int minSessionTimeout = -1; / value of -1 indicates unset, use default */ // 最大会话过期时间 protected int maxSessionTimeout = -1; // 会话跟踪器 protected SessionTracker sessionTracker; // 事务日志快照 private FileTxnSnapLog txnLogFactory = null; // Zookeeper内存数据库 private ZKDatabase zkDb; // protected long hzxid = 0; // 异常 public final static Exception ok = new Exception("No prob"); // 请求处理器 protected RequestProcessor firstProcessor; // 运行标志 protected volatile boolean running; / * This is the secret that we use to generate passwords, for the moment it * is more of a sanity check. */ // 生成密码的密钥 static final private long superSecret = 0XB3415C00L; // int requestsInProcess; // 未处理的ChangeRecord final List<ChangeRecord> outstandingChanges = new ArrayList<ChangeRecord>(); // this data structure must be accessed under the outstandingChanges lock // 记录path对应的ChangeRecord final HashMap<String, ChangeRecord> outstandingChangesForPath = new HashMap<String, ChangeRecord>(); // 连接工厂 private ServerCnxnFactory serverCnxnFactory; // 服务器统计数据 private final ServerStats serverStats; }

说明:类中包含了心跳频率,会话跟踪器(处理会话)、事务日志快照、内存数据库、请求处理器、未处理的ChangeRecord、服务器统计信息等。

2.4 类的构造函数

1. ZooKeeperServer()型构造函数

     public ZooKeeperServer() {
        serverStats = new ServerStats(this);
    }

说明:其只初始化了服务器的统计信息。

2. ZooKeeperServer(FileTxnSnapLog, int, int, int, DataTreeBuilder, ZKDatabase)型构造函数


public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime, int minSessionTimeout, int maxSessionTimeout, DataTreeBuilder treeBuilder, ZKDatabase zkDb) { // 给属性赋值 serverStats = new ServerStats(this); this.txnLogFactory = txnLogFactory; this.zkDb = zkDb; this.tickTime = tickTime; this.minSessionTimeout = minSessionTimeout; this.maxSessionTimeout = maxSessionTimeout; LOG.info("Created server with tickTime " + tickTime + " minSessionTimeout " + getMinSessionTimeout() + " maxSessionTimeout " + getMaxSessionTimeout() + " datadir " + txnLogFactory.getDataDir() + " snapdir " + txnLogFactory.getSnapDir()); }

说明:该构造函数会初始化服务器统计数据、事务日志工厂、心跳时间、会话时间(最短超时时间和最长超时时间)。

3. ZooKeeperServer(FileTxnSnapLog, int, DataTreeBuilder)型构造函数

     public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime,
            DataTreeBuilder treeBuilder) throws IOException {
        this(txnLogFactory, tickTime, -1, -1, treeBuilder,
                new ZKDatabase(txnLogFactory));
    }

说明:其首先会生成ZooKeeper内存数据库后,然后调用第二个构造函数进行初始化操作。

4. ZooKeeperServer(File, File, int)型构造函数 

     public ZooKeeperServer(File snapDir, File logDir, int tickTime)
            throws IOException {
        this( new FileTxnSnapLog(snapDir, logDir),
                tickTime, new BasicDataTreeBuilder());
    }

说明:其会调用同名构造函数进行初始化操作。

5. ZooKeeperServer(FileTxnSnapLog, DataTreeBuilder)型构造函数


public ZooKeeperServer(FileTxnSnapLog txnLogFactory, DataTreeBuilder treeBuilder) throws IOException { this(txnLogFactory, DEFAULT_TICK_TIME, -1, -1, treeBuilder, new ZKDatabase(txnLogFactory)); }

说明:其生成内存数据库之后再调用同名构造函数进行初始化操作。

2.5 核心函数分析

1. loadData函数 


public void loadData() throws IOException, InterruptedException { /* * When a new leader starts executing Leader#lead, it * invokes this method. The database, however, has been * initialized before running leader election so that * the server could pick its zxid for its initial vote. * It does it by invoking QuorumPeer#getLastLoggedZxid. * Consequently, we don't need to initialize it once more * and avoid the penalty of loading it a second time. Not * reloading it is particularly important for applications * that host a large database. * * The following if block checks whether the database has * been initialized or not. Note that this method is * invoked by at least one other method: * ZooKeeperServer#startdata. * * See ZOOKEEPER-1642 for more detail. */ if(zkDb.isInitialized()){ // 内存数据库已被初始化 // 设置为最后处理的Zxid setZxid(zkDb.getDataTreeLastProcessedZxid()); } else { // 未被初始化,则加载数据库 setZxid(zkDb.loadDataBase()); } // Clean up dead sessions LinkedList<Long> deadSessions = new LinkedList<Long>(); for (Long session : zkDb.getSessions()) { // 遍历所有的会话 if (zkDb.getSessionWithTimeOuts().get(session) == null) { // 删除过期的会话 deadSessions.add(session); } } // 完成DataTree的初始化 zkDb.setDataTreeInit(true); for (long session : deadSessions) { // 遍历过期会话 // XXX: Is lastProcessedZxid really the best thing to use? // 删除会话 killSession(session, zkDb.getDataTreeLastProcessedZxid()); } }

说明:该函数用于加载数据,其首先会判断内存库是否已经加载设置zxid,之后会调用killSession函数删除过期的会话,killSession会从sessionTracker中删除session,并且killSession最后会调用DataTree的killSession函数,其源码如下 


void killSession(long session, long zxid) { // the list is already removed from the ephemerals // so we do not have to worry about synchronizing on // the list. This is only called from FinalRequestProcessor // so there is no need for synchronization. The list is not // changed here. Only create and delete change the list which // are again called from FinalRequestProcessor in sequence. // 移除session,并获取该session对应的所有临时节点 HashSet<String> list = ephemerals.remove(session); if (list != null) { for (String path : list) { // 遍历所有临时节点 try { // 删除路径对应的节点 deleteNode(path, zxid); if (LOG.isDebugEnabled()) { LOG .debug("Deleting ephemeral node " + path + " for session 0x" + Long.toHexString(session)); } } catch (NoNodeException e) { LOG.warn("Ignoring NoNodeException for path " + path + " while removing ephemeral for dead session 0x" + Long.toHexString(session)); } } } }

说明:DataTree的killSession函数的逻辑首先移除session,然后取得该session下的所有临时节点,然后逐一删除临时节点。

2. submit函数 


public void submitRequest(Request si) { if (firstProcessor == null) { // 第一个处理器为空 synchronized (this) { try { while (!running) { // 直到running为true,否则继续等待 wait(1000); } } catch (InterruptedException e) { LOG.warn("Unexpected interruption", e); } if (firstProcessor == null) { throw new RuntimeException("Not started"); } } } try { touch(si.cnxn); // 是否为合法的packet boolean validpacket = Request.isValid(si.type); if (validpacket) { // 处理请求 firstProcessor.processRequest(si); if (si.cnxn != null) { incInProcess(); } } else { LOG.warn("Received packet at server of unknown type " + si.type); new UnimplementedRequestProcessor().processRequest(si); } } catch (MissingSessionException e) { if (LOG.isDebugEnabled()) { LOG.debug("Dropping request: " + e.getMessage()); } } catch (RequestProcessorException e) { LOG.error("Unable to process request:" + e.getMessage(), e); } }

说明:当firstProcessor为空时,并且running标志为false时,其会一直等待,直到running标志为true,之后调用touch函数判断session是否存在或者已经超时,之后判断请求的类型是否合法,合法则使用请求处理器进行处理。

3. processConnectRequest函数


public void processConnectRequest(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException { BinaryInputArchive bia = BinaryInputArchive.getArchive(new ByteBufferInputStream(incomingBuffer)); ConnectRequest connReq = new ConnectRequest(); // 反序列化 connReq.deserialize(bia, "connect"); if (LOG.isDebugEnabled()) { LOG.debug("Session establishment request from client " + cnxn.getRemoteSocketAddress() + " client's lastZxid is 0x" + Long.toHexString(connReq.getLastZxidSeen())); } boolean readOnly = false; try { // 是否为只读 readOnly = bia.readBool("readOnly"); cnxn.isOldClient = false; } catch (IOException e) { // this is ok -- just a packet from an old client which // doesn't contain readOnly field LOG.warn("Connection request from old client " + cnxn.getRemoteSocketAddress() + "; will be dropped if server is in r-o mode"); } if (readOnly == false && this instanceof ReadOnlyZooKeeperServer) { // 为只读模式但是该服务器是只读服务器,抛出异常 String msg = "Refusing session request for not-read-only client " + cnxn.getRemoteSocketAddress(); LOG.info(msg); throw new CloseRequestException(msg); } if (connReq.getLastZxidSeen() > zkDb.dataTree.lastProcessedZxid) { // 请求连接的zxid大于DataTree处理的最大的zxid,抛出异常 String msg = "Refusing session request for client " + cnxn.getRemoteSocketAddress() + " as it has seen zxid 0x" + Long.toHexString(connReq.getLastZxidSeen()) + " our last zxid is 0x" + Long.toHexString(getZKDatabase().getDataTreeLastProcessedZxid()) + " client must try another server"; LOG.info(msg); throw new CloseRequestException(msg); } // 获取超时时间 int sessionTimeout = connReq.getTimeOut(); // 获取密码 byte passwd[] = connReq.getPasswd(); // 获取最短超时时间 int minSessionTimeout = getMinSessionTimeout(); if (sessionTimeout < minSessionTimeout) { sessionTimeout = minSessionTimeout; } // 获取最长超时时间 int maxSessionTimeout = getMaxSessionTimeout(); if (sessionTimeout > maxSessionTimeout) { sessionTimeout = maxSessionTimeout; } // 设置超时时间 cnxn.setSessionTimeout(sessionTimeout); // We don't want to receive any packets until we are sure that the // session is setup // 不接收任何packet,直到会话创建成功 cnxn.disableRecv(); // 获取会话id long sessionId = connReq.getSessionId(); if (sessionId != 0) { // 表示重新创建会话 long clientSessionId = connReq.getSessionId(); LOG.info("Client attempting to renew session 0x" + Long.toHexString(clientSessionId) + " at " + cnxn.getRemoteSocketAddress()); // 关闭会话 serverCnxnFactory.closeSession(sessionId); // 设置会话id cnxn.setSessionId(sessionId); // 重新打开会话 reopenSession(cnxn, sessionId, passwd, sessionTimeout); } else { LOG.info("Client attempting to establish new session at " + cnxn.getRemoteSocketAddress()); // 创建会话 createSession(cnxn, passwd, sessionTimeout); } }

说明:其首先将传递的ByteBuffer进行反序列化,转化为相应的ConnectRequest,之后进行一系列判断(可能抛出异常),然后获取并判断该ConnectRequest中会话id是否为0,若为0,则表示可以创建会话,否则,重新打开会话。

4. processPacket函数 


public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException { // We have the request, now process and setup for next InputStream bais = new ByteBufferInputStream(incomingBuffer); BinaryInputArchive bia = BinaryInputArchive.getArchive(bais); // 创建请求头 RequestHeader h = new RequestHeader(); // 将头反序列化为RequestHeader h.deserialize(bia, "header"); // Through the magic of byte buffers, txn will not be // pointing // to the start of the txn incomingBuffer = incomingBuffer.slice(); if (h.getType() == OpCode.auth) { // 需要进行认证(有密码) LOG.info("got auth packet " + cnxn.getRemoteSocketAddress()); AuthPacket authPacket = new AuthPacket(); // 将ByteBuffer转化为AuthPacket ByteBufferInputStream.byteBuffer2Record(incomingBuffer, authPacket); // 获取AuthPacket的模式 String scheme = authPacket.getScheme(); AuthenticationProvider ap = ProviderRegistry.getProvider(scheme); Code authReturn = KeeperException.Code.AUTHFAILED; if(ap != null) { try { // 进行认证 authReturn = ap.handleAuthentication(cnxn, authPacket.getAuth()); } catch(RuntimeException e) { LOG.warn("Caught runtime exception from AuthenticationProvider: " + scheme + " due to " + e); authReturn = KeeperException.Code.AUTHFAILED; } } if (authReturn!= KeeperException.Code.OK) { // 认证失败 if (ap == null) { LOG.warn("No authentication provider for scheme: " + scheme + " has " + ProviderRegistry.listProviders()); } else { LOG.warn("Authentication failed for scheme: " + scheme); } // send a response... // 构造响应头 ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.AUTHFAILED.intValue()); // 发送响应 cnxn.sendResponse(rh, null, null); // ... and close connection // 关闭连接的信息 cnxn.sendBuffer(ServerCnxnFactory.closeConn); // 不接收任何packet cnxn.disableRecv(); } else { // 认证成功 if (LOG.isDebugEnabled()) { LOG.debug("Authentication succeeded for scheme: " + scheme); } LOG.info("auth success " + cnxn.getRemoteSocketAddress()); // 构造响应头 ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue()); // 发送响应 cnxn.sendResponse(rh, null, null); } return; } else { if (h.getType() == OpCode.sasl) { // 为SASL类型 // 处理SASL Record rsp = processSasl(incomingBuffer,cnxn); // 构造响应头 ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue()); // 发送响应 cnxn.sendResponse(rh,rsp, "response"); // not sure about 3rd arg..what is it? } else { // 不为SASL类型 // 创建请求 Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(), h.getType(), incomingBuffer, cnxn.getAuthInfo()); // 设置请求所有者 si.setOwner(ServerCnxn.me); // 提交请求 submitRequest(si); } } // cnxn.incrOutstandingRequests(h); }

说明:该函数首先将传递的ByteBuffer进行反序列,转化为相应的RequestHeader,然后根据该RequestHeader判断是否需要认证,若认证失败,则构造认证失败的响应并发送给客户端,然后关闭连接,并且再补接收任何packet。若认证成功,则构造认证成功的响应并发送给客户端。若不需要认证,则再判断其是否为SASL类型,若是,则进行处理,然后构造响应并发送给客户端,否则,构造请求并且提交请求。

三、总结

本篇分析了ZooKeeperServer的源码,了解了其对于请求和会话的处理,也谢谢各位园友的观看~

赞(0) 打赏

如未加特殊说明,此网站文章均为原创,转载必须注明出处。Java 技术驿站 » 【Zookeeper源码分析】—服务器(二)之ZooKeeperServer
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

Java 技术驿站 | 致力打造 Java 精品博客

联系作者优质文章

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏