Apache?Tomcat如何高并發(fā)處理請(qǐng)求

更新時(shí)間：2022年03月29日 16:45:48 作者：chenzw93

本文主要介紹了Apache?Tomcat如何高并發(fā)處理請(qǐng)求，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

介紹

作為常用的http協(xié)議服務(wù)器，tomcat應(yīng)用非常廣泛。tomcat也是遵循Servelt協(xié)議的，Servelt協(xié)議可以讓服務(wù)器與真實(shí)服務(wù)邏輯代碼進(jìn)行解耦。各自只需要關(guān)注Servlet協(xié)議即可。
對(duì)于tomcat是如何作為一個(gè)高性能的服務(wù)器的呢？你是不是也會(huì)有這樣的疑問(wèn)？

tomcat是如何接收網(wǎng)絡(luò)請(qǐng)求？

如何做到高性能的http協(xié)議服務(wù)器？

tomcat從8.0往后開(kāi)始使用了NIO非阻塞io模型，提高了吞吐量，本文的源碼是tomcat 9.0.48版本

接收Socket請(qǐng)求

org.apache.tomcat.util.net.Acceptor實(shí)現(xiàn)了Runnable接口，在一個(gè)單獨(dú)的線程中以死循環(huán)的方式一直進(jìn)行socket的監(jiān)聽(tīng)

線程的初始化及啟動(dòng)是在方法org.apache.tomcat.util.net.AbstractEndpoint#startAcceptorThread

有個(gè)很重要的屬性org.apache.tomcat.util.net.AbstractEndpoint；同時(shí)實(shí)現(xiàn)了run方法，方法中主要有以下功能：

請(qǐng)求最大連接數(shù)限制：最大為 8*1024；請(qǐng)你注意到達(dá)最大連接數(shù)后操作系統(tǒng)底層還是會(huì)接收客戶端連接，但用戶層已經(jīng)不再接收
獲取socketChannel

public void run() {
        int errorDelay = 0;
        try {
            // Loop until we receive a shutdown command
            while (!stopCalled) {
					...
                if (stopCalled) {
                    break;
                }
                state = AcceptorState.RUNNING;

                try {
                    //if we have reached max connections, wait
                    // 如果連接超過(guò)了 8*1024，則線程阻塞等待; 是使用org.apache.tomcat.util.threads.LimitLatch類實(shí)現(xiàn)了分享鎖(內(nèi)部實(shí)現(xiàn)了AbstractQueuedSynchronizer)
                    // 請(qǐng)你注意到達(dá)最大連接數(shù)后操作系統(tǒng)底層還是會(huì)接收客戶端連接，但用戶層已經(jīng)不再接收。
                    endpoint.countUpOrAwaitConnection();

                    // Endpoint might have been paused while waiting for latch
                    // If that is the case, don't accept new connections
                    if (endpoint.isPaused()) {
                        continue;
                    }

                    U socket = null;
                    try {
                        // Accept the next incoming connection from the server
                        // socket
                        // 抽象方法，不同的endPoint有不同的實(shí)現(xiàn)方法。NioEndPoint為例，實(shí)現(xiàn)方法為serverSock.accept()，這個(gè)方法主要看serverSock實(shí)例化時(shí)如果為阻塞，accept方法為阻塞；反之為立即返回，如果沒(méi)有socket鏈接，則為null
                        socket = endpoint.serverSocketAccept();
                    } catch (Exception ioe) {
                        // We didn't get a socket
                        endpoint.countDownConnection();
                        if (endpoint.isRunning()) {
                            // Introduce delay if necessary
                            errorDelay = handleExceptionWithDelay(errorDelay);
                            // re-throw
                            throw ioe;
                        } else {
                            break;
                        }
                    }
                    // Successful accept, reset the error delay
                    errorDelay = 0;

                    // Configure the socket
                    if (!stopCalled && !endpoint.isPaused()) {
                        // setSocketOptions() will hand the socket off to
                        // an appropriate processor if successful
                        // endPoint類的抽象方法，不同的endPoint有不同的實(shí)現(xiàn)。處理獲取到的socketChannel鏈接，如果該socket鏈接能正常處理，那么該方法會(huì)返回true，否則為false
                        if (!endpoint.setSocketOptions(socket)) {
                            endpoint.closeSocket(socket);
                        }
                    } else {
                        endpoint.destroySocket(socket);
                    }
                } catch (Throwable t) {
                    ...
                }
            }
        } finally {
            stopLatch.countDown();
        }
        state = AcceptorState.ENDED;
    }

