目錄

• 介紹
• 接收Socket請求
• Socket請求輪詢
• 請求具體處理
• 總結
• 參考：

介紹

作為常用的http協議服務器，tomcat應用非常廣泛。tomcat也是遵循Servelt協議的，Servelt協議可以讓服務器與真實服務邏輯代碼進行解耦。各自只需要關注Servlet協議即可。
對于tomcat是如何作為一個高性能的服務器的呢？你是不是也會有這樣的疑問？

tomcat是如何接收網絡請求？

如何做到高性能的http協議服務器？

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

接收Socket請求

org.apache.tomcat.util.net.Acceptor實現了Runnable接口，在一個單獨的線程中以死循環的方式一直進行socket的監聽

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

有個很重要的屬性org.apache.tomcat.util.net.AbstractEndpoint；同時實現了run方法，方法中主要有以下功能：

• 請求最大連接數限制： 最大為 8*1024；請你注意到達最大連接數后操作系統底層還是會接收客戶端連接，但用戶層已經不再接收
• 獲取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    // 如果連接超過了 8*1024，則線程阻塞等待; 是使用org.apache.tomcat.util.threads.LimitLatch類實現了分享鎖(內部實現了AbstractQueuedSynchronizer)    // 請你注意到達最大連接數后操作系統底層還是會接收客戶端連接，但用戶層已經不再接收。    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有不同的實現方法。NioEndPoint為例，實現方法為serverSock.accept()，這個方法主要看serverSock實例化時如果為阻塞，accept方法為阻塞；反之為立即返回，如果沒有socket鏈接，則為nullsocket = endpoint.serverSocketAccept();    } catch (Exception ioe) {// We didn"t get a socketendpoint.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有不同的實現。處理獲取到的socketChannel鏈接，如果該socket鏈接能正常處理，那么該方法會返回true，否則為falseif (!endpoint.setSocketOptions(socket)) {    endpoint.closeSocket(socket);}    } else {endpoint.destroySocket(socket);    }} catch (Throwable t) {    ...}    }} finally {    stopLatch.countDown();}state = AcceptorState.ENDED;    }

再來看下org.apache.tomcat.util.net.NioEndpoint#setSocketOptions方法的具體實現(NioEndpoint為例)

這個方法中主要做的事：

