Tomcat请求处理(二) -- 请求处理框架

q3256

书接上文。
当Tomcat的Acceptor监听到有请求到来时，就会结束阻塞，继续进行程序下面的动作。如下面的代码所示：
public void run() {
while (running) {
while (paused) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
}
try {
// 开始监听端口
Socket socket = serverSocketFactory.acceptSocket(serverSocket);
// 初始化Socket
serverSocketFactory.initSocket(socket);
// 处理Socket
if (!processSocket(socket)) {
try {
socket.close();
} catch (IOException e) {
}
}
} catch (IOException x) {
if (running)
log.error(sm.getString("endpoint.accept.fail"), x);
} catch (Throwable t) {
log.error(sm.getString("endpoint.accept.fail"), t);
}
}
}

结束阻塞后，首先是Socket的初始化，由于ServerSocketFactory的实现类DefaultServerSocketFactory并没有扩展initSocket()方法，所以这一步其实是什么都没做的。
接下来程序执行到了processSocket(socket);这一步了，这个方法的源代码如下所示：
protected boolean processSocket(Socket socket) {
try {
if (executor == null) {
// 得到一个Work并为它分配这个Socket
getWorkerThread().assign(socket);
} else {
executor.execute(new SocketProcessor(socket));
}
} catch (Throwable t) {
log.error(sm.getString("endpoint.process.fail"), t);
return false;
}
return true;
}

executor是一个外部的基于线程池的执行器，先不去考虑它，重点看一下getWorkerThread()和Worker#assign()
getWorkerThread()的源代码如下：

protected Worker getWorkerThread() {
// 获取一个Worker
Worker workerThread = createWorkerThread();
while (workerThread == null) {// 如果获取的Worker为Null
try {
synchronized (workers) {
// 等待workers里边Worker的回收，recycleWorkerThread()中会调用notify通知这个线程结束等待的
workers.wait();
}
} catch (InterruptedException e) {
}
// 等待结束，再次获取一个Worker
workerThread = createWorkerThread();
}
return workerThread;
}

这里这个createWorkerThread()也就是获取Worker的方法值得研究一下。
protected Worker createWorkerThread() {
synchronized (workers) {
if (workers.size() > 0) {// 如果堆栈中有剩余的Worker
// 当前在使用的Worker线程计数
curThreadsBusy++;
// 将一个Worker推出堆栈
return workers.pop();
}
// 如果堆栈中没有多余的Worker了，那么创建一个。
if ((maxThreads > 0) && (curThreads < maxThreads)) {
// 如果maxThreads有定义，并且当前的Worker数量小于这个值。
// 在recycleWorkerThread()中会将新创建的Worker放入堆栈的
curThreadsBusy++;
return (newWorkerThread());
} else {
if (maxThreads < 0) {
// maxThreads没有定义，可以有无限个Worker的情况。
curThreadsBusy++;
return (newWorkerThread());
} else {
return (null);
}
}
}
}

而newWorkerThread()方法还是很好理解的：
protected Worker newWorkerThread() {
Worker workerThread = new Worker();
workerThread.start();
return (workerThread);
}

创建一个Worker线程，并且开启线程。
经过上述的过程，就获得了一个Worker(并且已经开始运行了)，那么程序的下一步是调用Worker的assign()方法：
synchronized void assign(Socket socket) {
// 等待获取完上一个Socket
while (available) {
try {
wait();
} catch (InterruptedException e) {
}
}
// 保存新的Socket
this.socket = socket;
available = true;
notifyAll();
}

这样新的Worker已经有一个Socket对象去处理了，下面来看一下Worker的线程中做了哪些工作。
public void run() {
// 知道收到停止信号，否则一直循环
while (running) {
// 等待为这个线程设置一个Socket对象
Socket socket = await();
if (socket == null)
continue;
// 处理请求
if (!setSocketOptions(socket) || !handler.process(socket)) {
// 关闭Socket
try {
socket.close();
} catch (IOException e) {
}
}
// 回收Worker
socket = null;
recycleWorkerThread(this);
}
}

其中await()用于等待有Socket对象赋给当前的Worker已进行请求处理，代码如下：
private synchronized Socket await() {
// 等待获取Socket
while (!available) {
try {
wait();
} catch (InterruptedException e) {
}
}
// Socket已经收到
Socket socket = this.socket;
available = false;
// 这个notifyAll()是通知assign()中的等待Socket的处理已经开始了，
// 可以为这个Worker赋新的值了。
notifyAll();
return (socket);
}

而setSocketOptions()用于设定一些Socket的参数，如下所示：
protected boolean setSocketOptions(Socket socket) {
int step = 1;
try {
if (soLinger >= 0) {
socket.setSoLinger(true, soLinger);
}
if (tcpNoDelay) {
socket.setTcpNoDelay(tcpNoDelay);
}
if (soTimeout > 0) {
socket.setSoTimeout(soTimeout);
}
step = 2;
serverSocketFactory.handshake(socket);
} catch (Throwable t) {
if (log.isDebugEnabled()) {
if (step == 2) {
log.debug(sm.getString("endpoint.err.handshake"), t);
} else {
log.debug(sm.getString("endpoint.err.unexpected"), t);
}
}
return false;
}
return true;
}

handler.process(socket)封装了处理请求的全过程，下次再详细了解
而最后的recycleWorkerThread()用于回收Worker到堆栈中以备下次使用：
protected void recycleWorkerThread(Worker workerThread) {
synchronized (workers) {
workers.push(workerThread);
curThreadsBusy--;
workers.notify();
}
}

好了，请求处理的整个流程大致就是这样的，下次再详细了解请求的真实处理部分，就是handler.process(socket)的内容。