再來(lái)看下org.apache.tomcat.util.net.NioEndpoint#setSocketOptions方法的具體實(shí)現(xiàn)(NioEndpoint為例)

這個(gè)方法中主要做的事：

創(chuàng)建NioChannel
設(shè)置socket為非阻塞
將socket添加到Poller的隊(duì)列中

 protected boolean setSocketOptions(SocketChannel socket) {
        NioSocketWrapper socketWrapper = null;
        try {
            // Allocate channel and wrapper
            // 優(yōu)先使用已有的緩存nioChannel
            NioChannel channel = null;
            if (nioChannels != null) {
                channel = nioChannels.pop();
            }
            if (channel == null) {
                SocketBufferHandler bufhandler = new SocketBufferHandler(
                        socketProperties.getAppReadBufSize(),
                        socketProperties.getAppWriteBufSize(),
                        socketProperties.getDirectBuffer());
                if (isSSLEnabled()) {
                    channel = new SecureNioChannel(bufhandler, this);
                } else {
                    channel = new NioChannel(bufhandler);
                }
            }
            // 將nioEndpoint與NioChannel進(jìn)行包裝
            NioSocketWrapper newWrapper = new NioSocketWrapper(channel, this);
            channel.reset(socket, newWrapper);
            connections.put(socket, newWrapper);
            socketWrapper = newWrapper;

            // Set socket properties
            // Disable blocking, polling will be used
            // 設(shè)置當(dāng)前鏈接的socket為非阻塞
            socket.configureBlocking(false);
            if (getUnixDomainSocketPath() == null) {
                socketProperties.setProperties(socket.socket());
            }

            socketWrapper.setReadTimeout(getConnectionTimeout());
            socketWrapper.setWriteTimeout(getConnectionTimeout());
            socketWrapper.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
            // 將包裝后的nioChannel與nioEndpoint進(jìn)行注冊(cè)，注冊(cè)到Poller，將對(duì)應(yīng)的socket包裝類添加到Poller的隊(duì)列中，同時(shí)喚醒selector
            poller.register(socketWrapper);
            return true;
        } catch (Throwable t) {
            ExceptionUtils.handleThrowable(t);
            try {
                log.error(sm.getString("endpoint.socketOptionsError"), t);
            } catch (Throwable tt) {
                ExceptionUtils.handleThrowable(tt);
            }
            if (socketWrapper == null) {
                destroySocket(socket);
            }
        }
        // Tell to close the socket if needed
        return false;
    }

Socket請(qǐng)求輪詢

上一小節(jié)是接收到了socket請(qǐng)求，進(jìn)行包裝之后，將socket添加到了Poller的隊(duì)列上，并可能喚醒了Selector，本小節(jié)就來(lái)看看，Poller是如何進(jìn)行socket的輪詢的。

首先org.apache.tomcat.util.net.NioEndpoint.Poller也是實(shí)現(xiàn)了Runnable接口，是一個(gè)可以單獨(dú)啟動(dòng)的線程

初始化及啟動(dòng)是在org.apache.tomcat.util.net.NioEndpoint#startInternal

重要的屬性：

java.nio.channels.Selector：在Poller對(duì)象初始化的時(shí)候，就會(huì)啟動(dòng)輪詢器
SynchronizedQueue<PollerEvent>：同步的事件隊(duì)列

