Apache?Tomcat如何高并發(fā)處理請求
更新時間：2022年03月29日 16:45:48 作者：chenzw93
本文主要介紹了Apache?Tomcat如何高并發(fā)處理請求，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
介紹
作為常用的http協(xié)議服務(wù)器，tomcat應(yīng)用非常廣泛。tomcat也是遵循Servelt協(xié)議的，Servelt協(xié)議可以讓服務(wù)器與真實服務(wù)邏輯代碼進行解耦。各自只需要關(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接口，在一個單獨的線程中以死循環(huán)的方式一直進行socket的監(jiān)聽
線程的初始化及啟動是在方法org.apache.tomcat.util.net.AbstractEndpoint#startAcceptorThread
有個很重要的屬性org.apache.tomcat.util.net.AbstractEndpoint；同時實現(xiàn)了run方法，方法中主要有以下功能：
- 請求最大連接數(shù)限制： 最大為 8*1024；請你注意到達最大連接數(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)
// 請你注意到達最大連接數(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鏈接，則為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有不同的實現(xiàn)。處理獲取到的socketChannel鏈接，如果該socket鏈接能正常處理，那么該方法會返回true，否則為false
if (!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進行包裝
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è)置當前鏈接的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，將對應(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請求，進行包裝之后，將socket添加到了Poller的隊列上，并可能喚醒了Selector，本小節(jié)就來看看，Poller是如何進行socket的輪詢的。
首先org.apache.tomcat.util.net.NioEndpoint.Poller也是實現(xiàn)了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事件隊列中拉去，看是否已經(jīng)有了事件，如果有，則返回true
// 如果從隊列中拉取到了event(即上一步將NioSocketWrapper封裝為PollerEvent添加到次隊列中)，將socketChannel注冊到Selector上，標記為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)注冊的事件，如果有事件準備好，此時通過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上，標記為OP_READ事件，添加事件觸發(fā)時處理函數(shù)socketWrapper
sc.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獲取到的數(shù)據(jù)準備好的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默認為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ù)進行封裝，扔到了線程池里，開始具體的業(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這兩個單獨線程啟動之前創(chuàng)建
// tomcat使用了自定義的org.apache.tomcat.util.threads.TaskQueue，這塊tomcat也進行了小的適配開發(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的同步非阻塞去進行網(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()); // 當應(yīng)用層面的連接數(shù)到達最大值時，操作系統(tǒng)可以繼續(xù)接收連接，那么操作系統(tǒng)能繼續(xù)接收的最大連接數(shù)就是這個隊列長度，可以通過acceptCount 參數(shù)配置，默認是 100 serverSock.bind(addr, getAcceptCount()); } serverSock.configureBlocking(true); //mimic APR behaviororg.apache.tomcat.util.net.NioEndpoint#startInternal中初始化業(yè)務(wù)處理的線程池、連接限制器、Poller線程、Acceptor線程如何做到高性能的
http協(xié)議服務(wù)器？Tomcat把接收連接、檢測 I/O 事件以及處理請求進行了拆分，用不同規(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.NioSocketWrapperorg.apache.tomcat.util.net.NioEndpoint.SocketProcessor: 具體處理請求的線程類
參考：
NioEndpoint組件：Tomcat如何實現(xiàn)非阻塞I/O？
到此這篇關(guān)于Apache Tomcat如何高并發(fā)處理請求 的文章就介紹到這了,更多相關(guān)Apache Tomcat高并發(fā)請求 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
開發(fā)者開發(fā)部署web應(yīng)用時通常使用tomcat服務(wù)器，這篇文章主要介紹了tomcat的配置使用詳細版(小結(jié))，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-05-05
本文主要介紹了Tomca啟動閃退問題解決，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-07-07
這篇文章主要帶大家從源碼級深入剖析一下Tomcat類的加載原理，文中的示例代碼講解詳細，具有一定的學習價值，感興趣的小伙伴可以跟隨小編一起學習一下2023-06-06
tomcat雙擊startup.bat一閃而過(閃退)的問題解決
本文主要介紹了tomcat雙擊startup.bat一閃而過(閃退)的問題解決,可能是tomcat找不到你的jdk環(huán)境配置路徑,下面就來介紹一下解決方法,感興趣的可以了解一下2024-05-05
這篇文章主要介紹了關(guān)于tomcat下啟動solr遇到的問題及解決,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-07-07
這篇文章主要給大家介紹了關(guān)于tomcat8改了jar加載順序的踩坑記錄，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧2019-08-08
這篇文章主要給大家介紹了關(guān)于解決應(yīng)用啟動失敗但tomcat不報錯的方法，文中介紹的非常詳細，對大家具有一定的參考學習價值，需要的朋友們下面來一起看看吧。2017-06-06
這篇文章主要介紹了Tomcat多域名訪問詳解，具有一定參考價值，需要的朋友可以了解下。2017-11-11
Tomcat6.0與windows 2003 server 的IIS服務(wù)器集成
本例主要講解Tomcat6.0與windows 2003 server 的IIS服務(wù)器集成的問題，用到的工具版 本如下：jdk是6.0、Tomcat 6.0、windows 2003 server 的IIS。2009-08-08
這篇文章主要介紹了關(guān)于 Tomcat進程意外退出的問題解析的相關(guān)資料,非常不錯，具有參考借鑒價值，需要的朋友可以參考下2016-11-11