zookeeper之session管理

ゝ一世哀愁。 2022-09-13 05:18 131阅读 0赞

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16][]

## 1.客户端建立连接 ##

ClientCnxn.SendThread

public void run() {
        startConnect(serverAddress); //开始和服务端建立连接
    }
    
    private void startConnect(InetSocketAddress addr) throws IOException {
        clientCnxnSocket.connect(addr);
    }

ClientCnxnSocketNIO

void connect(InetSocketAddress addr) throws IOException {
        SocketChannel sock = createSock();
        registerAndConnect(sock, addr);
    }
    // register with the selection and connect
    void registerAndConnect(SocketChannel sock, InetSocketAddress addr) throws IOException {
        sockKey = sock.register(selector, SelectionKey.OP_CONNECT);
        boolean immediateConnect = sock.connect(addr); //和服务端建立连接，服务端会收到请求
        if (immediateConnect) {
            sendThread.primeConnection();
        }
    }

## 2.服务端接收连接 ##

2.1 NIOServerCnxnFactory.AcceptThread

public void run() {
        while (!stopped && !acceptSocket.socket().isClosed()) {
            select();
        }
    }
    
    private void select() {
        try {
            selector.select();//查找就绪的连接
            Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
            while (!stopped && selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selectedKeys.remove();
            }
        }
    }

2.2 NIOServerCnxnFactory.SelectorThread

public void run() {
        while (!stopped) {
            select();
            processAcceptedConnections(); //处理接收到的连接
            processInterestOpsUpdateRequests();
        }
    }
    
    private void processAcceptedConnections() {
        SocketChannel accepted;
        while (!stopped && (accepted = acceptedQueue.poll()) != null) {
            SelectionKey key = null;
            key = accepted.register(selector, SelectionKey.OP_READ);
            // 针对每个连接，创建一个NIOServerCnxn
            NIOServerCnxn cnxn = createConnection(accepted, key, this);
            key.attach(cnxn);
            addCnxn(cnxn);
        }
    }
    
    private void select() {
        selector.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.isReadable() || key.isWritable()) {
                handleIO(key);
            }
        }
    }
    
    private void handleIO(SelectionKey key) {
        IOWorkRequest workRequest = new IOWorkRequest(this, key);
        NIOServerCnxn cnxn = (NIOServerCnxn) key.attachment();
        touchCnxn(cnxn);
        workerPool.schedule(workRequest);
    }

2.2.1 NIOServerCnxnFactory.touchCnxn(NIOServerCnxn)

private ExpiryQueue<NIOServerCnxn> cnxnExpiryQueue;
    // Add or update cnxn in our cnxnExpiryQueue
    public void touchCnxn(NIOServerCnxn cnxn) {
        cnxnExpiryQueue.update(cnxn, cnxn.getSessionTimeout());
    }

2.2.1.1 ExpiryQueue：服务端管理session过期的队列

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

// 维护每个NIOServerCnxn对应的过期时间
    private final ConcurrentHashMap<E, Long> elemMap = new ConcurrentHashMap<E, Long>();
    // 维护每个过期时间对应的桶里有哪些NIOServerCnxn
    private final ConcurrentHashMap<Long, Set<E>> expiryMap = new ConcurrentHashMap<Long, Set<E>>();
    private final AtomicLong nextExpirationTime = new AtomicLong();
    
    public Long update(E elem, int timeout) {
        Long prevExpiryTime = elemMap.get(elem);//获取当前NIOServerCnxn对应的过期时间
        long now = Time.currentElapsedTime();
        Long newExpiryTime = roundToNextInterval(now + timeout);//获取下次过期时间
        if (newExpiryTime.equals(prevExpiryTime)) {
            return null; // No change, so nothing to update
        }
    
        // First add the elem to the new expiry time bucket in expiryMap.
        Set<E> set = expiryMap.get(newExpiryTime); //拿到下一个过期时间的桶
        if (set == null) {
            // Construct a ConcurrentHashSet using a ConcurrentHashMap
            set = Collections.newSetFromMap(new ConcurrentHashMap<E, Boolean>());
            // Put the new set in the map, but only if another thread hasn't beaten us to it
            Set<E> existingSet = expiryMap.putIfAbsent(newExpiryTime, set);
            if (existingSet != null) {
                set = existingSet;
            }
        }
        set.add(elem); //把原来的NIOServerCnxn移动到新的桶里
    
        // Map the elem to the new expiry time. If a different previous
        // mapping was present, clean up the previous expiry bucket.
        prevExpiryTime = elemMap.put(elem, newExpiryTime);
        if (prevExpiryTime != null && !newExpiryTime.equals(prevExpiryTime)) {
            Set<E> prevSet = expiryMap.get(prevExpiryTime);
            if (prevSet != null) {
                prevSet.remove(elem); //清空之前过期的桶
            }
        }
        return newExpiryTime;
    }