再來(lái)看下具體處理邏輯，run方法的源碼

		public void run() {
            // Loop until destroy() is called
            while (true) {

                boolean hasEvents = false;

                try {
                    if (!close) {
                        // 去SynchronizedQueue事件隊(duì)列中拉去，看是否已經(jīng)有了事件，如果有，則返回true
                        // 如果從隊(duì)列中拉取到了event(即上一步將NioSocketWrapper封裝為PollerEvent添加到次隊(duì)列中)，將socketChannel注冊(cè)到Selector上，標(biāo)記為SelectionKey.OP_READ，添加處理函數(shù)attachment(為Accetpor添加到Poller時(shí)的    
                        // NioSocketWrapper)
                        hasEvents = events();
                        if (wakeupCounter.getAndSet(-1) > 0) {
                            // If we are here, means we have other stuff to do
                            // Do a non blocking select
                            keyCount = selector.selectNow();
                        } else {
                            keyCount = selector.select(selectorTimeout);
                        }
                        wakeupCounter.set(0);
                    }
                    if (close) {
                        events();
                        timeout(0, false);
                        try {
                            selector.close();
                        } catch (IOException ioe) {
                            log.error(sm.getString("endpoint.nio.selectorCloseFail"), ioe);
                        }
                        break;
                    }
                    // Either we timed out or we woke up, process events first
                    if (keyCount == 0) {
                        hasEvents = (hasEvents | events());
                    }
                } catch (Throwable x) {
                    ExceptionUtils.handleThrowable(x);
                    log.error(sm.getString("endpoint.nio.selectorLoopError"), x);
                    continue;
                }

                Iterator<SelectionKey> iterator =
                    keyCount > 0 ? selector.selectedKeys().iterator() : null;
                // Walk through the collection of ready keys and dispatch
                // any active event.
                // selector輪詢獲取已經(jīng)注冊(cè)的事件，如果有事件準(zhǔn)備好，此時(shí)通過(guò)selectKeys方法就能拿到對(duì)應(yīng)的事件
                while (iterator != null && iterator.hasNext()) {
                    SelectionKey sk = iterator.next();
                    // 獲取到事件后，從迭代器刪除事件，防止事件重復(fù)輪詢
                    iterator.remove();
                    // 獲取事件的處理器，這個(gè)attachment是在event()方法中注冊(cè)的，后續(xù)這個(gè)事件的處理，就交給這個(gè)wrapper去處理
                    NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
                    // Attachment may be null if another thread has called
                    // cancelledKey()
                    if (socketWrapper != null) {
                        processKey(sk, socketWrapper);
                    }
                }

                // Process timeouts
                timeout(keyCount,hasEvents);
            }

            getStopLatch().countDown();
        }

在這里，有一個(gè)很重要的方法，org.apache.tomcat.util.net.NioEndpoint.Poller#events()，他是從Poller的事件隊(duì)列中獲取Acceptor接收到的可用socket，并將其注冊(cè)到Selector上

		/**
         * Processes events in the event queue of the Poller.
         *
         * @return <code>true</code> if some events were processed,
         *   <code>false</code> if queue was empty
         */
        public boolean events() {
            boolean result = false;

            PollerEvent pe = null;
            // 如果Acceptor將socket添加到隊(duì)列中，那么events.poll()方法就能拿到對(duì)應(yīng)的事件，否則拿不到就返回false
            for (int i = 0, size = events.size(); i < size && (pe = events.poll()) != null; i++ ) {
                result = true;
                NioSocketWrapper socketWrapper = pe.getSocketWrapper();
                SocketChannel sc = socketWrapper.getSocket().getIOChannel();
                int interestOps = pe.getInterestOps();
                if (sc == null) {
                    log.warn(sm.getString("endpoint.nio.nullSocketChannel"));
                    socketWrapper.close();
                } else if (interestOps == OP_REGISTER) {
                    // 如果是Acceptor剛添加到隊(duì)列中的事件，那么此時(shí)的ops就是OP_REGISTER
                    try {，
                        // 將次socket注冊(cè)到selector上，標(biāo)記為OP_READ事件，添加事件觸發(fā)時(shí)處理函數(shù)socketWrapper
                        sc.register(getSelector(), SelectionKey.OP_READ, socketWrapper);
                    } catch (Exception x) {
                        log.error(sm.getString("endpoint.nio.registerFail"), x);
                    }
                } else {
                    // ？？這里的邏輯，不清楚什么情況下會(huì)進(jìn)入到這個(gè)分支里面
                    final SelectionKey key = sc.keyFor(getSelector());
                    if (key == null) {
                        // The key was cancelled (e.g. due to socket closure)
                        // and removed from the selector while it was being
                        // processed. Count down the connections at this point
                        // since it won't have been counted down when the socket
                        // closed.
                        socketWrapper.close();
                    } else {
                        final NioSocketWrapper attachment = (NioSocketWrapper) key.attachment();
                        if (attachment != null) {
                            // We are registering the key to start with, reset the fairness counter.
                            try {
                                int ops = key.interestOps() | interestOps;
                                attachment.interestOps(ops);
                                key.interestOps(ops);
                            } catch (CancelledKeyException ckx) {
                                cancelledKey(key, socketWrapper);
                            }
                        } else {
                            cancelledKey(key, socketWrapper);
                        }
                    }
                }
                if (running && !paused && eventCache != null) {
                    pe.reset();
                    eventCache.push(pe);
                }
            }

            return result;
        }

