springboot内置tomcat之NIO处理流程一览

第springboot内置tomcat之NIO处理流程一览目录前言tomcat组件Acceptor组件Poller总结大致流程为相较于BIO模型的tomcat，NIO的优势分析

前言

springboot内置的tomcat目前默认是基于NIO来实现的，本文介绍下tomcat接受请求的一些组件及组件之间的关联

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的位置构造了一个Poller数组，Poller是一个实现了Runnable的类，并且启动了该线程类，

getPollerThreadCount()方法返回了2和当前物理机CPU内核数的最小值，即创建的数组最大值为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();

创建了多个Acceptor类，Acceptor也是实现了Runnable的线程类，创建个数默认是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很明显是一个阻塞模型，即接受客户端连接的，当没有客户端连接时会处于阻塞，这里可以看到默认情况下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并且扔到了一个队列里面，然后当前类就结束了

由此可得Acceptor的作用就是接受客户端连接，并且把连接封装起来扔到了一个队列中

Poller

我们前面已经创建并且启动了多个Poller线程类，默认的数量是小于等于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轮训器

socket.getIOChannel().register(

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

}catch(Exceptionx){

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

}else{

......

可以看出大概的意思就是从刚才我们放进去的队列events里面取数据放到了eventCache里面，eventCache的定义SynchronizedStackeventCache，当取到数据后返回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的模型了，也就是多路复用的模型，轮询来判断key的状态，当key是可读或者可写时执行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时提前初始化好的，默认情况下核心线程是10，最大线程是200。线程池的配置可以根据我们自己配