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);
}
网友评论