zookeeper源码分析之watcher机制

灰太狼 2022-09-08 10:25 147阅读 0赞

# 源码设计猜想  #

1.  客户端和服务端建立连接，使用NIO通信
    
        ZooKeeper zooKeeper = new ZooKeeper("127.0.0.1:2181", 30000, new Watcher() {
            @Override
            public void process(WatchedEvent event) { //全局监听回调
                System.out.println("nodePath=" + event.getPath() + ", eventType=" + event.getType());
            }
        });
2.  发送相关操作到服务端
    
    1.  非事务性操作
        
            zooKeeper.exists("/watcher", true);
    2.  事务性操作
        
            zooKeeper.setData("/watcher", "hello".getBytes(), -1);
3.  传递数据  
    数据协议：jute协议  
    序列化、反序列化
4.  服务端对接收的数据进行处理
    
    1.  单机模式：CRUD操作
    2.  集群模式：CRUD操作，还涉及到数据同步
5.  基于exists，如果带有watch机制，节点数据发生变化，需要触发对应客户端的watch回调
    
    1.  在服务端维护节点path和connection的关系
    2.  在客户端维护节点path和watch回调的关系

# 客户端建立连接 #

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16][]

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
        boolean canBeReadOnly) throws IOException {
        // createDefaultHostProvider(connectString) 客户端的负载均衡
        this(connectString, sessionTimeout, watcher, canBeReadOnly, createDefaultHostProvider(connectString));
    }
    
    // default hostprovider
    private static HostProvider createDefaultHostProvider(String connectString) {
        return new StaticHostProvider(new ConnectStringParser(connectString).getServerAddresses());
    }
    
    public StaticHostProvider(Collection<InetSocketAddress> serverAddresses) {
        init(serverAddresses, System.currentTimeMillis() ^ this.hashCode(), new Resolver() {
            @Override
            public InetAddress[] getAllByName(String name) throws UnknownHostException {
                return InetAddress.getAllByName(name);
            }
        });
    }

初始化网络通信协议ClientCnxnSocketNIO，构建客户端监听manager，创建数据发送线程和watch线程，从而创建客户端连接，启动两个线程

// Initiating client connection
    public ZooKeeper(String connectString,int sessionTimeout,Watcher watcher,boolean canBeReadOnly,
        HostProvider aHostProvider,ZKClientConfig clientConfig) throws IOException {
    
        if (clientConfig == null) {
            clientConfig = new ZKClientConfig(); //Initialize zk client properties from java system property
        }
        this.clientConfig = clientConfig;
        // 初始化客户端watchManager去管理path与watcher的关系
        watchManager = defaultWatchManager();
        // 全局的watch，watcher回调的watch
        watchManager.defaultWatcher = watcher;
        ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
        hostProvider = aHostProvider; //客户端的负载处理器
    
        //getClientCnxnSocket通过反射构建客户端网络通信实例ClientCnxnSocketNIO
        cnxn = createConnection(connectStringParser.getChrootPath(),hostProvider,sessionTimeout,
            this, watchManager, getClientCnxnSocket(), canBeReadOnly);
        cnxn.start();//启动数据发送线程和wathch线程
    }
    
    static class ZKWatchManager implements ClientWatchManager {
        private final Map<String, Set<Watcher>> dataWatches = new HashMap<String, Set<Watcher>>();
        private final Map<String, Set<Watcher>> existWatches = new HashMap<String, Set<Watcher>>();
        private final Map<String, Set<Watcher>> childWatches = new HashMap<String, Set<Watcher>>();
        private final Map<String, Set<Watcher>> persistentWatches = new HashMap<String, Set<Watcher>>();
        private final Map<String, Set<Watcher>> persistentRecursiveWatches = new HashMap<String, Set<Watcher>>();
        private boolean disableAutoWatchReset;
    }
    // 构建客户端网络通信实例
    private ClientCnxnSocket getClientCnxnSocket() throws IOException {
        String clientCnxnSocketName = getClientConfig().getProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET);
        if (clientCnxnSocketName == null || clientCnxnSocketName.equals(ClientCnxnSocketNIO.class.getSimpleName())) {
            clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();
        } else if (clientCnxnSocketName.equals(ClientCnxnSocketNetty.class.getSimpleName())) {
            clientCnxnSocketName = ClientCnxnSocketNetty.class.getName();
        }
        try {
            Constructor<?> clientCxnConstructor = Class.forName(clientCnxnSocketName).getDeclaredConstructor(ZKClientConfig.class);
            ClientCnxnSocket clientCxnSocket = (ClientCnxnSocket) clientCxnConstructor.newInstance(getClientConfig());
            return clientCxnSocket;
        } catch (Exception e) {
            throw new IOException("Couldn't instantiate " + clientCnxnSocketName, e);
        }
    }
    // 客户端连接
    public class ClientCnxn {
        public ClientCnxn(
            sendThread = new SendThread(clientCnxnSocket);//发送数据线程
            eventThread = new EventThread();//事件响应线程-watch线程
        }
        public void start() {
            sendThread.start();
            eventThread.start();
        }
    }

