這篇文章主要介紹了zk工廠方法如何實現(xiàn)NIOServerCnxnFactory，具有一定借鑒價值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。

創(chuàng)新互聯(lián)是一家集網(wǎng)站建設,杞縣企業(yè)網(wǎng)站建設,杞縣品牌網(wǎng)站建設,網(wǎng)站定制,杞縣網(wǎng)站建設報價,網(wǎng)絡營銷,網(wǎng)絡優(yōu)化,杞縣網(wǎng)站推廣為一體的創(chuàng)新建站企業(yè)，幫助傳統(tǒng)企業(yè)提升企業(yè)形象加強企業(yè)競爭力?？沙浞譂M足這一群體相比中小企業(yè)更為豐富、高端、多元的互聯(lián)網(wǎng)需求。同時我們時刻保持專業(yè)、時尚、前沿，時刻以成就客戶成長自我，堅持不斷學習、思考、沉淀、凈化自己，讓我們?yōu)楦嗟钠髽I(yè)打造出實用型網(wǎng)站。

NIOServerCnxnFactory類

內(nèi)部類

AbstractSelectThread

AcceptThread

SelectorThread

屬性

ZOOKEEPER_NIO_SESSIONLESS_CNXN_TIMEOUT	10s session過期時間
ZOOKEEPER_NIO_NUM_SELECTOR_THREADS	selector 線程數(shù)
ZOOKEEPER_NIO_NUM_WORKER_THREADS	worker 線程數(shù)
directBuffer	buffer用來線程間數(shù)據(jù)交互
ipMap	限制ip上連接數(shù)
cnxnExpiryQueue	連接失效時間分桶隊列
workerPool	WorkerService worker執(zhí)行服務
acceptThread	接收新連接，simple round-robin 分配到選擇線程
selectorThreads

方法

停止接收

private void pauseAccept(long millisecs) {
    acceptKey.interestOps(0);
    try {
        selector.select(millisecs);
    } catch (IOException e) {
        // ignore
    } finally {
        acceptKey.interestOps(SelectionKey.OP_ACCEPT);
    }
}

private boolean doAccept() {
    boolean accepted = false;
    SocketChannel sc = null;
    try {
        sc = acceptSocket.accept();
        accepted = true;
        InetAddress ia = sc.socket().getInetAddress();
        int cnxncount = getClientCnxnCount(ia);

        if (maxClientCnxns > 0 && cnxncount >= maxClientCnxns) {
            throw new IOException("Too many connections from " + ia + " - max is " + maxClientCnxns);
        }

        LOG.debug("Accepted socket connection from {}", sc.socket().getRemoteSocketAddress());

        sc.configureBlocking(false);

        // Round-robin assign this connection to a selector thread
        if (!selectorIterator.hasNext()) {
            selectorIterator = selectorThreads.iterator();
        }
        SelectorThread selectorThread = selectorIterator.next();
        if (!selectorThread.addAcceptedConnection(sc)) {
            throw new IOException("Unable to add connection to selector queue"
                                  + (stopped ? " (shutdown in progress)" : ""));
        }
        acceptErrorLogger.flush();
    } catch (IOException e) {
        // accept, maxClientCnxns, configureBlocking
        ServerMetrics.getMetrics().CONNECTION_REJECTED.add(1);
        acceptErrorLogger.rateLimitLog("Error accepting new connection: " + e.getMessage());
        fastCloseSock(sc);
    }
    return accepted;
}




private void processAcceptedConnections() {
    SocketChannel accepted;
    while (!stopped && (accepted = acceptedQueue.poll()) != null) {
        SelectionKey key = null;
        try {
            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);
        }
    }
}



configure


獲取客戶端連接數(shù)
private int getClientCnxnCount(InetAddress cl) {
    Set<NIOServerCnxn> s = ipMap.get(cl);
    if (s == null) {
        return 0;
    }
    return s.size();
}


創(chuàng)建連接
protected NIOServerCnxn createConnection(SocketChannel sock, SelectionKey sk, SelectorThread selectorThread) throws IOException {
    return new NIOServerCnxn(zkServer, sock, sk, this, selectorThread);
}


創(chuàng)建連接
private void addCnxn(NIOServerCnxn cnxn) throws IOException {
    InetAddress addr = cnxn.getSocketAddress();
    if (addr == null) {
        throw new IOException("Socket of " + cnxn + " has been closed");
    }
    Set<NIOServerCnxn> set = ipMap.get(addr);
    if (set == null) {
        // in general we will see 1 connection from each
        // host, setting the initial cap to 2 allows us
        // to minimize mem usage in the common case
        // of 1 entry --  we need to set the initial cap
        // to 2 to avoid rehash when the first entry is added
        // Construct a ConcurrentHashSet using a ConcurrentHashMap
        set = Collections.newSetFromMap(new ConcurrentHashMap<NIOServerCnxn, Boolean>(2));
        // Put the new set in the map, but only if another thread
        // hasn't beaten us to it
        Set<NIOServerCnxn> existingSet = ipMap.putIfAbsent(addr, set);
        if (existingSet != null) {
            set = existingSet;
        }
    }
    set.add(cnxn);

    cnxns.add(cnxn);
    touchCnxn(cnxn);
}

思考：
為什么單機和集群模式啟動不一樣
單機可以直接從日志，快照恢復數(shù)據(jù)
集群根據(jù)角色劃分，涉及到數(shù)據(jù)同步

感謝你能夠認真閱讀完這篇文章，希望小編分享的“zk工廠方法如何實現(xiàn)NIOServerCnxnFactory”這篇文章對大家有幫助，同時也希望大家多多支持創(chuàng)新互聯(lián)，關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，更多相關(guān)知識等著你來學習!

當前題目：zk工廠方法如何實現(xiàn)NIOServerCnxnFactory
本文鏈接：http://jinyejixie.com/article18/jdosgp.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供自適應網(wǎng)站、ChatGPT、App開發(fā)、外貿(mào)網(wǎng)站建設、手機網(wǎng)站建設、網(wǎng)站導航

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)