摘要：本文主要介绍了tomcat内部处理HTTP请求的Connector部分
在上一篇文章中已经介绍了tomcat在能处理HTTP请求之前所做的准备，今天这篇文章就开始正式开始介绍tomcat处理HTTP请求。在上篇文章说到下面代码：
//代码清单1 在JIOEndpoint中
@Override
protected AbstractEndpoint.Acceptor createAcceptor() {
return new Acceptor();
}
/**
* The background thread that listens for incoming TCP/IP connections and
* hands them off to an appropriate processor.
JIOEndpoint的内部类
*/
protected class Acceptor extends AbstractEndpoint.Acceptor {
@Override
public void run() {
int errorDelay = 0;
// Loop until we receive a shutdown command
while (running) {
//各种条件判断略
state = AcceptorState.RUNNING;
try {
//if we have reached max connections, wait
countUpOrAwaitConnection();
Socket socket = null;
try {
// Accept the next incoming connection from the server
// socket
//1111111111
socket = serverSocketFactory.acceptSocket(serverSocket);
} catch (IOException ioe) {
//异常处理 略
}
// Successful accept, reset the error delay
errorDelay = 0;
// Configure the socket
//22222222 设置socket的各种属性
if (running && !paused && setSocketOptions(socket)) {
// Hand this socket off to an appropriate processor
//33333333
if (!processSocket(socket)) {
countDownConnection();
// Close socket right away
closeSocket(socket);
}
} else {
countDownConnection();
// Close socket right away
closeSocket(socket);
}
} catch (IOException x) {
//异常处理 略
} catch (NullPointerException npe) {
//异常处理 略
} catch (Throwable t) {
//异常处理 略
}
}
state = AcceptorState.ENDED;
}
}
上篇文章我们知道在标注1的地方会阻塞，一直到有请求过来才会继续往下走。在标注2的地方setSocketOptions(socket)设置socket的各种属性：
//代码清单2
protected boolean setSocketOptions(Socket socket) {
try {
// 1: Set socket options: timeout, linger, etc
// 设置socket的参数
socketProperties.setProperties(socket);
} catch (SocketException s) {
//error here is common if the client has reset the connection
if (log.isDebugEnabled()) {
log.debug(sm.getString("endpoint.err.unexpected"), s);
}
// Close the socket
return false;
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error(sm.getString("endpoint.err.unexpected"), t);
// Close the socket
return false;
}
return true;
}
public void setProperties(Socket socket) throws SocketException{
if (rxBufSize != null)
socket.setReceiveBufferSize(rxBufSize.intValue());
if (txBufSize != null)
socket.setSendBufferSize(txBufSize.intValue());
if (ooBInline !=null)
socket.setOOBInline(ooBInline.booleanValue());
if (soKeepAlive != null)
socket.setKeepAlive(soKeepAlive.booleanValue());
if (performanceConnectionTime != null && performanceLatency != null &&
performanceBandwidth != null)
socket.setPerformancePreferences(
performanceConnectionTime.intValue(),
performanceLatency.intValue(),
performanceBandwidth.intValue());
if (soReuseAddress != null)
socket.setReuseAddress(soReuseAddress.booleanValue());
if (soLingerOn != null && soLingerTime != null)
socket.setSoLinger(soLingerOn.booleanValue(),
soLingerTime.intValue());
if (soTimeout != null && soTimeout.intValue() >= 0)
socket.setSoTimeout(soTimeout.intValue());
if (tcpNoDelay != null)
socket.setTcpNoDelay(tcpNoDelay.booleanValue());
}
继续看processSocket(socket):
//代码清单3
protected boolean processSocket(Socket socket) {
// Process the request from this socket
try {
//将socket包装成了 socketWrapper类
SocketWrapper<Socket> wrapper = new SocketWrapper<Socket>(socket);
wrapper.setKeepAliveLeft(getMaxKeepAliveRequests());
wrapper.setSecure(isSSLEnabled());
// During shutdown, executor may be null - avoid NPE
if (!running) {
return false;
}
//获取线程池执行 `SocketProcessor` 任务
getExecutor().execute(new SocketProcessor(wrapper));
} catch (RejectedExecutionException x) {
//异常处理 略
} catch (Throwable t) {
//异常处理 略
}
return true;
}
可以看到先把socket包装成socketWrapper类，紧接着将socketwrapper对象作为构造函数的参数传递给SocketProcessor类并且提交给线程池执行，可见SocketProcessor类是个Runnable，查看SocketProcessor类的run方法。
//代码清单4
@Override
public void run() {
boolean launch = false;
synchronized (socket) {
try {
SocketState state = SocketState.OPEN;
try {
// SSL handshake
serverSocketFactory.handshake(socket.getSocket());
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
if (log.isDebugEnabled()) {
log.debug(sm.getString("endpoint.err.handshake"), t);
}
// Tell to close the socket
state = SocketState.CLOSED;
}
if ((state != SocketState.CLOSED)) {
if (status == null) {
state = handler.process(socket, SocketStatus.OPEN_READ);
} else {
//1111111111 使用handler 处理socket
state = handler.process(socket,status);
}
}
if (state == SocketState.CLOSED) {
// Close socket
if (log.isTraceEnabled()) {
log.trace("Closing socket:"+socket);
}
countDownConnection();
try {
socket.getSocket().close();
} catch (IOException e) {
// Ignore
}
} else if (state == SocketState.OPEN ||
state == SocketState.UPGRADING ||
state == SocketState.UPGRADING_TOMCAT ||
state == SocketState.UPGRADED){
socket.setKeptAlive(true);
socket.access();
launch = true;
} else if (state == SocketState.LONG) {
socket.access();
waitingRequests.add(socket);
}
} finally {
if (launch) {
try {
getExecutor().execute(new SocketProcessor(socket, SocketStatus.OPEN_READ));
} catch (RejectedExecutionException x) {
log.warn("Socket reprocessing request was rejected for:"+socket,x);
try {
//unable to handle connection at this time
handler.process(socket, SocketStatus.DISCONNECT);
} finally {
countDownConnection();
}
} catch (NullPointerException npe) {
if (running) {
log.error(sm.getString("endpoint.launch.fail"),
npe);
}
}
}
}
}
socket = null;
// Finish up this request
}
}
代码虽然长，但是都是对很多情况做了很多判断处理。主要执行的是在标注1的地方调用handler的process(socket)处理socket。而因为SocketProcessor是JIoEndpoint的内部类，所以handler变量是JIoEndpoint的成员变量。在上一篇文章中，我们在说JIoEndPoint是在Http11Protocol类的构造函数中初始化的，而handler变量也是在这里设置的：
//代码清单5
public Http11Protocol() {
endpoint = new JIoEndpoint();
cHandler = new Http11ConnectionHandler(this);
((JIoEndpoint) endpoint).setHandler(cHandler);
setSoLinger(Constants.DEFAULT_CONNECTION_LINGER);
setSoTimeout(Constants.DEFAULT_CONNECTION_TIMEOUT);
setTcpNoDelay(Constants.DEFAULT_TCP_NO_DELAY);
}
所以handler变量是Http11ConnectionHandler的实例。我们继续查看其process(socket)方法，最终我们在其父类AbstractConnectionHandler中找到了实现：（方法有删减）
//代码清单6
public SocketState process(SocketWrapper<S> wrapper,SocketStatus status) {
//基本判断
if (wrapper == null) {
// Nothing to do. Socket has been closed.
return SocketState.CLOSED;
}
//获取包装类的socket
S socket = wrapper.getSocket();
if (socket == null) {
// Nothing to do. Socket has been closed.
return SocketState.CLOSED;
}
//从Socket的连接缓存connections（用于缓存长连接的Socket）中获取Socket对应的Http11Processor
Processor<S> processor = connections.get(socket);
if (status == SocketStatus.DISCONNECT && processor == null) {
// Nothing to do. Endpoint requested a close and there is no
// longer a processor associated with this socket.
return SocketState.CLOSED;
}
wrapper.setAsync(false);
ContainerThreadMarker.markAsContainerThread();
try {
//如果连接缓存connections中不存在Socket对应的Http11Processor，则从可以循环使用的recycledProcessors（类型为ConcurrentLinkedQueue）中获取
if (processor == null) {
processor = recycledProcessors.poll();
}
//如果recycledProcessors中也没有可以使用的Http11Processor，则调用createProcessor方法创建ttp11Processor
if (processor == null) {
//1111111
processor = createProcessor();
}
//如果Connector 标签设置了 SSLEnabled=true 需要给Http11Processor 设置SSL相关的属性
initSsl(wrapper, processor);
SocketState state = SocketState.CLOSED;
do {
if (status == SocketStatus.CLOSE_NOW) {
processor.errorDispatch();
state = SocketState.CLOSED;
} else if (status == SocketStatus.DISCONNECT &&
!processor.isComet()) {
// Do nothing here, just wait for it to get recycled
// Don't do this for Comet we need to generate an end
// event (see BZ 54022)
} else if (processor.isAsync()) {
//222222222
//如果 processor 是异步的 调用 asyncDispatch 方法处理
state = processor.asyncDispatch(status);
} else if (state == SocketState.ASYNC_END) {
state = processor.asyncDispatch(status);
// release() won't get called so in case this request
// takes a long time to process remove the socket from
// the waiting requests now else the async timeout will
// fire
getProtocol().endpoint.removeWaitingRequest(wrapper);
if (state == SocketState.OPEN) {
// There may be pipe-lined data to read. If the data
// isn't processed now, execution will exit this
// loop and call release() which will recycle the
// processor (and input buffer) deleting any
// pipe-lined data. To avoid this, process it now.
state = processor.process(wrapper);
}
} else if (processor.isComet()) {
state = processor.event(status);
} else if (processor.getUpgradeInbound() != null) {
state = processor.upgradeDispatch();
} else if (processor.isUpgrade()) {
state = processor.upgradeDispatch(status);
} else {
// 33333333
state = processor.process(wrapper);
}
if (state != SocketState.CLOSED && processor.isAsync()) {
state = processor.asyncPostProcess();
}
if (state == SocketState.UPGRADING) {
// Get the HTTP upgrade handler
HttpUpgradeHandler httpUpgradeHandler =
processor.getHttpUpgradeHandler();
// Release the Http11 processor to be re-used
release(wrapper, processor, false, false);
// Create the upgrade processor
processor = createUpgradeProcessor(
wrapper, httpUpgradeHandler);
// Mark the connection as upgraded
wrapper.setUpgraded(true);
// Associate with the processor with the connection
connections.put(socket, processor);
// Initialise the upgrade handler (which may trigger
// some IO using the new protocol which is why the lines
// above are necessary)
// This cast should be safe. If it fails the error
// handling for the surrounding try/catch will deal with
// it.
httpUpgradeHandler.init((WebConnection) processor);
} else if (state == SocketState.UPGRADING_TOMCAT) {
// Get the UpgradeInbound handler
org.apache.coyote.http11.upgrade.UpgradeInbound inbound =
processor.getUpgradeInbound();
// Release the Http11 processor to be re-used
release(wrapper, processor, false, false);
// Create the light-weight upgrade processor
processor = createUpgradeProcessor(wrapper, inbound);
inbound.onUpgradeComplete();
}
if (getLog().isDebugEnabled()) {
getLog().debug("Socket: [" + wrapper +
"], Status in: [" + status +
"], State out: [" + state + "]");
}
} while (state == SocketState.ASYNC_END ||
state == SocketState.UPGRADING ||
state == SocketState.UPGRADING_TOMCAT);
//一直到最后都是 socket的后续处理
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
connections.put(socket, processor);
longPoll(wrapper, processor);
} else if (state == SocketState.OPEN) {
// In keep-alive but between requests. OK to recycle
// processor. Continue to poll for the next request.
connections.remove(socket);
release(wrapper, processor, false, true);
} else if (state == SocketState.SENDFILE) {
// Sendfile in progress. If it fails, the socket will be
// closed. If it works, the socket will be re-added to the
// poller
connections.remove(socket);
release(wrapper, processor, false, false);
} else if (state == SocketState.UPGRADED) {
// Need to keep the connection associated with the processor
connections.put(socket, processor);
// Don't add sockets back to the poller if this was a
// non-blocking write otherwise the poller may trigger
// multiple read events which may lead to thread starvation
// in the connector. The write() method will add this socket
// to the poller if necessary.
if (status != SocketStatus.OPEN_WRITE) {
longPoll(wrapper, processor);
}
} else {
// Connection closed. OK to recycle the processor. Upgrade
// processors are not recycled.
connections.remove(socket);
if (processor.isUpgrade()) {
processor.getHttpUpgradeHandler().destroy();
} else if (processor instanceof org.apache.coyote.http11.upgrade.UpgradeProcessor) {
// NO-OP
} else {
release(wrapper, processor, true, false);
}
}
return state;
} catch(java.net.SocketException e) {
// SocketExceptions are normal
getLog().debug(sm.getString(
"abstractConnectionHandler.socketexception.debug"), e);
} catch (java.io.IOException e) {
// IOExceptions are normal
getLog().debug(sm.getString(
"abstractConnectionHandler.ioexception.debug"), e);
}
// Future developers: if you discover any other
// rare-but-nonfatal exceptions, catch them here, and log as
// above.
catch (Throwable e) {
ExceptionUtils.handleThrowable(e);
// any other exception or error is odd. Here we log it
// with "ERROR" level, so it will show up even on
// less-than-verbose logs.
getLog().error(
sm.getString("abstractConnectionHandler.error"), e);
}
// Make sure socket/processor is removed from the list of current
// connections
connections.remove(socket);
// Don't try to add upgrade processors back into the pool
if (!(processor instanceof org.apache.coyote.http11.upgrade.UpgradeProcessor)
&& !processor.isUpgrade()) {
release(wrapper, processor, true, false);
}
return SocketState.CLOSED;
}
源码虽然长了点，但是主要做的事情就在标注2和标注2的地方，在标注2的地方主要针对的是异步的处理，标注3的地方针对的是同步的处理，我们以同步为例继续查看其源码之前，我们先看下标注1的地方如果没有找到合适的processor处理器，我们需要新建一个processor,查看下createProcessor()方法的源码：
//代码清单7
@Override
protected Http11Processor createProcessor() {
Http11Processor processor = new Http11Processor(
proto.getMaxHttpHeaderSize(), (JIoEndpoint)proto.endpoint,
proto.getMaxTrailerSize(), proto.getAllowedTrailerHeadersAsSet(),
proto.getMaxExtensionSize(), proto.getMaxSwallowSize());
//1111111111
processor.setAdapter(proto.adapter);
processor.setMaxKeepAliveRequests(proto.getMaxKeepAliveRequests());
processor.setKeepAliveTimeout(proto.getKeepAliveTimeout());
processor.setConnectionUploadTimeout(
proto.getConnectionUploadTimeout());
processor.setDisableUploadTimeout(proto.getDisableUploadTimeout());
processor.setCompressionMinSize(proto.getCompressionMinSize());
processor.setCompression(proto.getCompression());
processor.setNoCompressionUserAgents(proto.getNoCompressionUserAgents());
processor.setCompressableMimeTypes(proto.getCompressableMimeTypes());
processor.setRestrictedUserAgents(proto.getRestrictedUserAgents());
processor.setSocketBuffer(proto.getSocketBuffer());
processor.setMaxSavePostSize(proto.getMaxSavePostSize());
processor.setServer(proto.getServer());
processor.setDisableKeepAlivePercentage(
proto.getDisableKeepAlivePercentage());
register(processor);
return processor;
}
这个方法就是新建个Http11Processor对象然后为这个对象设置许多属性，这里我们主要看标注1的地方，为processor对象设置adapter对象，指向的是proto变量的adapter对象，我们查看下声明：
//代码清单8
protected Http11Protocol proto;
指向的是Http11Protocol类，而createProcessor()是Http11ConnectionHandler类的方法，而类Http11ConnectionHandler初始化的时候是在Http11Protocol的构造函数中：
//代码清单9
public Http11Protocol() {
endpoint = new JIoEndpoint();
cHandler = new Http11ConnectionHandler(this);
((JIoEndpoint) endpoint).setHandler(cHandler);
setSoLinger(Constants.DEFAULT_CONNECTION_LINGER);
setSoTimeout(Constants.DEFAULT_CONNECTION_TIMEOUT);
setTcpNoDelay(Constants.DEFAULT_TCP_NO_DELAY);
}
所以其实指向的就是一开始初始化的Http11Protocol对象。而我们在上一篇文章中讲到Connector的init()方法的时候，看到其中有这样的几行代码：
//代码清单10
// Initialize adapter
adapter = new CoyoteAdapter(this);
protocolHandler.setAdapter(adapter);
这里protocolHandler指向的就是Http11Protocol实例，所以就是在Connector的init()方法里设置了adapter对象，而adapter对象指向的是CoyoteAdapter的实例，所以在创建Http11Processor对象的时候，设置他的adapter属性指向的也是这个CoyoteAdapter实例。那么createProcessor()方法就看完了，我们继续查看上面标注3的processor.process(wrapper)方法(同步socket处理)，最终我们在AbstractHttp11Processor类中找到了process()方法：
//代码清单11
@Override
public SocketState process(SocketWrapper<S> socketWrapper)
throws IOException {
RequestInfo rp = request.getRequestProcessor();
rp.setStage(org.apache.coyote.Constants.STAGE_PARSE);
// Setting up the I/O
setSocketWrapper(socketWrapper);
getInputBuffer().init(socketWrapper, endpoint);//初始化 输入流
getOutputBuffer().init(socketWrapper, endpoint);//初始化 输出流
// Flags
keepAlive = true;
comet = false;
openSocket = false;
sendfileInProgress = false;
readComplete = true;
if (endpoint.getUsePolling()) {
keptAlive = false;
} else {
keptAlive = socketWrapper.isKeptAlive();
}
if (disableKeepAlive()) {
socketWrapper.setKeepAliveLeft(0);
}
while (!getErrorState().isError() && keepAlive && !comet && !isAsync() &&
upgradeInbound == null &&
httpUpgradeHandler == null && !endpoint.isPaused()) {
// Parsing the request header
try {
setRequestLineReadTimeout();
// 解析请求行
if (!getInputBuffer().parseRequestLine(keptAlive)) {
if (handleIncompleteRequestLineRead()) {
break;
}
}
if (endpoint.isPaused()) {
// 503 - Service unavailable
response.setStatus(503);
setErrorState(ErrorState.CLOSE_CLEAN, null);
} else {
keptAlive = true;
// Set this every time in case limit has been changed via JMX
request.getMimeHeaders().setLimit(endpoint.getMaxHeaderCount());
// Currently only NIO will ever return false here
// 解析请求头
if (!getInputBuffer().parseHeaders()) {
// We've read part of the request, don't recycle it
// instead associate it with the socket
openSocket = true;
readComplete = false;
break;
}
if (!disableUploadTimeout) {
setSocketTimeout(connectionUploadTimeout);
}
}
} catch (IOException e) {
//异常处理 略
} catch (Throwable t) {
//异常处理 略
}
if (!getErrorState().isError()) {
// Setting up filters, and parse some request headers
rp.setStage(org.apache.coyote.Constants.STAGE_PREPARE);
try {
//设置过滤器 协议，方法，请求头等
prepareRequest();
} catch (Throwable t) {
//异常处理 略
}
}
if (maxKeepAliveRequests == 1) {
keepAlive = false;
} else if (maxKeepAliveRequests > 0 &&
socketWrapper.decrementKeepAlive() <= 0) {
keepAlive = false;
}
// Process the request in the adapter
if (!getErrorState().isError()) {
try {
rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
//111111111111 将request response 交给adapter处理
adapter.service(request, response);
// Handle when the response was committed before a serious
// error occurred. Throwing a ServletException should both
// set the status to 500 and set the errorException.
// If we fail here, then the response is likely already
// committed, so we can't try and set headers.
if(keepAlive && !getErrorState().isError() && (
response.getErrorException() != null ||
(!isAsync() &&
statusDropsConnection(response.getStatus())))) {
setErrorState(ErrorState.CLOSE_CLEAN, null);
}
setCometTimeouts(socketWrapper);
} catch (InterruptedIOException e) {
//异常处理 略
} catch (HeadersTooLargeException e) {
//异常处理 略
} catch (Throwable t) {
//异常处理 略
}
}
// Finish the handling of the request
rp.setStage(org.apache.coyote.Constants.STAGE_ENDINPUT);
if (!isAsync() && !comet) {
if (getErrorState().isError()) {
// If we know we are closing the connection, don't drain
// input. This way uploading a 100GB file doesn't tie up the
// thread if the servlet has rejected it.
getInputBuffer().setSwallowInput(false);
} else {
// Need to check this again here in case the response was
// committed before the error that requires the connection
// to be closed occurred.
checkExpectationAndResponseStatus();
}
endRequest();
}
rp.setStage(org.apache.coyote.Constants.STAGE_ENDOUTPUT);
// If there was an error, make sure the request is counted as
// and error, and update the statistics counter
if (getErrorState().isError()) {
response.setStatus(500);
}
request.updateCounters();
if (!isAsync() && !comet || getErrorState().isError()) {
if (getErrorState().isIoAllowed()) {
getInputBuffer().nextRequest();
getOutputBuffer().nextRequest();
}
}
if (!disableUploadTimeout) {
if(endpoint.getSoTimeout() > 0) {
setSocketTimeout(endpoint.getSoTimeout());
} else {
setSocketTimeout(0);
}
}
rp.setStage(org.apache.coyote.Constants.STAGE_KEEPALIVE);
if (breakKeepAliveLoop(socketWrapper)) {
break;
}
}
rp.setStage(org.apache.coyote.Constants.STAGE_ENDED);
if (getErrorState().isError() || endpoint.isPaused()) {
return SocketState.CLOSED;
} else if (isAsync() || comet) {
return SocketState.LONG;
} else if (isUpgrade()) {
return SocketState.UPGRADING;
} else if (getUpgradeInbound() != null) {
return SocketState.UPGRADING_TOMCAT;
} else {
if (sendfileInProgress) {
return SocketState.SENDFILE;
} else {
if (openSocket) {
if (readComplete) {
return SocketState.OPEN;
} else {
return SocketState.LONG;
}
} else {
return SocketState.CLOSED;
}
}
}
}
代码比较重要的地方都加上了注释，其中关于请求行请求头的解析因为篇幅问题就不展开叙述了，需要的可以自行查看，代码不算太复杂。在代码中我们能看到Http11Processor最终把socket封装成为request和response，然后把request和response交给了adapter来处理，我们继续查看adapter的处理方法service()：
//代码清单12
/**
* Service method.
*/
@Override
public void service(org.apache.coyote.Request req,org.apache.coyote.Response res) throws Exception {
Request request = (Request) req.getNote(ADAPTER_NOTES);
Response response = (Response) res.getNote(ADAPTER_NOTES);
if (request == null) {
// Create objects
request = connector.createRequest();
request.setCoyoteRequest(req);
response = connector.createResponse();
response.setCoyoteResponse(res);
// Link objects
request.setResponse(response);
response.setRequest(request);
// Set as notes
req.setNote(ADAPTER_NOTES, request);
res.setNote(ADAPTER_NOTES, response);
// Set query string encoding
req.getParameters().setQueryStringEncoding
(connector.getURIEncoding());
}
if (connector.getXpoweredBy()) {
response.addHeader("X-Powered-By", POWERED_BY);
}
boolean comet = false;
boolean async = false;
boolean postParseSuccess = false;
try {
// Parse and set Catalina and configuration specific
// request parameters
req.getRequestProcessor().setWorkerThreadName(Thread.currentThread().getName());
//1111111111 解析request 包含了对request的相当多的操作
postParseSuccess = postParseRequest(req, request, res, response);
if (postParseSuccess) {
//check valves if we support async
request.setAsyncSupported(connector.getService().getContainer().getPipeline().isAsyncSupported());
// Calling the container
//22222222222 调用StandardEngine的pipeline 处理request和response
connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);
if (request.isComet()) {
if (!response.isClosed() && !response.isError()) {
if (request.getAvailable() || (request.getContentLength() > 0 && (!request.isParametersParsed()))) {
// Invoke a read event right away if there are available bytes
if (event(req, res, SocketStatus.OPEN_READ)) {
comet = true;
res.action(ActionCode.COMET_BEGIN, null);
} else {
return;
}
} else {
comet = true;
res.action(ActionCode.COMET_BEGIN, null);
}
} else {
// Clear the filter chain, as otherwise it will not be reset elsewhere
// since this is a Comet request
request.setFilterChain(null);
}
}
}
if (request.isAsync()) {
async = true;
} else if (!comet) {
try {
request.finishRequest();
response.finishResponse();
} finally {
if (postParseSuccess) {
// Log only if processing was invoked.
// If postParseRequest() failed, it has already logged it.
// If context is null this was the start of a comet request
// that failed and has already been logged.
((Context) request.getMappingData().context).logAccess(
request, response,
System.currentTimeMillis() - req.getStartTime(),
false);
}
req.action(ActionCode.POST_REQUEST , null);
}
}
} catch (IOException e) {
// Ignore
} finally {
req.getRequestProcessor().setWorkerThreadName(null);
AtomicBoolean error = new AtomicBoolean(false);
res.action(ActionCode.IS_ERROR, error);
// Recycle the wrapper request and response
if (!comet && !async || error.get()) {
request.recycle();
response.recycle();
} else {
// Clear converters so that the minimum amount of memory
// is used by this processor
request.clearEncoders();
response.clearEncoders();
}
}
}
方法很长，但是需要关注的地方就标注1的地方和标注的2的地方。在标注1的地方调用了私有方法postParseRequest()来将org.apache.coyote.Request中的内容解析到org.apache.catalina.connector.Request。在标注2的地方connector.getService().getContainer()返回的是StandardEngine容器，因为StandardEngine只有基本阀所以getPipeline().getFirst()返回的应该是StandardEngineValve,所以标注2的地方最终是调用了StandardEngineValve的invoke()方法，传递参数是解析完成的org.apache.catalina.connector.Request和org.apache.catalina.connector.Response(关于容器和pipeline的疑问可以查看之前相关文章)。
我们来查看下私有方法postParseRequest()：
//代码清单13
/**
* Parse additional request parameters.
*/
protected boolean postParseRequest(org.apache.coyote.Request req,
Request request,
org.apache.coyote.Response res,
Response response)
throws Exception {
// XXX the processor may have set a correct scheme and port prior to this point,
// in ajp13 protocols dont make sense to get the port from the connector...
// otherwise, use connector configuration
if (! req.scheme().isNull()) {
// use processor specified scheme to determine secure state
request.setSecure(req.scheme().equals("https"));
} else {
// use connector scheme and secure configuration, (defaults to
// "http" and false respectively)
req.scheme().setString(connector.getScheme());
request.setSecure(connector.getSecure());
}
// FIXME: the code below doesnt belongs to here,
// this is only have sense
// in Http11, not in ajp13..
// At this point the Host header has been processed.
// Override if the proxyPort/proxyHost are set
String proxyName = connector.getProxyName();
int proxyPort = connector.getProxyPort();
if (proxyPort != 0) {
req.setServerPort(proxyPort);
}
if (proxyName != null) {
req.serverName().setString(proxyName);
}
// Copy the raw URI to the decodedURI
MessageBytes decodedURI = req.decodedURI();
decodedURI.duplicate(req.requestURI());
// Parse the path parameters. This will:
// - strip out the path parameters
// - convert the decodedURI to bytes
//解析请求中的参数
parsePathParameters(req, request);
// URI decoding
// %xx decoding of the URL
try {
//URI decoding的转换
req.getURLDecoder().convert(decodedURI, false);
} catch (IOException ioe) {
res.setStatus(400);
res.setMessage("Invalid URI: " + ioe.getMessage());
connector.getService().getContainer().logAccess(
request, response, 0, true);
return false;
}
// Normalization
//处理请求中的特殊字符 比如// ./ ../ 等
if (!normalize(req.decodedURI())) {
res.setStatus(400);
res.setMessage("Invalid URI");
connector.getService().getContainer().logAccess(
request, response, 0, true);
return false;
}
// Character decoding
//将字节转换为字符
convertURI(decodedURI, request);
// Check that the URI is still normalized
// 判断URI中是否包含特殊字符
if (!checkNormalize(req.decodedURI())) {
res.setStatus(400);
res.setMessage("Invalid URI character encoding");
connector.getService().getContainer().logAccess(
request, response, 0, true);
return false;
}
// Request mapping.
MessageBytes serverName;
if (connector.getUseIPVHosts()) {
serverName = req.localName();
if (serverName.isNull()) {
// well, they did ask for it
res.action(ActionCode.REQ_LOCAL_NAME_ATTRIBUTE, null);
}
} else {
serverName = req.serverName();
}
if (request.isAsyncStarted()) {
//TODO SERVLET3 - async
//reset mapping data, should prolly be done elsewhere
request.getMappingData().recycle();
}
// Version for the second mapping loop and
// Context that we expect to get for that version
String version = null;
Context versionContext = null;
boolean mapRequired = true;
while (mapRequired) {
// This will map the the latest version by default
//匹配请求对应的Host Context Wrapper
// 1111111111111
connector.getMapper().map(serverName, decodedURI, version,
request.getMappingData());
request.setContext((Context) request.getMappingData().context);
request.setWrapper((Wrapper) request.getMappingData().wrapper);
// If there is no context at this point, it is likely no ROOT context
// has been deployed
if (request.getContext() == null) {
res.setStatus(404);
res.setMessage("Not found");
// No context, so use host
Host host = request.getHost();
// Make sure there is a host (might not be during shutdown)
if (host != null) {
host.logAccess(request, response, 0, true);
}
return false;
}
// Now we have the context, we can parse the session ID from the URL
// (if any). Need to do this before we redirect in case we need to
// include the session id in the redirect
String sessionID;
if (request.getServletContext().getEffectiveSessionTrackingModes()
.contains(SessionTrackingMode.URL)) {
// Get the session ID if there was one
sessionID = request.getPathParameter(
SessionConfig.getSessionUriParamName(
request.getContext()));
if (sessionID != null) {
request.setRequestedSessionId(sessionID);
request.setRequestedSessionURL(true);
}
}
// Look for session ID in cookies and SSL session
//从cookie中解析sessionId
parseSessionCookiesId(req, request);
//从ssl中解析sessionId
parseSessionSslId(request);
sessionID = request.getRequestedSessionId();
mapRequired = false;
if (version != null && request.getContext() == versionContext) {
// We got the version that we asked for. That is it.
} else {
version = null;
versionContext = null;
Object[] contexts = request.getMappingData().contexts;
// Single contextVersion means no need to remap
// No session ID means no possibility of remap
if (contexts != null && sessionID != null) {
// Find the context associated with the session
for (int i = (contexts.length); i > 0; i--) {
Context ctxt = (Context) contexts[i - 1];
if (ctxt.getManager().findSession(sessionID) != null) {
// We found a context. Is it the one that has
// already been mapped?
if (!ctxt.equals(request.getMappingData().context)) {
// Set version so second time through mapping
// the correct context is found
version = ctxt.getWebappVersion();
versionContext = ctxt;
// Reset mapping
request.getMappingData().recycle();
mapRequired = true;
// Recycle session info in case the correct
// context is configured with different settings
request.recycleSessionInfo();
}
break;
}
}
}
}
if (!mapRequired && request.getContext().getPaused()) {
// Found a matching context but it is paused. Mapping data will
// be wrong since some Wrappers may not be registered at this
// point.
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// Should never happen
}
// Reset mapping
request.getMappingData().recycle();
mapRequired = true;
}
}
// Possible redirect
MessageBytes redirectPathMB = request.getMappingData().redirectPath;
if (!redirectPathMB.isNull()) {
String redirectPath = urlEncoder.encode(redirectPathMB.toString());
String query = request.getQueryString();
if (request.isRequestedSessionIdFromURL()) {
// This is not optimal, but as this is not very common, it
// shouldn't matter
redirectPath = redirectPath + ";" +
SessionConfig.getSessionUriParamName(
request.getContext()) +
"=" + request.getRequestedSessionId();
}
if (query != null) {
// This is not optimal, but as this is not very common, it
// shouldn't matter
redirectPath = redirectPath + "?" + query;
}
response.sendRedirect(redirectPath);
request.getContext().logAccess(request, response, 0, true);
return false;
}
// Filter trace method
if (!connector.getAllowTrace()
&& req.method().equalsIgnoreCase("TRACE")) {
Wrapper wrapper = request.getWrapper();
String header = null;
if (wrapper != null) {
String[] methods = wrapper.getServletMethods();
if (methods != null) {
for (int i=0; i<methods.length; i++) {
if ("TRACE".equals(methods[i])) {
continue;
}
if (header == null) {
header = methods[i];
} else {
header += ", " + methods[i];
}
}
}
}
res.setStatus(405);
res.addHeader("Allow", header);
res.setMessage("TRACE method is not allowed");
request.getContext().logAccess(request, response, 0, true);
return false;
}
doConnectorAuthenticationAuthorization(req, request);
return true;
}
方法很长但是没做任何删减，重要的逻辑都加了注释，因为在讲解session的文章中我们查看过解析session的部分方法，所以这里我们只看一个比较重要的，其他的有兴趣可以自行查看。根据请求匹配host,context的方法：
//代码清单14
connector.getMapper().map(serverName,decodedURI, version,request.getMappingData());
我们查看connector的getMapper()方法：
//代码清单15
/**
* Mapper.
*/
protected Mapper mapper = new Mapper();
/**
* Return the mapper.
*/
public Mapper getMapper() {
return (mapper);
}
可以看到返回的mapper对象就是上篇文章中介绍的Mapper的实例对象，map()方法调用的就是Mapper类的map()方法：
//代码清单16
/**
* Map the specified host name and URI, mutating the given mapping data.
*
* @param host Virtual host name
* @param uri URI
* @param mappingData This structure will contain the result of the mapping
* operation
*/
public void map(MessageBytes host, MessageBytes uri, String version,
MappingData mappingData)
throws Exception {
if (host.isNull()) {
host.getCharChunk().append(defaultHostName);
}
host.toChars();
uri.toChars();
internalMap(host.getCharChunk(), uri.getCharChunk(), version,
mappingData);
}
最后调用了internalMap()方法：
//代码清单17
/**
* Map the specified URI.
*/
private final void internalMap(CharChunk host, CharChunk uri,
String version, MappingData mappingData) throws Exception {
if (mappingData.host != null) {
// The legacy code (dating down at least to Tomcat 4.1) just
// skipped all mapping work in this case. That behaviour has a risk
// of returning an inconsistent result.
// I do not see a valid use case for it.
throw new AssertionError();
}
uri.setLimit(-1);
// Virtual host mapping
Host[] hosts = this.hosts;
//查找匹配的host
Host mappedHost = exactFindIgnoreCase(hosts, host);
if (mappedHost == null) {
if (defaultHostName == null) {
return;
}
//如果查找不到 使用defaultHostName再次进行查询
mappedHost = exactFind(hosts, defaultHostName);
if (mappedHost == null) {
return;
}
}
//设置匹配到的host
mappingData.host = mappedHost.object;
// Context mapping
ContextList contextList = mappedHost.contextList;
Context[] contexts = contextList.contexts;
int nesting = contextList.nesting;
//查找匹配的context
int pos = find(contexts, uri);
if (pos == -1) {
return;
}
int lastSlash = -1;
int uriEnd = uri.getEnd();
int length = -1;
boolean found = false;
Context context = null;
while (pos >= 0) {
context = contexts[pos];
//比对context
if (uri.startsWith(context.name)) {
length = context.name.length();
if (uri.getLength() == length) {
found = true;
break;
} else if (uri.startsWithIgnoreCase("/", length)) {
found = true;
break;
}
}
if (lastSlash == -1) {
lastSlash = nthSlash(uri, nesting + 1);
} else {
lastSlash = lastSlash(uri);
}
uri.setEnd(lastSlash);
pos = find(contexts, uri);
}
uri.setEnd(uriEnd);
if (!found) {
if (contexts[0].name.equals("")) {
context = contexts[0];
} else {
context = null;
}
}
if (context == null) {
return;
}
mappingData.contextPath.setString(context.name);
ContextVersion contextVersion = null;
ContextVersion[] contextVersions = context.versions;
final int versionCount = contextVersions.length;
if (versionCount > 1) {
Object[] contextObjects = new Object[contextVersions.length];
for (int i = 0; i < contextObjects.length; i++) {
contextObjects[i] = contextVersions[i].object;
}
mappingData.contexts = contextObjects;
if (version != null) {
contextVersion = exactFind(contextVersions, version);
}
}
if (contextVersion == null) {
// Return the latest version
// The versions array is known to contain at least one element
contextVersion = contextVersions[versionCount - 1];
}
//设置匹配的context
mappingData.context = contextVersion.object;
mappingData.contextSlashCount = contextVersion.slashCount;
// Wrapper mapping
if (!contextVersion.isPaused()) {
//设置匹配的wrapper
internalMapWrapper(contextVersion, uri, mappingData);
}
}
看到这里postParseRequest()方法就看完了，可以看到该方法对将org.apache.coyote.Request中的内容解析到org.apache.catalina.connector.Request中。再多说一句 adapter是适配器的意思他起到的作用正是把org.apache.coyote.Request适配给org.apache.catalina.connector.Request，这里使用的正是设计模式中的适配器模式。
一个HTTP请求的流转从tomcat接收开始，从endPoint接收请求到交由Http11ConnectionHandler处理，再交由Http11Processor处理，再交由CoyoteAdapter处理，到这里那么HTTP请求的处理Connector部分以及将请求从Connector适配到Container处理CoyoteAdapter部分都已讲解完毕，后面我们会继续查看处理的后半段也就是Container部分是如何处理Connector传递过来的HTTP请求的。