2.2.2 WorkerService.schedule(workRequest)

public void schedule(WorkRequest workRequest) {
        schedule(workRequest, 0);
    }
    public void schedule(WorkRequest workRequest, long id) {
        ScheduledWorkRequest scheduledWorkRequest = new ScheduledWorkRequest(workRequest);
        int size = workers.size();
        int workerNum = ((int) (id % size) + size) % size;
        ExecutorService worker = workers.get(workerNum);
        worker.execute(scheduledWorkRequest);
    }

WorkerService.ScheduledWorkRequest

private class ScheduledWorkRequest implements Runnable {
        @Override
        public void run() {
            workRequest.doWork();
        }
    }

WorkerService.WorkRequest.doWork()

public abstract static class WorkRequest {
        public abstract void doWork() throws Exception;
    }

2.3 NIOServerCnxnFactory.IOWorkRequest.doWork()

private class IOWorkRequest extends WorkerService.WorkRequest {
        public void doWork() throws InterruptedException {
            if (key.isReadable() || key.isWritable()) {
                cnxn.doIO(key);
                touchCnxn(cnxn);
            }
        }
    }

2.4 NIOServerCnxn.doIO(SelectionKey)

// Handles read/write IO on connection.
    void doIO(SelectionKey k) throws InterruptedException {
        readPayload();
    }
    
    private void readPayload() throws IOException, InterruptedException, ClientCnxnLimitException {
        if (!initialized) {//第一次未初始化时，读取连接请求
            readConnectRequest();
        } else {
            readRequest();
        }
    }
    
    private void readConnectRequest() throws IOException, InterruptedException, ClientCnxnLimitException {
        zkServer.processConnectRequest(this, incomingBuffer);
        initialized = true;
    }

2.5 ZooKeeperServer

// 处理连接请求
    public void processConnectRequest(ServerCnxn cnxn, ByteBuffer incomingBuffer) {
        BinaryInputArchive bia = BinaryInputArchive.getArchive(new ByteBufferInputStream(incomingBuffer));
        ConnectRequest connReq = new ConnectRequest();//创建连接请求
        connReq.deserialize(bia, "connect"); //反序列化连接请求参数
        long sessionId = connReq.getSessionId(); //创建一个sessionId
        int sessionTimeout = connReq.getTimeOut();
        byte[] passwd = connReq.getPasswd();
        cnxn.setSessionTimeout(sessionTimeout);
        if (sessionId == 0) {
            long id = createSession(cnxn, passwd, sessionTimeout); //创建session
        }
    }
    
    long createSession(ServerCnxn cnxn, byte[] passwd, int timeout) {
        //sessionTracker去创建一个sessionId
        long sessionId = sessionTracker.createSession(timeout);
        cnxn.setSessionId(sessionId);
        Request si = new Request(cnxn, sessionId, 0, OpCode.createSession, to, null);
        submitRequest(si);
        return sessionId;
    }
    
    public void submitRequest(Request si) {
        enqueueRequest(si);
    }
    
    public void enqueueRequest(Request si) {
        requestThrottler.submitRequest(si);
    }

