springboot內(nèi)置tomcat之NIO處理流程一覽

第springboot內(nèi)置tomcat之NIO處理流程一覽目錄前言tomcat組件Acceptor組件Poller總結(jié)大致流程為相較于BIO模型的tomcat，NIO的優(yōu)勢分析

前言

springboot內(nèi)置的tomcat目前默認(rèn)是基于NIO來實現(xiàn)的，本文介紹下tomcat接受請求的一些組件及組件之間的關(guān)聯(lián)

tomcat組件

本文只介紹NIO中tomcat的組件

我們直接看NIO的核心類NioEndpoint的startInternal方法

Acceptor組件

publicvoidstartInternal()throwsException{

if(!running){

running=true;

paused=false;

processorCache=newSynchronizedStack(SynchronizedStack.DEFAULT_SIZE,

socketProperties.getProcessorCache());

eventCache=newSynchronizedStack(SynchronizedStack.DEFAULT_SIZE,

socketProperties.getEventCache());

nioChannels=newSynchronizedStack(SynchronizedStack.DEFAULT_SIZE,

socketProperties.getBufferPool());

//Createworkercollection

if(getExecutor()==null){

createExecutor();

initializeConnectionLatch();

//Startpollerthreads

//核心代碼1

pollers=newPoller[getPollerThreadCount()];

for(inti=0;ipollers.length;i++){

pollers[i]=newPoller();

ThreadpollerThread=newThread(pollers[i],getName()+"-ClientPoller-"+i);

pollerThread.setPriority(threadPriority);

pollerThread.setDaemon(true);

pollerThread.start();

//核心代碼2

startAcceptorThreads();

看核心代碼1的位置構(gòu)造了一個Poller數(shù)組，Poller是一個實現(xiàn)了Runnable的類，并且啟動了該線程類，

getPollerThreadCount()方法返回了2和當(dāng)前物理機CPU內(nèi)核數(shù)的最小值，即創(chuàng)建的數(shù)組最大值為2

接下來看核心代碼2startAcceptorThreads()

protectedfinalvoidstartAcceptorThreads(){

intcount=getAcceptorThreadCount();

acceptors=newArrayList(count);

for(inti=0;icount;i++){

AcceptorUacceptor=newAcceptor(this);

StringthreadName=getName()+"-Acceptor-"+i;

acceptor.setThreadName(threadName);

acceptors.add(acceptor);

Threadt=newThread(acceptor,threadName);

t.setPriority(getAcceptorThreadPriority());

t.setDaemon(getDaemon());

t.start();

創(chuàng)建了多個Acceptor類，Acceptor也是實現(xiàn)了Runnable的線程類，創(chuàng)建個數(shù)默認(rèn)是1

然后我們看下Acceptor啟動后做了什么，我們直接看run方法

publicvoidrun(){

......

Usocket=null;

try{

//Acceptthenextincomingconnectionfromtheserver

//socket

//核心代碼1

socket=endpoint.serverSocketAccept();

}catch(Exceptionioe){

......

//Successfulaccept,resettheerrordelay

errorDelay=0;

//Configurethesocket

if(endpoint.isRunning()!endpoint.isPaused()){

//setSocketOptions()willhandthesocketoffto

//anappropriateprocessorifsuccessful

//核心代碼2

if(!endpoint.setSocketOptions(socket)){

endpoint.closeSocket(socket);

......

核心代碼1很明顯是一個阻塞模型，即接受客戶端連接的，當(dāng)沒有客戶端連接時會處于阻塞，這里可以看到默認(rèn)情況下tomcat在nio模式下只有一個Acceptor線程類來接受連接

然后看核心代碼2

protectedbooleansetSocketOptions(SocketChannelsocket){

//Processtheconnection

try{

//disableblocking,APRstyle,wearegonnabepollingit

socket.configureBlocking(false);

Socketsock=socket.socket();

socketProperties.setProperties(sock);

NioChannelchannel=nioChannels.pop();

if(channel==null){

SocketBufferHandlerbufhandler=newSocketBufferHandler(

socketProperties.getAppReadBufSize(),

socketProperties.getAppWriteBufSize(),

socketProperties.getDirectBuffer());

if(isSSLEnabled()){

channel=newSecureNioChannel(socket,bufhandler,selectorPool,this);

}else{

channel=newNioChannel(socket,bufhandler);

}else{

channel.setIOChannel(socket);

channel.reset();

//核心代碼

getPoller0().register(channel);

}catch(Throwablet){

......

returntrue;

我們看核心代碼getPoller0().register(channel)

publicvoidregister(finalNioChannelsocket){

socket.setPoller(this);

NioSocketWrapperka=newNioSocketWrapper(socket,NioEndpoint.this);

socket.setSocketWrapper(ka);

ka.setPoller(this);

ka.setReadTimeout(getConnectionTimeout());

ka.setWriteTimeout(getConnectionTimeout());

ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());

ka.setSecure(isSSLEnabled());

PollerEventr=eventCache.pop();

erestOps(SelectionKey.OP_READ);//thisiswhatOP_REGISTERturnsinto.

if(r==null)r=newPollerEvent(socket,ka,OP_REGISTER);

elser.reset(socket,ka,OP_REGISTER);

//核心代碼

addEvent(r);

看addEvent

privatevoidaddEvent(PollerEventevent){

events.offer(event);

if(wakeupCounter.incrementAndGet()==0)selector.wakeup();

events的定義

privatefinalSynchronizedQueuePollerEventevents=

newSynchronizedQueue();

這里可以看到封裝了一個PollerEvent并且扔到了一個隊列里面，然后當(dāng)前類就結(jié)束了

由此可得Acceptor的作用就是接受客戶端連接，并且把連接封裝起來扔到了一個隊列中

Poller

我們前面已經(jīng)創(chuàng)建并且啟動了多個Poller線程類，默認(rèn)的數(shù)量是小于等于2的。

然后我們看下Poller類做了什么，同樣我們看run方法

@Override

publicvoidrun(){

//Loopuntildestroy()iscalled

while(true){

booleanhasEvents=false;

try{

if(!close){

//核心代碼1

hasEvents=events();

.......

IteratorSelectionKeyiterator=

keyCount0selector.selectedKeys().iterator():null;

//Walkthroughthecollectionofreadykeysanddispatch

//anyactiveevent.

while(iterator!=nulliterator.hasNext()){

SelectionKeysk=iterator.next();

NioSocketWrapperattachment=(NioSocketWrapper)sk.attachment();

//Attachmentmaybenullifanotherthreadhascalled

//cancelledKey()

if(attachment==null){

iterator.remove();

}else{

iterator.remove();

//核心代碼2

processKey(sk,attachment);

}//while

//processtimeouts

timeout(keyCount,hasEvents);

}//while

getStopLatch().countDown();

先看核心代碼1hasEvents=events()

publicbooleanevents(){

booleanresult=false;

PollerEventpe=null;

for(inti=0,size=events.size();isize(pe=events.poll())!=null;i++){

result=true;

try{

//核心代碼

pe.run();

pe.reset();

if(running!paused){

eventCache.push(pe);

}catch(Throwablex){

log.error("",x);

returnresult;

核心代碼run

@Override

publicvoidrun(){

if(interestOps==OP_REGISTER){

try{

//核心代碼，注冊到selector輪訓(xùn)器

socket.getIOChannel().register(

socket.getPoller().getSelector(),SelectionKey.OP_READ,socketWrapper);

}catch(Exceptionx){

log.error(sm.getString("endpoint.nio.registerFail"),x);

}else{

......

可以看出大概的意思就是從剛才我們放進去的隊列events里面取數(shù)據(jù)放到了eventCache里面，eventCache的定義SynchronizedStackeventCache，當(dāng)取到數(shù)據(jù)后返回true，這個時候就會進入核心代碼2處的processKey(sk,attachment)，也就是開始處理請求了

protectedvoidprocessKey(SelectionKeysk,NioSocketWrapperattachment){

try{

if(close){

cancelledKey(sk);

}elseif(sk.isValid()attachment!=null){

if(sk.isReadable()||sk.isWritable()){

if(attachment.getSendfileData()!=null){

processSendfile(sk,attachment,false);

}else{

unreg(sk,attachment,sk.readyOps());

booleancloseSocket=false;

//Readgoesbeforewrite

if(sk.isReadable()){

//核心代碼

if(!processSocket(attachment,SocketEvent.OPEN_READ,true)){

closeSocket=true;

if(!closeSocketsk.isWritable()){

if(!processSocket(attachment,SocketEvent.OPEN_WRITE,true)){

closeSocket=true;

......

這里就可以看到我們的NIO的模型了，也就是多路復(fù)用的模型，輪詢來判斷key的狀態(tài)，當(dāng)key是可讀或者可寫時執(zhí)行processSocket

publicbooleanprocessSocket(SocketWrapperBaseSsocketWrapper,

SocketEventevent,booleandispatch){

try{

if(socketWrapper==null){

returnfalse;

SocketProcessorBaseSsc=processorCache.pop();

if(sc==null){

sc=createSocketProcessor(socketWrapper,event);

}else{

sc.reset(socketWrapper,event);

//核心代碼

Executorexecutor=getExecutor();

if(dispatchexecutor!=null){

executor.execute(sc);

}else{

sc.run();

}catch(RejectedExecutionExceptionree){

getLog().warn(sm.getString("endpoint.executor.fail",socketWrapper),ree);

returnfalse;

}catch(Throwablet){

ExceptionUtils.handleThrowable(t);

//ThismeanswegotanOOMorsimilarcreatingathread,orthat

//thepoolanditsqueuearefull

getLog().error(sm.getString("cess.fail"),t);

returnfalse;

returntrue;

這里就是核心代碼了，可以看到getExecutor()方法，獲取線程池，這個線程池是在初始化tomcat時提前初始化好的，默認(rèn)情況下核心線程是10，最大線程是200。線程池的配置可以根據(jù)我們自己配