zooKeeper.exists("/watcher", true);

public Stat exists(String path, boolean watch) throws KeeperException, InterruptedException {
        return exists(path, watch ? watchManager.defaultWatcher : null); //如果watch为true则使用默认监听
    }
    
    public Stat exists(final String path, Watcher watcher) throws KeeperException, InterruptedException {
        RequestHeader h = new RequestHeader();
        h.setType(ZooDefs.OpCode.exists); //操作类型:exists/create/delete/setData/getData...
        ExistsRequest request = new ExistsRequest();
        request.setPath(serverPath);
        request.setWatch(watcher != null);
        SetDataResponse response = new SetDataResponse();
        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);
        }
    
        return response.getStat().getCzxid() == -1 ? null : response.getStat();
    }
    // 提交事件到queue
    public ReplyHeader submitRequest(RequestHeader h, Record request, Record response, WatchRegistration watchRegistration, 
        WatchDeregistration watchDeregistration) throws InterruptedException {
        ReplyHeader r = new ReplyHeader();
        //把请求封装为packet，然后放入outgoingQueue中
        Packet packet = queuePacket(h,r,request,response,null,null,null,null,watchRegistration,watchDeregistration);
        synchronized (packet) {//加锁等待packet完成
            if (requestTimeout > 0) {
                waitForPacketFinish(r, packet);// Wait for request completion with timeout
            } else {            
                while (!packet.finished) {// Wait for request completion infinitely
                    packet.wait();
                }
            }
        }
        if (r.getErr() == Code.REQUESTTIMEOUT.intValue()) {
            sendThread.cleanAndNotifyState();
        }
        return r;
    }

客户端端网络通信运行

public void run() {
        clientCnxnSocket.introduce(this, sessionId, outgoingQueue);
        clientCnxnSocket.updateNow();
        clientCnxnSocket.updateLastSendAndHeard();
        while (state.isAlive()) {
            serverAddress = hostProvider.next(1000);
            startConnect(serverAddress);
            clientCnxnSocket.updateNow();
            clientCnxnSocket.updateLastSendAndHeard();
            clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);
        }
    }

发送发送doTransport

void doTransport(int waitTimeOut, Queue<Packet> pendingQueue, ClientCnxn cnxn) throws IOException, InterruptedException {
        selector.select(waitTimeOut);
        Set<SelectionKey> selected;
        synchronized (this) {
            selected = selector.selectedKeys();
        }
        updateNow();
        for (SelectionKey k : selected) {
            SocketChannel sc = ((SocketChannel) k.channel());
            if ((k.readyOps() & SelectionKey.OP_CONNECT) != 0) {
                if (sc.finishConnect()) {
                    updateLastSendAndHeard();
                    updateSocketAddresses();
                    sendThread.primeConnection();
                }
            } else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
                doIO(pendingQueue, cnxn);
            }
        }
        if (sendThread.getZkState().isConnected()) {
            if (findSendablePacket(outgoingQueue, sendThread.tunnelAuthInProgress()) != null) {
                enableWrite();
            }
        }
        selected.clear();
    }