2.6 RequestThrottler请求限流阀

// 阻塞队列保存提交的请求
    private final LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();
    public void submitRequest(Request request) {
        submittedRequests.add(request);
    }
    
    public void run() {
        while (true) {
            Request request = submittedRequests.take();
            zks.submitRequestNow(request);
        }
    }

2.7 ZooKeeperServer

public void submitRequestNow(Request si) {
        touch(si.cnxn); //更新会话连接超时时间
        boolean validpacket = Request.isValid(si.type);
        if (validpacket) {
            firstProcessor.processRequest(si);
        }
    }

2.7.1 ZooKeeperServer.touch(ServerCnxn)

void touch(ServerCnxn cnxn) throws MissingSessionException {
        long id = cnxn.getSessionId();
        int to = cnxn.getSessionTimeout();
        if (!sessionTracker.touchSession(id, to)) {
            throw new MissingSessionException("No session with sessionid 0x" + Long.toHexString(id)  + " exists, probably expired and removed");
        }
    }

SessionTrackerImpl

public synchronized boolean touchSession(long sessionId, int timeout) {
        SessionImpl s = sessionsById.get(sessionId);
        updateSessionExpiry(s, timeout);
        return true;
    }
    
    private void updateSessionExpiry(SessionImpl s, int timeout) {
        sessionExpiryQueue.update(s, timeout); //更新session过期时间到ExpiryQueue
    }

2.7.2 firstProcessor.processRequest(si);

PrepRequestProcessor.processRequest(request)

public void processRequest(Request request) {
        request.prepQueueStartTime = Time.currentElapsedTime();
        submittedRequests.add(request);
    }
    
    public void run() {
        while (true) {
            Request request = submittedRequests.take();
            pRequest(request);
        }
    }
    
    protected void pRequest(Request request) throws RequestProcessorException {
        request.setHdr(null);
        request.setTxn(null);
        switch (request.type) {
        case OpCode.createSession: //针对连接请求做处理
        case OpCode.closeSession:
            if (!request.isLocalSession()) {
                pRequest2Txn(request.type, zks.getNextZxid(), request, null, true);
            }
            break;
        }
    }
    
    protected void pRequest2Txn(int type, long zxid, Request request, Record record, boolean deserialize) {
        switch (type) {
        case OpCode.createSession:
            int to = request.request.getInt();
            request.setTxn(new CreateSessionTxn(to));
            zks.sessionTracker.trackSession(request.sessionId, to);
            zks.setOwner(request.sessionId, request.getOwner());
            break;
        }
    }

FinalRequestProcessor.processRequest(request)

public void processRequest(Request request) {
        ProcessTxnResult rc = zks.processTxn(request);
        switch (request.type) {
            case OpCode.createSession: {
                lastOp = "SESS";
                updateStats(request, lastOp, lastZxid);
                zks.finishSessionInit(request.cnxn, true);
                return;
            }
        }
        if (path == null || rsp == null) {
            cnxn.sendResponse(hdr, rsp, "response"); //服务端将请求返回，这时客户端会收到服务端响应
        }
    }

ZooKeeperServer.finishSessionInit(ServerCnxn, valid)

public void finishSessionInit(ServerCnxn cnxn, boolean valid) {
        // register with JMX
        if (valid) {
            if (serverCnxnFactory != null && serverCnxnFactory.cnxns.contains(cnxn)) {
                serverCnxnFactory.registerConnection(cnxn);
            } else if (secureServerCnxnFactory != null && secureServerCnxnFactory.cnxns.contains(cnxn)) {
                secureServerCnxnFactory.registerConnection(cnxn);
            }
        }
    }

## 3.客户端收到服务端响应 ##

ClientCnxnSocketNIO.doIO(Queue<Packet>, ClientCnxn)