還有一個(gè)重要方法就是org.apache.tomcat.util.net.NioEndpoint.Poller#processKey，上一個(gè)方法是獲取event，并注冊(cè)到selector，那這個(gè)方法就是通過(guò)Selector獲取到的數(shù)據(jù)準(zhǔn)備好的event，并開(kāi)始封裝成對(duì)應(yīng)的業(yè)務(wù)處理線程SocketProcessorBase，扔到線程池里開(kāi)始處理

	    protected void processKey(SelectionKey sk, NioSocketWrapper socketWrapper) {
            try {
                if (close) {
                    cancelledKey(sk, socketWrapper);
                } else if (sk.isValid()) {
                    if (sk.isReadable() || sk.isWritable()) {
                        if (socketWrapper.getSendfileData() != null) {
                            processSendfile(sk, socketWrapper, false);
                        } else {
                            unreg(sk, socketWrapper, sk.readyOps());
                            boolean closeSocket = false;
                            // Read goes before write
                            if (sk.isReadable()) {
                                //這里如果是異步的操作，就會(huì)走這里
                                if (socketWrapper.readOperation != null) {
                                    if (!socketWrapper.readOperation.process()) {
                                        closeSocket = true;
                                    }
                                } else if (socketWrapper.readBlocking) {
                                    // readBlocking默認(rèn)為false
                                    synchronized (socketWrapper.readLock) {
                                        socketWrapper.readBlocking = false;
                                        socketWrapper.readLock.notify();
                                    }
                                } else if (!processSocket(socketWrapper, SocketEvent.OPEN_READ, true)) {
                                    // 處理正常的事件，這里的processSocket就要正式開(kāi)始處理請(qǐng)求了。
                                    // 將對(duì)應(yīng)的事件封裝成對(duì)應(yīng)的線程，然后交給線程池去處理正式的請(qǐng)求業(yè)務(wù)
                                    closeSocket = true;
                                }
                            }
                            if (!closeSocket && sk.isWritable()) {
                                if (socketWrapper.writeOperation != null) {
                                    if (!socketWrapper.writeOperation.process()) {
                                        closeSocket = true;
                                    }
                                } else if (socketWrapper.writeBlocking) {
                                    synchronized (socketWrapper.writeLock) {
                                        socketWrapper.writeBlocking = false;
                                        socketWrapper.writeLock.notify();
                                    }
                                } else if (!processSocket(socketWrapper, SocketEvent.OPEN_WRITE, true)) {
                                    closeSocket = true;
                                }
                            }
                            if (closeSocket) {
                                cancelledKey(sk, socketWrapper);
                            }
                        }
                    }
                } else {
                    // Invalid key
                    cancelledKey(sk, socketWrapper);
                }
            } catch (CancelledKeyException ckx) {
                cancelledKey(sk, socketWrapper);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                log.error(sm.getString("endpoint.nio.keyProcessingError"), t);
            }
        }