• 創建NioChannel
• 設置socket為非阻塞
• 將socket添加到Poller的隊列中

 protected boolean setSocketOptions(SocketChannel socket) {NioSocketWrapper socketWrapper = null;try {    // Allocate channel and wrapper    // 優先使用已有的緩存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進行包裝    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    // 設置當前鏈接的socket為非阻塞    socket.configureBlocking(false);    if (getUnixDomainSocketPath() == null) {socketProperties.setProperties(socket.socket());    }    socketWrapper.setReadTimeout(getConnectionTimeout());    socketWrapper.setWriteTimeout(getConnectionTimeout());    socketWrapper.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());    // 將包裝后的nioChannel與nioEndpoint進行注冊，注冊到Poller，將對應的socket包裝類添加到Poller的隊列中，同時喚醒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 neededreturn false;    }

Socket請求輪詢

上一小節是接收到了socket請求，進行包裝之后，將socket添加到了Poller的隊列上，并可能喚醒了Selector，本小節就來看看，Poller是如何進行socket的輪詢的。

首先org.apache.tomcat.util.net.NioEndpoint.Poller也是實現了Runnable接口，是一個可以單獨啟動的線程

初始化及啟動是在org.apache.tomcat.util.net.NioEndpoint#startInternal

重要的屬性：

• java.nio.channels.Selector：在Poller對象初始化的時候，就會啟動輪詢器
• SynchronizedQueue<PollerEvent>：同步的事件隊列

再來看下具體處理邏輯，run方法的源碼

		public void run() {    // Loop until destroy() is called    while (true) {boolean hasEvents = false;try {    if (!close) {// 去SynchronizedQueue事件隊列中拉去，看是否已經有了事件，如果有，則返回true// 如果從隊列中拉取到了event(即上一步將NioSocketWrapper封裝為PollerEvent添加到次隊列中)，將socketChannel注冊到Selector上，標記為SelectionKey.OP_READ，添加處理函數attachment(為Accetpor添加到Poller時的    // 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輪詢獲取已經注冊的事件，如果有事件準備好，此時通過selectKeys方法就能拿到對應的事件while (iterator != null && iterator.hasNext()) {    SelectionKey sk = iterator.next();    // 獲取到事件后，從迭代器刪除事件，防止事件重復輪詢    iterator.remove();    // 獲取事件的處理器，這個attachment是在event()方法中注冊的，后續這個事件的處理，就交給這個wrapper去處理    NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();    // Attachment may be null if another thread has called    // cancelledKey()    if (socketWrapper != null) {processKey(sk, socketWrapper);    }}// Process timeoutstimeout(keyCount,hasEvents);    }    getStopLatch().countDown();}

在這里，有一個很重要的方法，org.apache.tomcat.util.net.NioEndpoint.Poller#events()，他是從Poller的事件隊列中獲取Acceptor接收到的可用socket，并將其注冊到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添加到隊列中，那么events.poll()方法就能拿到對應的事件，否則拿不到就返回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剛添加到隊列中的事件，那么此時的ops就是OP_REGISTER    try {，// 將次socket注冊到selector上，標記為OP_READ事件，添加事件觸發時處理函數socketWrappersc.register(getSelector(), SelectionKey.OP_READ, socketWrapper);    } catch (Exception x) {log.error(sm.getString("endpoint.nio.registerFail"), x);    }} else {    // ？？這里的邏輯，不清楚什么情況下會進入到這個分支里面    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;}

還有一個重要方法就是org.apache.tomcat.util.net.NioEndpoint.Poller#processKey，上一個方法是獲取event，并注冊到selector，那這個方法就是通過Selector獲取到的數據準備好的event，并開始封裝成對應的業務處理線程SocketProcessorBase，扔到線程池里開始處理

	    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()) {//這里如果是異步的操作，就會走這里if (socketWrapper.readOperation != null) {    if (!socketWrapper.readOperation.process()) {closeSocket = true;    }} else if (socketWrapper.readBlocking) {    // readBlocking默認為false    synchronized (socketWrapper.readLock) {socketWrapper.readBlocking = false;socketWrapper.readLock.notify();    }} else if (!processSocket(socketWrapper, SocketEvent.OPEN_READ, true)) {    // 處理正常的事件，這里的processSocket就要正式開始處理請求了。    // 將對應的事件封裝成對應的線程，然后交給線程池去處理正式的請求業務    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);    }}

請求具體處理

上一步，Selector獲取到了就緒的請求socket，然后根據socket注冊的觸發處理函數等，將這些數據進行封裝，扔到了線程池里，開始具體的業務邏輯處理。本節就是從工作線程封裝開始，org.apache.tomcat.util.net.SocketProcessorBase為工作線程類的抽象類，實現了Runnable接口，不同的Endpoint實現具體的處理邏輯，本節以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;    }    // 優先使用已經存在的線程    SocketProcessorBase<S> sc = null;    if (processorCache != null) {sc = processorCache.pop();    }    if (sc == null) {sc = createSocketProcessor(socketWrapper, event);    } else {sc.reset(socketWrapper, event);    }    // 獲取線程池。線程池的初始化，是在Acceptor、Poller這兩個單獨線程啟動之前創建    // tomcat使用了自定義的org.apache.tomcat.util.threads.TaskQueue，這塊tomcat也進行了小的適配開發    // 核心線程為10個，最大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;    }

上面的方法是得到了處理業務邏輯的線程SocketProcessorBase，NioEndpoint內部類org.apache.tomcat.util.net.NioEndpoint.SocketProcessor繼承了這個抽象類，也就是具體的業務處理邏輯在org.apache.tomcat.util.net.NioEndpoint.SocketProcessor#doRun方法中，最終調用到我們的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 {    // 握手相關判斷邏輯   ... } catch (IOException x) {  ...}// 三次握手成功了if (handshake == 0) {    SocketState state = SocketState.OPEN;    // Process the request from this socket    // event為SocketEvent.OPEN_READ，這個變量是org.apache.tomcat.util.net.NioEndpoint.Poller#processKey方法賦值    if (event == null) {state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);    } else {// 這里就開始正式處理請求了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 cacheif (running && !paused && processorCache != null) {    processorCache.push(this);}    }}

總結

• Tomcat是如何接收網絡請求？

  使用java nio的同步非阻塞去進行網絡監聽。

  org.apache.tomcat.util.net.AbstractEndpoint#bindWithCleanup中初始化網絡監聽、SSL

  		{	    ....    serverSock = ServerSocketChannel.open();    socketProperties.setProperties(serverSock.socket());    InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset());    // 當應用層面的連接數到達最大值時，操作系統可以繼續接收連接，那么操作系統能繼續接收的最大連接數就是這個隊列長度，可以通過acceptCount 參數配置，默認是 100    serverSock.bind(addr, getAcceptCount());}serverSock.configureBlocking(true); //mimic APR behavior

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

• 如何做到高性能的http協議服務器？

  Tomcat把接收連接、檢測 I/O 事件以及處理請求進行了拆分，用不同規模的線程去做對應的事情，這也是tomcat能高并發處理請求的原因。不讓線程阻塞，盡量讓CPU忙起來

• 是怎么設計的呢？

  通過接口、抽象類等，將不同的處理邏輯拆分，各司其職

  • org.apache.tomcat.util.net.AbstractEndpoint：I/O事件的檢測、處理邏輯都在這個類的實現類里面。使用模板方法，不同的協議有不同的實現方法。NioEndpoint/Nio2Endpoint/AprEndpoint
    • org.apache.tomcat.util.net.NioEndpoint.Poller：引用了java.nio.channels.Selector，內部有個事件隊列，監聽I/O事件具體就是在這里做的
    • org.apache.tomcat.util.net.NioEndpoint.NioSocketWrapper
    • org.apache.tomcat.util.net.NioEndpoint.SocketProcessor: 具體處理請求的線程類

參考：

到此這篇關于Apache Tomcat如何高并發處理請求 的文章就介紹到這了,更多相關Apache Tomcat高并發請求 內容請搜索以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持！