Apache?Tomcat如何高并發(fā)處理請求
介紹
作為常用的http
協(xié)議服務(wù)器,tomcat
應(yīng)用非常廣泛。tomcat也是遵循Servelt協(xié)議的,Servelt
協(xié)議可以讓服務(wù)器與真實服務(wù)邏輯代碼進(jìn)行解耦。各自只需要關(guān)注Servlet
協(xié)議即可。
對于tomcat是如何作為一個高性能的服務(wù)器的呢?你是不是也會有這樣的疑問?
tomcat是如何接收網(wǎng)絡(luò)請求?
如何做到高性能的http
協(xié)議服務(wù)器?
tomcat從8.0往后開始使用了NIO非阻塞io模型,提高了吞吐量,本文的源碼是tomcat 9.0.48版本
接收Socket請求
org.apache.tomcat.util.net.Acceptor
實現(xiàn)了Runnable
接口,在一個單獨(dú)的線程中以死循環(huán)的方式一直進(jìn)行socket的監(jiān)聽
線程的初始化及啟動是在方法org.apache.tomcat.util.net.AbstractEndpoint#startAcceptorThread
有個很重要的屬性org.apache.tomcat.util.net.AbstractEndpoint
;同時實現(xiàn)了run
方法,方法中主要有以下功能:
- 請求最大連接數(shù)限制: 最大為 8*1024;請你注意到達(dá)最大連接數(shù)后操作系統(tǒng)底層還是會接收客戶端連接,但用戶層已經(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 // 如果連接超過了 8*1024,則線程阻塞等待; 是使用org.apache.tomcat.util.threads.LimitLatch類實現(xiàn)了分享鎖(內(nèi)部實現(xiàn)了AbstractQueuedSynchronizer) // 請你注意到達(dá)最大連接數(shù)后操作系統(tǒng)底層還是會接收客戶端連接,但用戶層已經(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有不同的實現(xiàn)方法。NioEndPoint為例,實現(xiàn)方法為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有不同的實現(xiàn)。處理獲取到的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
方法的具體實現(xiàn)(NioEndpoint為例)
這個方法中主要做的事:
- 創(chuàng)建NioChannel
- 設(shè)置socket為非阻塞
- 將socket添加到Poller的隊列中
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)行注冊,注冊到Poller,將對應(yīng)的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 needed return false; }
Socket請求輪詢
上一小節(jié)是接收到了socket請求,進(jìn)行包裝之后,將socket添加到了Poller
的隊列上,并可能喚醒了Selector
,本小節(jié)就來看看,Poller是如何進(jìn)行socket的輪詢的。
首先org.apache.tomcat.util.net.NioEndpoint.Poller
也是實現(xiàn)了Runnable接口,是一個可以單獨(dú)啟動的線程
初始化及啟動是在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事件隊列中拉去,看是否已經(jīng)有了事件,如果有,則返回true// 如果從隊列中拉取到了event(即上一步將NioSocketWrapper封裝為PollerEvent添加到次隊列中),將socketChannel注冊到Selector上,標(biāo)記為SelectionKey.OP_READ,添加處理函數(shù)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輪詢獲取已經(jīng)注冊的事件,如果有事件準(zhǔn)備好,此時通過selectKeys方法就能拿到對應(yīng)的事件 while (iterator != null && iterator.hasNext()) { SelectionKey sk = iterator.next(); // 獲取到事件后,從迭代器刪除事件,防止事件重復(fù)輪詢 iterator.remove(); // 獲取事件的處理器,這個attachment是在event()方法中注冊的,后續(xù)這個事件的處理,就交給這個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(); }
在這里,有一個很重要的方法,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()方法就能拿到對應(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剛添加到隊列中的事件,那么此時的ops就是OP_REGISTER try {,// 將次socket注冊到selector上,標(biāo)記為OP_READ事件,添加事件觸發(fā)時處理函數(shù)socketWrappersc.register(getSelector(), SelectionKey.OP_READ, socketWrapper); } catch (Exception x) {log.error(sm.getString("endpoint.nio.registerFail"), x); } } else { // ??這里的邏輯,不清楚什么情況下會進(jìn)入到這個分支里面 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
獲取到的數(shù)據(jù)準(zhǔn)備好的event,并開始封裝成對應(yīng)的業(yè)務(wù)處理線程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默認(rèn)為false synchronized (socketWrapper.readLock) { socketWrapper.readBlocking = false; socketWrapper.readLock.notify(); } } else if (!processSocket(socketWrapper, SocketEvent.OPEN_READ, true)) { // 處理正常的事件,這里的processSocket就要正式開始處理請求了。 // 將對應(yīng)的事件封裝成對應(yī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); } }
請求具體處理
上一步,Selector獲取到了就緒的請求socket,然后根據(jù)socket注冊的觸發(fā)處理函數(shù)等,將這些數(shù)據(jù)進(jìn)行封裝,扔到了線程池里,開始具體的業(yè)務(wù)邏輯處理。本節(jié)就是從工作線程封裝開始,org.apache.tomcat.util.net.SocketProcessorBase
為工作線程類的抽象類,實現(xiàn)了Runnable接口,不同的Endpoint實現(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這兩個單獨(dú)線程啟動之前創(chuàng)建 // tomcat使用了自定義的org.apache.tomcat.util.threads.TaskQueue,這塊tomcat也進(jìn)行了小的適配開發(fā) // 核心線程為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; }
上面的方法是得到了處理業(yè)務(wù)邏輯的線程SocketProcessorBase,NioEndpoint內(nèi)部類org.apache.tomcat.util.net.NioEndpoint.SocketProcessor
繼承了這個抽象類,也就是具體的業(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,這個變量是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 cache if (running && !paused && processorCache != null) { processorCache.push(this); } } }
總結(jié)
Tomcat
是如何接收網(wǎng)絡(luò)請求?使用java nio的同步非阻塞去進(jìn)行網(wǎng)絡(luò)監(jiān)聽。
org.apache.tomcat.util.net.AbstractEndpoint#bindWithCleanup
中初始化網(wǎng)絡(luò)監(jiān)聽、SSL{ .... serverSock = ServerSocketChannel.open(); socketProperties.setProperties(serverSock.socket()); InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset()); // 當(dāng)應(yīng)用層面的連接數(shù)到達(dá)最大值時,操作系統(tǒng)可以繼續(xù)接收連接,那么操作系統(tǒng)能繼續(xù)接收的最大連接數(shù)就是這個隊列長度,可以通過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把接收連接、檢測 I/O 事件以及處理請求進(jìn)行了拆分,用不同規(guī)模的線程去做對應(yīng)的事情,這也是tomcat能高并發(fā)處理請求的原因。不讓線程阻塞,盡量讓CPU忙起來
是怎么設(shè)計的呢?
通過接口、抽象類等,將不同的處理邏輯拆分,各司其職
org.apache.tomcat.util.net.AbstractEndpoint
:I/O事件的檢測、處理邏輯都在這個類的實現(xiàn)類里面。使用模板方法,不同的協(xié)議有不同的實現(xiàn)方法。NioEndpoint/Nio2Endpoint/AprEndpointorg.apache.tomcat.util.net.NioEndpoint.Poller
:引用了java.nio.channels.Selector
,內(nèi)部有個事件隊列,監(jiān)聽I/O事件具體就是在這里做的org.apache.tomcat.util.net.NioEndpoint.NioSocketWrapper
org.apache.tomcat.util.net.NioEndpoint.SocketProcessor
: 具體處理請求的線程類
參考:
NioEndpoint組件:Tomcat如何實現(xiàn)非阻塞I/O?
Java NIO淺析
到此這篇關(guān)于Apache Tomcat如何高并發(fā)處理請求 的文章就介紹到這了,更多相關(guān)Apache Tomcat高并發(fā)請求 內(nèi)容請搜索本站以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持本站!
版權(quán)聲明:本站文章來源標(biāo)注為YINGSOO的內(nèi)容版權(quán)均為本站所有,歡迎引用、轉(zhuǎn)載,請保持原文完整并注明來源及原文鏈接。禁止復(fù)制或仿造本網(wǎng)站,禁止在非www.sddonglingsh.com所屬的服務(wù)器上建立鏡像,否則將依法追究法律責(zé)任。本站部分內(nèi)容來源于網(wǎng)友推薦、互聯(lián)網(wǎng)收集整理而來,僅供學(xué)習(xí)參考,不代表本站立場,如有內(nèi)容涉嫌侵權(quán),請聯(lián)系alex-e#qq.com處理。