請(qǐng)求具體處理

上一步，Selector獲取到了就緒的請(qǐng)求socket，然后根據(jù)socket注冊(cè)的觸發(fā)處理函數(shù)等，將這些數(shù)據(jù)進(jìn)行封裝，扔到了線程池里，開(kāi)始具體的業(yè)務(wù)邏輯處理。本節(jié)就是從工作線程封裝開(kāi)始，org.apache.tomcat.util.net.SocketProcessorBase為工作線程類的抽象類，實(shí)現(xiàn)了Runnable接口，不同的Endpoint實(shí)現(xiàn)具體的處理邏輯，本節(jié)以NioEndpoint為例

以下為org.apache.tomcat.util.net.AbstractEndpoint#processSocket方法源碼

    /**
     * Process the given SocketWrapper with the given status. Used to trigger
     * processing as if the Poller (for those endpoints that have one)
     * selected the socket.
     *
     * @param socketWrapper The socket wrapper to process
     * @param event         The socket event to be processed
     * @param dispatch      Should the processing be performed on a new
     *                          container thread
     *
     * @return if processing was triggered successfully
     */
    public boolean processSocket(SocketWrapperBase<S> socketWrapper,
            SocketEvent event, boolean dispatch) {
        try {
            if (socketWrapper == null) {
                return false;
            }
            // 優(yōu)先使用已經(jīng)存在的線程
            SocketProcessorBase<S> sc = null;
            if (processorCache != null) {
                sc = processorCache.pop();
            }
            if (sc == null) {
                sc = createSocketProcessor(socketWrapper, event);
            } else {
                sc.reset(socketWrapper, event);
            }
            // 獲取線程池。線程池的初始化，是在Acceptor、Poller這兩個(gè)單獨(dú)線程啟動(dòng)之前創(chuàng)建
            // tomcat使用了自定義的org.apache.tomcat.util.threads.TaskQueue，這塊tomcat也進(jìn)行了小的適配開(kāi)發(fā)
            // 核心線程為10個(gè)，最大200線程
            Executor executor = getExecutor();
            if (dispatch && executor != null) {
                executor.execute(sc);
            } else {
                sc.run();
            }
        } catch (RejectedExecutionException ree) {
            getLog().warn(sm.getString("endpoint.executor.fail", socketWrapper) , ree);
            return false;
        } catch (Throwable t) {
            ExceptionUtils.handleThrowable(t);
            // This means we got an OOM or similar creating a thread, or that
            // the pool and its queue are full
            getLog().error(sm.getString("endpoint.process.fail"), t);
            return false;
        }
        return true;
    }

上面的方法是得到了處理業(yè)務(wù)邏輯的線程SocketProcessorBase，NioEndpoint內(nèi)部類org.apache.tomcat.util.net.NioEndpoint.SocketProcessor繼承了這個(gè)抽象類，也就是具體的業(yè)務(wù)處理邏輯在org.apache.tomcat.util.net.NioEndpoint.SocketProcessor#doRun方法中，最終調(diào)用到我們的Servlet

        protected void doRun() {
            /*
             * Do not cache and re-use the value of socketWrapper.getSocket() in
             * this method. If the socket closes the value will be updated to
             * CLOSED_NIO_CHANNEL and the previous value potentially re-used for
             * a new connection. That can result in a stale cached value which
             * in turn can result in unintentionally closing currently active
             * connections.
             */
            Poller poller = NioEndpoint.this.poller;
            if (poller == null) {
                socketWrapper.close();
                return;
            }

            try {
                int handshake = -1;
                try {
                    // 握手相關(guān)判斷邏輯
                   ... 
                } catch (IOException x) {
                  ...
                }
                // 三次握手成功了
                if (handshake == 0) {
                    SocketState state = SocketState.OPEN;
                    // Process the request from this socket
                    // event為SocketEvent.OPEN_READ，這個(gè)變量是org.apache.tomcat.util.net.NioEndpoint.Poller#processKey方法賦值
                    if (event == null) {
                        state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
                    } else {
                        // 這里就開(kāi)始正式處理請(qǐng)求了
                        state = getHandler().process(socketWrapper, event);
                    }
                    if (state == SocketState.CLOSED) {
                        poller.cancelledKey(getSelectionKey(), socketWrapper);
                    }
                } else if (handshake == -1 ) {
                    getHandler().process(socketWrapper, SocketEvent.CONNECT_FAIL);
                    poller.cancelledKey(getSelectionKey(), socketWrapper);
                } else if (handshake == SelectionKey.OP_READ){
                    socketWrapper.registerReadInterest();
                } else if (handshake == SelectionKey.OP_WRITE){
                    socketWrapper.registerWriteInterest();
                }
            } catch (CancelledKeyException cx) {
                poller.cancelledKey(getSelectionKey(), socketWrapper);
            } catch (VirtualMachineError vme) {
                ExceptionUtils.handleThrowable(vme);
            } catch (Throwable t) {
                log.error(sm.getString("endpoint.processing.fail"), t);
                poller.cancelledKey(getSelectionKey(), socketWrapper);
            } finally {
                socketWrapper = null;
                event = null;
                //return to cache
                if (running && !paused && processorCache != null) {
                    processorCache.push(this);
                }
            }
        }