io操作

void doIO(Queue<Packet> pendingQueue, ClientCnxn cnxn) throws InterruptedException, IOException {
        SocketChannel sock = (SocketChannel) sockKey.channel();
        if (sockKey.isReadable()) {
            int rc = sock.read(incomingBuffer);
        }
        if (sockKey.isWritable()) {
            Packet p = findSendablePacket(outgoingQueue, sendThread.tunnelAuthInProgress());
            sock.write(p.bb);
        }
    }

# watch机制的绑定和事件回调 #

zookeeper server启动

public void run() {
        try {
            while (!stopped) {
                select();
                processAcceptedConnections();
                processInterestOpsUpdateRequests();
            }
            // Close connections still pending on the selector. Any others
            // with in-flight work, let drain out of the work queue.
            for (SelectionKey key : selector.keys()) {
                NIOServerCnxn cnxn = (NIOServerCnxn) key.attachment();
                if (cnxn.isSelectable()) {
                    cnxn.close(ServerCnxn.DisconnectReason.SERVER_SHUTDOWN);
                }
                cleanupSelectionKey(key);
            }
            SocketChannel accepted;
            while ((accepted = acceptedQueue.poll()) != null) {
                fastCloseSock(accepted);
            }
            updateQueue.clear();
        } finally {
            closeSelector();
            // This will wake up the accept thread and the other selector
            // threads, and tell the worker thread pool to begin shutdown.
            NIOServerCnxnFactory.this.stop();
            LOG.info("selector thread exitted run method");
        }
    }

## NIOServerCnxnFactory.select()获取客户端连接 ##

private void select() {
        try {
            selector.select(); //服务端查找就绪的连接，客户端建立连接，select被唤醒
            // 遍历所有被唤醒的连接
            Set<SelectionKey> selected = selector.selectedKeys();
            ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>(selected);
            Collections.shuffle(selectedList);
            Iterator<SelectionKey> selectedKeys = selectedList.iterator();
            while (!stopped && selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selected.remove(key);
    
                if (!key.isValid()) {
                    cleanupSelectionKey(key);
                    continue;
                }
                if (key.isReadable() || key.isWritable()) {
                    handleIO(key);
                } else {
                    LOG.warn("Unexpected ops in select {}", key.readyOps());
                }
            }
        } catch (IOException e) {
            LOG.warn("Ignoring IOException while selecting", e);
        }
    }

NIOServerCnxnFactory.doAccept()

private boolean doAccept() {
        boolean accepted = false;
        SocketChannel sc = null;
        try {
            sc = acceptSocket.accept(); //获取客户端连接
            accepted = true;
            InetAddress ia = sc.socket().getInetAddress();
            int cnxncount = getClientCnxnCount(ia);
            sc.configureBlocking(false);//设置IO非阻塞
            // Round-robin assign this connection to a selector thread
            if (!selectorIterator.hasNext()) {
                selectorIterator = selectorThreads.iterator();
            }
            // 获取一个sub-selector
            SelectorThread selectorThread = selectorIterator.next();
            acceptErrorLogger.flush();
        } catch (IOException e) {
            ServerMetrics.getMetrics().CONNECTION_REJECTED.add(1);
            acceptErrorLogger.rateLimitLog("Error accepting new connection: " + e.getMessage());
            fastCloseSock(sc);
        }
        return accepted;
    }
    
    public boolean addAcceptedConnection(SocketChannel accepted) {
        if (stopped || !acceptedQueue.offer(accepted)) {//将客户端连接添加到acceptedQueue队列
            return false;
        }
        wakeupSelector(); //唤醒selector，就会唤醒select方法
        return true;
    }
    
    private void select() {
        try {
            selector.select(); //轮询IO事件，通过wakeupSelector唤醒
            Set<SelectionKey> selected = selector.selectedKeys();
            ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>(selected);
            Collections.shuffle(selectedList);
            Iterator<SelectionKey> selectedKeys = selectedList.iterator();
            while (!stopped && selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selected.remove(key);
                if (key.isReadable() || key.isWritable()) {
                    handleIO(key); //读或写事件，进行IO处理
                }
            }
        } catch (IOException e) {
            LOG.warn("Ignoring IOException while selecting", e);
        }
    }

