Tomcat7线程模型
tomcat 的nio 线程模型也是reactor 模型,由accept 线程负责接受连接请求,把请求转发给其中一个Poller
线程,去注册读事件,Poller 线程就负责该连接的读和写,交给后面的线程池去处理,从读报文,触发后面的servlet请求都由线程池的线程完成。
Accept线程
backlog = 100; 默认是100,也就是tcp的accept 队列为100,默认还是比较少的。
最大连接数
maxConnections = 10000; 如果连接数超过了maxConnections,则等待连接释放,其实这里底层TCP 链接是还可以建立的,只有内核的accept 队列没有满,假如tomcat的链接数达到了10000,accept线程就不从accept的队列取出链接,这样就很容易导致不能建立链接了。
核心代码Run 方法如下:
int errorDelay = 0;
// Loop until we receive a shutdown command
while (running) {
// Loop if endpoint is paused
while (paused && running) {
state = AcceptorState.PAUSED;
try {
Thread.sleep(50);
} catch (InterruptedException e) {
// Ignore
}
}
if (!running) {
break;
}
state = AcceptorState.RUNNING;
try {
//if we have reached max connections, wait
countUpOrAwaitConnection();
SocketChannel socket = null;
try {
// Accept the next incoming connection from the server
// socket
socket = serverSock.accept();
} catch (IOException ioe) {
// We didn't get a socket
countDownConnection();
if (running) {
// 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 (running && !paused) {
// setSocketOptions() will hand the socket off to
//这里把sock 分发到poller 线程
// an appropriate processor if successful
if (!setSocketOptions(socket)) {
closeSocket(socket);
}
} else {
closeSocket(socket);
}
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error(sm.getString("endpoint.accept.fail"), t);
}
}
state = AcceptorState.ENDED;
}
Poller 线程
Poller线程负责轮询注册在对应selector 上连接的读写请求事件。因为Accept接收到链接请求后,回封装成一个event,放到Poller的事件队列,poller 回从里面取出事件获取socket。
Poller 线程个数 pollerThreadCount默认2个
pollerThreadCount = Math.min(2,Runtime.getRuntime().availableProcessors());
Accept 选择poller是Round Robin,所以两个poller线程负责的socket 各占一半
同步读
poller io 线程还有点和reactor 模型不一样的是,poller 线程不负责具体的读http 消息,而是有可读事件时,分配给 SocketProcessor 来处理,SocketProcessor 是一个task,具体由tomcat的工作线程池来执行,所以一个连接上的http 请求数据报的读取和poller 的线程是异步的,正是因为这样,poller 在分配一个读事件给SocketProcessor 后,就取消了可读事件的监听,下面是poller worker线程的processKey 方法,用来分配读写事件。
protected void processKey(SelectionKey sk, NioSocketWrapper attachment) {
try {
if ( close ) {
cancelledKey(sk);
} else if ( sk.isValid() && attachment != null ) {
if (sk.isReadable() || sk.isWritable() ) {
if ( attachment.getSendfileData() != null ) {
processSendfile(sk,attachment, false);
} else {
//先取消读事件,意思是防止读
unreg(sk, attachment, sk.readyOps());
boolean closeSocket = false;
// Read goes before write
if (sk.isReadable()) {
//创建socketprocessor来读http 请求包和业务逻辑的执行
if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
closeSocket = true;
}
}
if (!closeSocket && sk.isWritable()) {
if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
closeSocket = true;
}
}
if (closeSocket) {
cancelledKey(sk);
}
}
}
} else {
//invalid key
cancelledKey(sk);
}
} catch ( CancelledKeyException ckx ) {
cancelledKey(sk);
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error("",t);
}
}
注意上面的unreg方法如下
protected void unreg(SelectionKey sk, NioSocketWrapper attachment, int readyOps) {
//this is a must, so that we don't have multiple threads messing with the socket
reg(sk,attachment,sk.interestOps()& (~readyOps));
}
官方解释是说防止多个线程同时读一个socket,也就是一个请求连接的数据。想象一种场景,如果一个hSocketProcess ttp请求的包只来了一部分,也就是SocketProcess 在等待后面一部分,后面部分来的时候,触发读事件,重新创建一个SocketProcessor,这样会导致两个processor 同时处理一个socket数据,会导致混乱。
何时重新注册读事件
- 1 上次请求处理完成,会重新注册读事件,因为连接是持久keeplivve的
- 2 处理半包的情况,需要重新注册读事件
//状态为LONG时,代表半包的状态,没有读完,需要等待,并重新注册可读事件,
//而且socket 关联的process 不能从connectionsremove掉
if (state == SocketState.LONG) {
// In the middle of processing a request/response. Keep the
// socket associated with the processor. Exact requirements
// depend on type of long poll
//longPoll 如果不是异步请求,会注册读事件
longPoll(wrapper, processor);
if (processor.isAsync()) {
getProtocol().addWaitingProcessor(processor);
}
} else if (state == SocketState.OPEN) {
// In keep-alive but between requests. OK to recycle
// processor. Continue to poll for the next request.
//处理完成的请求,可以remove掉process,因为不知道下次请求什么时候来,
// 同时也需要重新注册读事件
connections.remove(socket);
getLog().info("Tomcat process finish start to release process "+processor.getRequest().toString());
release(processor);
getLog().info("Tomcat release process "+processor.getRequest().toString()+ "start to register read event for next read!!!");
wrapper.registerReadInterest();
}
所以从上面的分析可以得出结论,tomcat nio 模型读不同于netty的reactor 模型,io 读写由io 线程负责,读完了就交给业务线程支持,继续读后面的请求数据。但是tomcat是一个请求读完,处理完业务逻辑,再继续读下一个请求的数据,这对http 这种独占的协议无可厚非,如果想在http协议上实现类似rpc 自定义协议的连接复用时,即发请求可以不用等当前请求返回,就可以继续发,对发送多可以实现少量的连接发送大量的请求,但是由于服务端不能并发的读,必然会导致读缓冲区瞬间满了,不能被读走的请求,由于tcp 滑动窗口因子,也会导致发送方停止下来
工作线程池executor
执行请求的线程池
public void createExecutor() {
internalExecutor = true;
TaskQueue taskqueue = new TaskQueue();
TaskThreadFactory tf = new TaskThreadFactory(getName() + "-exec-", daemon, getThreadPriority());
executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), 60, TimeUnit.SECONDS,taskqueue, tf);
taskqueue.setParent( (ThreadPoolExecutor) executor);
}
- 队列:无界队列 TaskQueue,
- 最小线 程minSpareThreads = 10
- 最大线程 maxThreads = 200
TaskQueue
taskQueue 对 offer方法做了些手脚,就是让exeecutor的核心线程池达到最大值,如果按正常的逻辑,当线程超过CoreSize 时,任务回往offer到TaskQueue 中,而tomcat的TaskQueue 是无界的队列,所以默认的话tomcat都只有core size个线程在跑,这样估计吞吐量不够,所以tomcat的TaskQueue修改了offer方法,如下:
@Override
public boolean offer(Runnable o) {
//we can't do any checks
if (parent==null) return super.offer(o);
//we are maxed out on threads, simply queue the object
if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
//we have idle threads, just add it to the queue
if (parent.getSubmittedCount()<(parent.getPoolSize())) return super.offer(o);
//if we have less threads than maximum force creation of a new thread
//关键点在这里,只要工作线程小于最大值,就返回false,这时线程池会去创建新的线程。
if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
//if we reached here, we need to add it to the queue
return super.offer(o);
}
如果用来tomcat sever.xml 指定的 exector ,即把Executor 启用
<Executor name="tomcatThreadPool" namePrefix="catalina-exec-"
maxThreads="150" minSpareThreads="4" maxQueueSize="1000"/>
则创建的是Tomcat 自己实现的StandardThreadExecutor,该线程池唯一不同的是,可以指定队列容量的大小,默认是Integer.MAX_VALUE,相当于无界l。
可以通过maxQueueSize
属性指定,代码如下:
@Override
protected void startInternal() throws LifecycleException {
taskqueue = new TaskQueue(maxQueueSize);
TaskThreadFactory tf = new TaskThreadFactory(namePrefix,daemon,getThreadPriority());
executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), maxIdleTime, TimeUnit.MILLISECONDS,taskqueue, tf);
executor.setThreadRenewalDelay(threadRenewalDelay);
if (prestartminSpareThreads) {
executor.prestartAllCoreThreads();
}
taskqueue.setParent(executor);
setState(LifecycleState.STARTING);
}
Tomcat 异步处理
Servlet3.0 支持了异步,tomcat7 对异步也要支持,在tomcat的工作线程处理完后,如果时异步的话,不能结束掉当前这个请求,要等待业务线程触发了asyncContext.complete() 方法,执行这个complete时,tomcat 会把该请求对于的socketprocess 获取到,再教给上面说的executor 去执行。所以我们在通过request.startAsynce()时,最好不要用asyncContext.start()方法去执行一些操作,这样的话,这个异步处理还是需要tomcat的线程,来执行,就没有意义了。
Tomcat 异步写
tomcat 的 response flush时,是阻塞的,如果写缓冲区不可用,则会阻塞住flush的线程,如果想要异步flush。则需要给response的outputStream 添加一个writerListener,有了writerListener tomcat就异步写,不会阻塞。但是需要注意的是,必须用tomcat的ServletOutputStream 才支持,默认的servlet api 下的ServletOutputStream是没有该方法的。
public abstract voidsetWriteListener(javax.servlet.WriteListener listener);
// If we know that the request is bad this early, add the
// Connection: close header.
if (keepAlive && statusDropsConnection(statusCode)) {
keepAlive = false;
}
if (!keepAlive) {
// Avoid adding the close header twice
if (!connectionClosePresent) {
headers.addValue(Constants.CONNECTION).setString(
Constants.CLOSE);
}
} else if (!http11 && !getErrorState().isError()) {
headers.addValue(Constants.CONNECTION).setString(Constants.KEEPALIVE);
}