void doIO(Queue<Packet> pendingQueue, ClientCnxn cnxn) throws InterruptedException, IOException {
        if (sockKey.isReadable()) {
            if (!initialized) {
                readConnectResult();
            }
        }
    }
    
    void readConnectResult() throws IOException {
        ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
        BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
        ConnectResponse conRsp = new ConnectResponse();
        conRsp.deserialize(bbia, "connect");
        this.sessionId = conRsp.getSessionId(); //连接建立完成
        sendThread.onConnected(conRsp.getTimeOut(), this.sessionId, conRsp.getPasswd(), isRO);
    }

## 4. 心跳包发送 ##

ClientCnxn.SendThread

public void run() {
        clientCnxnSocket.introduce(this, sessionId, outgoingQueue);
        clientCnxnSocket.updateNow();
        clientCnxnSocket.updateLastSendAndHeard();
        while (state.isAlive()) {
            //如果连接建立，每隔段时间发送PING请求
            if (state.isConnected()) {
                //1000(1 second) is to prevent race condition missing to send the second ping
                //also make sure not to send too many pings when readTimeout is small
                int timeToNextPing = readTimeout / 2
                                     - clientCnxnSocket.getIdleSend()
                                     - ((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);
                //send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL
                if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {
                    sendPing();
                    clientCnxnSocket.updateLastSend();
                } else {
                    if (timeToNextPing < to) {
                        to = timeToNextPing;
                    }
                }
            }
        }
    }
    // 发送PING请求给服务端
    private void sendPing() {
        lastPingSentNs = System.nanoTime();
        RequestHeader h = new RequestHeader(ClientCnxn.PING_XID, OpCode.ping);
        queuePacket(h, null, null, null, null, null, null, null, null);
    }
    
    public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request, Record response, AsyncCallback cb, String clientPath, String serverPath, Object ctx, WatchRegistration watchRegistration, WatchDeregistration watchDeregistration) {
        Packet packet = null;
    
        // Note that we do not generate the Xid for the packet yet. It is
        // generated later at send-time, by an implementation of ClientCnxnSocket::doIO(),
        // where the packet is actually sent.
        packet = new Packet(h, r, request, response, watchRegistration);
        packet.cb = cb;
        packet.ctx = ctx;
        packet.clientPath = clientPath;
        packet.serverPath = serverPath;
        packet.watchDeregistration = watchDeregistration;
        // The synchronized block here is for two purpose:
        // 1. synchronize with the final cleanup() in SendThread.run() to avoid race
        // 2. synchronized against each packet. So if a closeSession packet is added,
        // later packet will be notified.
        synchronized (state) {
            if (!state.isAlive() || closing) {
                conLossPacket(packet);
            } else {
                // If the client is asking to close the session then mark as closing
                if (h.getType() == OpCode.closeSession) {
                    closing = true;
                }
                outgoingQueue.add(packet);
            }
        }
        sendThread.getClientCnxnSocket().packetAdded();
        return packet;
    }
    
    public void run() {
        clientCnxnSocket.introduce(this, sessionId, outgoingQueue);
        clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);
    }

服务端会收到客户端的PING请求，重复上面2.4的过程

NIOServerCnxn.doIO(SelectionKey)

void doIO(SelectionKey k) throws InterruptedException {
        if (isPayload) { // not the case for 4letterword
            readPayload()
        }
    }
    
    private void readPayload() throws IOException, InterruptedException, ClientCnxnLimitException {
        if (incomingBuffer.remaining() == 0) {
            if (!initialized) {
                readConnectRequest();
            } else {
                readRequest();
            }
        }
    }
    
    private void readConnectRequest() throws IOException, InterruptedException, ClientCnxnLimitException {
        zkServer.processConnectRequest(this, incomingBuffer);
        initialized = true;
    }

除了PING请求以外，其他正常的CRUD请求也会对session续约

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/536f07aa71d94056a78640ddab0e1f7f.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWmhhb1RpbmdMb25n_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/677cc2c2490448f090214743edc095b2.png