NIOServerCnxnFactory.handleIO()处理IO事件 - 异步处理

private void handleIO(SelectionKey key) {
        IOWorkRequest workRequest = new IOWorkRequest(this, key);
        NIOServerCnxn cnxn = (NIOServerCnxn) key.attachment();
        cnxn.disableSelectable();
        key.interestOps(0);
        touchCnxn(cnxn);
        workerPool.schedule(workRequest);
    }

WorkerService.schedule()

// 从线程池中取一个线程来处理IO事件
    public void schedule(WorkRequest workRequest, long id) {
        ScheduledWorkRequest scheduledWorkRequest = new ScheduledWorkRequest(workRequest);
        // If we have a worker thread pool, use that; otherwise, do the work directly.
        int size = workers.size();
        if (size > 0) {
            try {
                // make sure to map negative ids as well to [0, size-1]
                int workerNum = ((int) (id % size) + size) % size;
                ExecutorService worker = workers.get(workerNum);
                worker.execute(scheduledWorkRequest);
            }
        } else {
            // When there is no worker thread pool, do the work directly and wait for its completion
            scheduledWorkRequest.run();
        }
    }
    
    public void run() {
        try {
            workRequest.doWork();
        } catch (Exception e) {
            workRequest.cleanup();
        }
    }

NIOServerCnxnFactory.doWork()

public void doWork() throws InterruptedException {
        if (key.isReadable() || key.isWritable()) {
            cnxn.doIO(key);
            touchCnxn(cnxn);
        }
        cnxn.enableSelectable();
    }

NIOServerCnxn.doIO() - 处理客户端连接的IO数据

void doIO(SelectionKey k) throws InterruptedException {
        if (k.isReadable()) { //从客户端读取数据
            int rc = sock.read(incomingBuffer); //从socketChannel中读取客户端数据
            if (incomingBuffer.remaining() == 0) {
                boolean isPayload;
                if (incomingBuffer == lenBuffer) { // start of next request
                    incomingBuffer.flip(); //反转
                    isPayload = readLength(k);
                    incomingBuffer.clear();
                } else {
                    isPayload = true; // continuation
                }
                if (isPayload) { // not the case for 4letterword
                    readPayload();
                } else {
                    // four letter words take care need not do anything else
                    return;
                }
            }
        }
        if (k.isWritable()) {
            handleWrite(k);
        }
    }

NIOServerCnxn.readPayload()

// Read the request payload
    private void readPayload() throws IOException, InterruptedException, ClientCnxnLimitException {
        if (incomingBuffer.remaining() != 0) { // have we read length bytes?
            int rc = sock.read(incomingBuffer); // socket is non-blocking, so ok
            if (rc < 0) {
                handleFailedRead();
            }
        }
        if (incomingBuffer.remaining() == 0) { // have we read length bytes?
            incomingBuffer.flip();
            packetReceived(4 + incomingBuffer.remaining());
            if (!initialized) {
                readConnectRequest();
            } else {
                readRequest();
            }
            lenBuffer.clear();
            incomingBuffer = lenBuffer;
        }
    }
    
    private void readRequest() throws IOException {
        zkServer.processPacket(this, incomingBuffer); //进入数据包业务处理
    }

ZooKeeperServer.processPacket() - 处理数据包数据

public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
        // exists类型的请求，会进入到这段逻辑
        // cnxn: 写数据到客户端时，需要ServerConnection
        Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(), h.getType(), incomingBuffer, cnxn.getAuthInfo());
        int length = incomingBuffer.limit();
        si.setOwner(ServerCnxn.me);
        submitRequest(si);
    }
    
    public void submitRequest(Request si) {
        enqueueRequest(si);
    }
    
    public void enqueueRequest(Request si) {
        requestThrottler.submitRequest(si);
    }

RequestThrottler.submitRequest()

public void submitRequest(Request request) {
        private final LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();
        submittedRequests.add(request);//生产者消费者模式
    }
    
    public void run() {
        while (true) {
            Request request = submittedRequests.take();
            zks.submitRequestNow(request);
        }
    }

ZooKeeperServer.submitRequestNow()

public void submitRequestNow(Request si) {
        //All requests are passed to the request processor it should wait for setting up the request processor chain.
        firstProcessor.processRequest(si); //请求责任链
    }
    //建立requestProcessor责任链: PrepRequestProcessor -> SyncRequestProcessor -> FinalRequestProcessor
    protected void setupRequestProcessors() {
        RequestProcessor finalProcessor = new FinalRequestProcessor(this);
        RequestProcessor syncProcessor = new SyncRequestProcessor(this, finalProcessor);
        ((SyncRequestProcessor) syncProcessor).start();
        firstProcessor = new PrepRequestProcessor(this, syncProcessor);
        ((PrepRequestProcessor) firstProcessor).start();
    }

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

PrepRequestProcessor.processRequest()

public void processRequest(Request request) {
        request.prepQueueStartTime = Time.currentElapsedTime();
        submittedRequests.add(request);
        ServerMetrics.getMetrics().PREP_PROCESSOR_QUEUED.add(1);
    }
    
    public void run() {
        try {
            while (true) {
                Request request = submittedRequests.take();
                long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
                if (request.type == OpCode.ping) {
                    traceMask = ZooTrace.CLIENT_PING_TRACE_MASK;
                }
                if (Request.requestOfDeath == request) {
                    break;
                }
                request.prepStartTime = Time.currentElapsedTime();
                pRequest(request);
            }
        } catch (Exception e) {
            handleException(this.getName(), e);
        }
    }
    // Create a Txn and check session
    protected void pRequest(Request request) throws RequestProcessorException {
        switch (request.type) {
        case OpCode.createContainer:
        case OpCode.create:
        case OpCode.create2:
            CreateRequest create2Request = new CreateRequest();
            pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);
            break;
        case OpCode.createTTL:
            CreateTTLRequest createTtlRequest = new CreateTTLRequest();
            pRequest2Txn(request.type, zks.getNextZxid(), request, createTtlRequest, true);
            break;
        case OpCode.deleteContainer:
        case OpCode.delete:
            DeleteRequest deleteRequest = new DeleteRequest();
            pRequest2Txn(request.type, zks.getNextZxid(), request, deleteRequest, true);
            break;
        case OpCode.setData:
            SetDataRequest setDataRequest = new SetDataRequest();
            pRequest2Txn(request.type, zks.getNextZxid(), request, setDataRequest, true);
            break;
        case OpCode.reconfig:
            ReconfigRequest reconfigRequest = new ReconfigRequest();
            ByteBufferInputStream.byteBuffer2Record(request.request, reconfigRequest);
            pRequest2Txn(request.type, zks.getNextZxid(), request, reconfigRequest, true);
            break;
        case OpCode.setACL:
            SetACLRequest setAclRequest = new SetACLRequest();
            pRequest2Txn(request.type, zks.getNextZxid(), request, setAclRequest, true);
            break;
        case OpCode.check:
            CheckVersionRequest checkRequest = new CheckVersionRequest();
            pRequest2Txn(request.type, zks.getNextZxid(), request, checkRequest, true);
            break;
        case OpCode.multi:
            MultiOperationRecord multiRequest = new MultiOperationRecord();
        //All the rest don't need to create a Txn - just verify session
        case OpCode.sync:
        case OpCode.exists:
        case OpCode.getData:
        case OpCode.getACL:
        case OpCode.getChildren:
        case OpCode.getAllChildrenNumber:
        case OpCode.getChildren2:
        case OpCode.ping:
        case OpCode.setWatches:
        case OpCode.setWatches2:
        case OpCode.checkWatches:
        case OpCode.removeWatches:
        case OpCode.getEphemerals:
        case OpCode.multiRead:
        case OpCode.addWatch:
            zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
            break;
        default:
            LOG.warn("unknown type {}", request.type);
            break;
        }
        request.zxid = zks.getZxid();
        nextProcessor.processRequest(request);
    }