總結(jié)

Tomcat是如何接收網(wǎng)絡(luò)請(qǐng)求？

使用java nio的同步非阻塞去進(jìn)行網(wǎng)絡(luò)監(jiān)聽(tīng)。

org.apache.tomcat.util.net.AbstractEndpoint#bindWithCleanup中初始化網(wǎng)絡(luò)監(jiān)聽(tīng)、SSL

		{	
            ....
            serverSock = ServerSocketChannel.open();
            socketProperties.setProperties(serverSock.socket());
            InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset());
            // 當(dāng)應(yīng)用層面的連接數(shù)到達(dá)最大值時(shí)，操作系統(tǒng)可以繼續(xù)接收連接，那么操作系統(tǒng)能繼續(xù)接收的最大連接數(shù)就是這個(gè)隊(duì)列長(zhǎng)度，可以通過(guò)acceptCount 參數(shù)配置，默認(rèn)是 100
            serverSock.bind(addr, getAcceptCount());
        }
        serverSock.configureBlocking(true); //mimic APR behavior

org.apache.tomcat.util.net.NioEndpoint#startInternal中初始化業(yè)務(wù)處理的線程池、連接限制器、Poller線程、Acceptor線程

如何做到高性能的http協(xié)議服務(wù)器？

Tomcat把接收連接、檢測(cè) I/O 事件以及處理請(qǐng)求進(jìn)行了拆分，用不同規(guī)模的線程去做對(duì)應(yīng)的事情，這也是tomcat能高并發(fā)處理請(qǐng)求的原因。不讓線程阻塞，盡量讓CPU忙起來(lái)

是怎么設(shè)計(jì)的呢？

通過(guò)接口、抽象類等，將不同的處理邏輯拆分，各司其職
- org.apache.tomcat.util.net.AbstractEndpoint：I/O事件的檢測(cè)、處理邏輯都在這個(gè)類的實(shí)現(xiàn)類里面。使用模板方法，不同的協(xié)議有不同的實(shí)現(xiàn)方法。NioEndpoint/Nio2Endpoint/AprEndpoint
  - org.apache.tomcat.util.net.NioEndpoint.Poller：引用了java.nio.channels.Selector，內(nèi)部有個(gè)事件隊(duì)列，監(jiān)聽(tīng)I(yíng)/O事件具體就是在這里做的
  - org.apache.tomcat.util.net.NioEndpoint.NioSocketWrapper
  - org.apache.tomcat.util.net.NioEndpoint.SocketProcessor: 具體處理請(qǐng)求的線程類

參考：

NioEndpoint組件：Tomcat如何實(shí)現(xiàn)非阻塞I/O？

Java NIO淺析

到此這篇關(guān)于Apache Tomcat如何高并發(fā)處理請(qǐng)求的文章就介紹到這了,更多相關(guān)Apache Tomcat高并發(fā)請(qǐng)求內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: