tomcat 源码分析之--请求接受以及缓冲池

farmer111

　　tomcat  从接受请求到处理的流程大概如下：
　　首先配置文件中的connector都对应一个Connector类，这个类根据配置信息确定对应的
　　ProtocalHandler(处理connector配置对应的协议).

public void start() throws LifecycleException {
if( !initialized )
initialize();
// Validate and update our current state
if (started ) {
if(log.isInfoEnabled())
log.info(sm.getString("coyoteConnector.alreadyStarted"));
return;
}
lifecycle.fireLifecycleEvent(START_EVENT, null);
started = true;
// We can't register earlier - the JMX registration of this happens
// in Server.start callback
if ( this.oname != null ) {
// We are registred - register the adapter as well.
try {
Registry.getRegistry(null, null).registerComponent
(protocolHandler, createObjectName(this.domain,"ProtocolHandler"), null);
} catch (Exception ex) {
log.error(sm.getString
("coyoteConnector.protocolRegistrationFailed"), ex);
}
} else {
if(log.isInfoEnabled())
log.info(sm.getString
("coyoteConnector.cannotRegisterProtocol"));
}
try {
protocolHandler.start();
} catch (Exception e) {
String errPrefix = "";
if(this.service != null) {
errPrefix += "service.getName(): \"" + this.service.getName() + "\"; ";
}
throw new LifecycleException
(errPrefix + " " + sm.getString
("coyoteConnector.protocolHandlerStartFailed", e));
}
if( this.domain != null ) {
mapperListener.setDomain( domain );
//mapperListener.setEngine( service.getContainer().getName() );
mapperListener.init();
try {
ObjectName mapperOname = createObjectName(this.domain,"Mapper");
if (log.isDebugEnabled())
log.debug(sm.getString(
"coyoteConnector.MapperRegistration", mapperOname));
Registry.getRegistry(null, null).registerComponent
(mapper, mapperOname, "Mapper");
} catch (Exception ex) {
log.error(sm.getString
("coyoteConnector.protocolRegistrationFailed"), ex);
}
}
}
　　抛开其他代码，注意上面方法里面的这一句：protocolHandler.start();
   比如http1.1协议对应的ProtocalHandler是：Http11Protocol。
　　而这个ProtocalHanler依靠一个对应的EndPoint对象来处理请求，http11protocal的start方法代码
　　如下：

public void start() throws Exception {
if (this.domain != null) {
try {
tpOname = new ObjectName
(domain + ":" + "type=ThreadPool,name=" + getName());
Registry.getRegistry(null, null)
.registerComponent(endpoint, tpOname, null );
} catch (Exception e) {
log.error("Can't register endpoint");
}
rgOname=new ObjectName
(domain + ":type=GlobalRequestProcessor,name=" + getName());
Registry.getRegistry(null, null).registerComponent
( cHandler.global, rgOname, null );
}
try {
endpoint.start();
} catch (Exception ex) {
log.error(sm.getString("http11protocol.endpoint.starterror"), ex);
throw ex;
}
if (log.isInfoEnabled())
log.info(sm.getString("http11protocol.start", getName()));
}
　　注意其中的代码：        endpoint.start();
　　Endpoint是对应的protocalhandler的工具类，每个endpoinst都监听一个对应的ServerSocket并且处理请求，start方法代码如下：

public void start()
throws Exception {
// Initialize socket if not done before
if (!initialized) {
init();
}
if (!running) {
running = true;
paused = false;
// Create worker collection
if (executor == null) {
workers = new WorkerStack(maxThreads);
}
// Start acceptor threads
for (int i = 0; i < acceptorThreadCount; i++) {
Thread acceptorThread = new Thread(new Acceptor(), getName() + "-Acceptor-" + i);
acceptorThread.setPriority(threadPriority);
acceptorThread.setDaemon(daemon);
acceptorThread.start();
}
}
}
　　很容易看出，其实tomcat的请求监听也不是单线程的，有多个线程在进行监听。那么每个监听线程在监听到socket请求之后做什么工作呢？就是处理socket.这是通过调用Endpoint的processSocket方法实现的：

  protected boolean processSocket(Socket socket) {
try {
if (executor == null) {
getWorkerThread().assign(socket);
} else {
executor.execute(new SocketProcessor(socket));
}
} catch (Throwable t) {
// This means we got an OOM or similar creating a thread, or that
// the pool and its queue are full
log.error(sm.getString("endpoint.process.fail"), t);
return false;
}
return true;
}
　　如果在  配置文件里面定义了一个executor,那么就使用此executor.否则使用默认的Worker缓冲池。下面就来分析一下worker缓冲池的实现：
　　worker是存放在WorkerStack里面的，WorkerStack维护了一个worker数组。当endpoint里面需要worker线程的时候，会调用如下方法：

protected Worker createWorkerThread() {
synchronized (workers) {
if (workers.size() > 0) {
curThreadsBusy++;
return workers.pop();
}
if ((maxThreads > 0) && (curThreads < maxThreads)) {
curThreadsBusy++;
if (curThreadsBusy == maxThreads) {
log.info(sm.getString("endpoint.info.maxThreads",
Integer.toString(maxThreads), address,
Integer.toString(port)));
}
return (newWorkerThread());
} else {
if (maxThreads < 0) {
curThreadsBusy++;
return (newWorkerThread());
} else {
return (null);
}
}
}
}
　　这个方法不多说了，一看就差不多明白大概原理。但是可能还有一点疑问：新增的worker如何放到WorkerStack里面呢？ 这个就要看一下Worker的实现了：worker运行最后有一句代码：recycleWorkerThread。

    protected void recycleWorkerThread(Worker workerThread) {
synchronized (workers) {
workers.push(workerThread);
curThreadsBusy--;
workers.notify();
}
}
　　至此，一个缓冲池轻松实现了。