SyncRequestProcessor.processRequest()

public void processRequest(final Request request) {
        request.syncQueueStartTime = Time.currentElapsedTime();
        queuedRequests.add(request);
        ServerMetrics.getMetrics().SYNC_PROCESSOR_QUEUED.add(1);
    }
    
    public void run() {
        resetSnapshotStats();
        while (true) {
            long pollTime = Math.min(zks.getMaxWriteQueuePollTime(), getRemainingDelay());
            Request si = queuedRequests.poll(pollTime, TimeUnit.MILLISECONDS);
            if (si == null) {
                flush();
                si = queuedRequests.take();
            }
            if (si == REQUEST_OF_DEATH) {
                break;
            }
    
            long startProcessTime = Time.currentElapsedTime();
            // track the number of records written to the log
            if (zks.getZKDatabase().append(si)) {
                if (shouldSnapshot()) {
                    resetSnapshotStats();
                    // roll the log
                    zks.getZKDatabase().rollLog();
                    // take a snapshot
                    if (!snapThreadMutex.tryAcquire()) {
                        LOG.warn("Too busy to snap, skipping");
                    } else {
                        new ZooKeeperThread("Snapshot Thread") {
                            public void run() {
                                try {
                                    zks.takeSnapshot();
                                } catch (Exception e) {
                                    LOG.warn("Unexpected exception", e);
                                } finally {
                                    snapThreadMutex.release();
                                }
                            }
                        }.start();
                    }
                }
            } else if (toFlush.isEmpty()) {
                // optimization for read heavy workloads
                // iff this is a read, and there are no pending
                // flushes (writes), then just pass this to the next processor
                if (nextProcessor != null) {
                    nextProcessor.processRequest(si);
                    if (nextProcessor instanceof Flushable) {
                        ((Flushable) nextProcessor).flush();
                    }
                }
                continue;
            }
            toFlush.add(si);
            if (shouldFlush()) {
                flush();
            }
        }
    }

FinalRequestProcessor.processRequest()

public void processRequest(Request request) {
        long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
        if (request.type == OpCode.ping) {
            traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
        }
        if (LOG.isTraceEnabled()) {
            ZooTrace.logRequest(LOG, traceMask, 'E', request, "");
        }
        ProcessTxnResult rc = zks.processTxn(request);
        ServerCnxn cnxn = request.cnxn;
        long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
        switch (request.type) {
        case OpCode.ping: {
            lastOp = "PING";
            updateStats(request, lastOp, lastZxid);
            cnxn.sendResponse(new ReplyHeader(ClientCnxn.PING_XID, lastZxid, 0), null, "response");
            return;
        }
        case OpCode.create: {
            lastOp = "CREA";
            rsp = new CreateResponse(rc.path);
            err = Code.get(rc.err);
            requestPathMetricsCollector.registerRequest(request.type, rc.path);
            break;
        }
        case OpCode.delete:
        case OpCode.deleteContainer: {
            lastOp = "DELE";
            err = Code.get(rc.err);
            requestPathMetricsCollector.registerRequest(request.type, rc.path);
            break;
        }
        case OpCode.setData: {
            lastOp = "SETD";
            rsp = new SetDataResponse(rc.stat);
            err = Code.get(rc.err);
            requestPathMetricsCollector.registerRequest(request.type, rc.path);
            break;
        }
        case OpCode.sync: {
            lastOp = "SYNC";
            SyncRequest syncRequest = new SyncRequest();
            ByteBufferInputStream.byteBuffer2Record(request.request, syncRequest);
            rsp = new SyncResponse(syncRequest.getPath());
            requestPathMetricsCollector.registerRequest(request.type, syncRequest.getPath());
            break;
        }
        case OpCode.check: {
            lastOp = "CHEC";
            rsp = new SetDataResponse(rc.stat);
            err = Code.get(rc.err);
            break;
        }
        case OpCode.exists: {
            lastOp = "EXIS";
            ExistsRequest existsRequest = new ExistsRequest();
            ByteBufferInputStream.byteBuffer2Record(request.request, existsRequest);
            path = existsRequest.getPath();
            if (path.indexOf('\0') != -1) {
                throw new KeeperException.BadArgumentsException();
            }
            // 得到stat信息，cnxn：NIOServerCnxn(每一个客户端连接，都会生成一个NIOServerCnxn对象实例)
            Stat stat = zks.getZKDatabase().statNode(path, existsRequest.getWatch() ? cnxn : null);
            rsp = new ExistsResponse(stat);
            requestPathMetricsCollector.registerRequest(request.type, path);
            break;
        }
        case OpCode.getData: {
            lastOp = "GETD";
            GetDataRequest getDataRequest = new GetDataRequest();
            ByteBufferInputStream.byteBuffer2Record(request.request, getDataRequest);
            path = getDataRequest.getPath();
            rsp = handleGetDataRequest(getDataRequest, cnxn, request.authInfo);
            requestPathMetricsCollector.registerRequest(request.type, path);
            break;
        }
    
        ReplyHeader hdr = new ReplyHeader(request.cxid, lastZxid, err.intValue());
        updateStats(request, lastOp, lastZxid);
    
        if (path == null || rsp == null) {
            cnxn.sendResponse(hdr, rsp, "response");
        } else {
            int opCode = request.type;
            Stat stat = null;
            // Serialized read and get children responses could be cached by the connection
            // object. Cache entries are identified by their path and last modified zxid,
            // so these values are passed along with the response.
            switch (opCode) {
                case OpCode.getData : {
                    GetDataResponse getDataResponse = (GetDataResponse) rsp;
                    stat = getDataResponse.getStat();
                    cnxn.sendResponse(hdr, rsp, "response", path, stat, opCode);
                    break;
                }
                case OpCode.getChildren2 : {
                    GetChildren2Response getChildren2Response = (GetChildren2Response) rsp;
                    stat = getChildren2Response.getStat();
                    cnxn.sendResponse(hdr, rsp, "response", path, stat, opCode);
                    break;
                }
                default:
                    cnxn.sendResponse(hdr, rsp, "response");
            }
        }
        if (request.type == OpCode.closeSession) {
            cnxn.sendCloseSession();
        }
    }

## processAcceptedConnections()处理接收到的连接 ##

private void processAcceptedConnections() {
        SocketChannel accepted;
        while (!stopped && (accepted = acceptedQueue.poll()) != null) {
            SelectionKey key = null;
            try {//将接受到连接注册到selector
                key = accepted.register(selector, SelectionKey.OP_READ);
                NIOServerCnxn cnxn = createConnection(accepted, key, this);
                key.attach(cnxn);
                addCnxn(cnxn);
            } catch (IOException e) {// register, createConnection
                cleanupSelectionKey(key);
                fastCloseSock(accepted);
            }
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/4dc562031853408daa2ce4068a17b0d0.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/4ac6f47f34634a5884acfa4a6704096a.png