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

书接上文。

当Tomcat的Acceptor监听到有请求到来时，就会结束阻塞，继续进行程序下面的动作。如下面的代码所示：

Java代码



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);这一步了，这个方法的源代码如下所示：

Java代码



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()的源代码如下：

Java代码



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的方法值得研究一下。

Java代码



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()方法还是很好理解的：

Java代码



protected Worker newWorkerThread() {

Worker workerThread = new Worker();

workerThread.start();

return (workerThread);

}

创建一个Worker线程，并且开启线程。

经过上述的过程，就获得了一个Worker(并且已经开始运行了)，那么程序的下一步是调用Worker的assign()方法：

Java代码



synchronized void assign(Socket socket) {

// 等待获取完上一个Socket

while (available) {

try {

wait();

} catch (InterruptedException e) {

}

}

// 保存新的Socket

this.socket = socket;

available = true;

notifyAll();

}

这样新的Worker已经有一个Socket对象去处理了，下面来看一下Worker的线程中做了哪些工作。

Java代码



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已进行请求处理，代码如下：

Java代码



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的参数，如下所示：

Java代码



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到堆栈中以备下次使用：

Java代码



protected void recycleWorkerThread(Worker workerThread) {

synchronized (workers) {

workers.push(workerThread);

curThreadsBusy--;

workers.notify();

}

}

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