在《TOMCAT源碼分析——請求原理分析（上）》一文中已經介紹了關于Tomcat7.0處理請求前作的初始化和準備工作，請讀者在閱讀本文前確保掌握《TOMCAT源碼分析——請求原理分析（上）》一文中的相關知識以及HTTP協議和TCP協議的一些內容。本文重點講解Tomcat7.0在準備好接受請求后，請求過程的原理分析。
在正式開始之前，我們先來看看圖1中的Tomcat請求處理架構。
圖1 Tomcat請求處理架構
圖1列出了Tomcat請求處理架構中的主要組件，這里對它們做個簡單介紹：
默認情況下，Tomcat只有一個Acceptor線程，Acceptor不斷循環從ServerSocket中獲取Socket，當并發數大的情況下，這里會不會有性能問題？我想說的是，Acceptor的實現非常輕量級，它只負責兩個動作：獲取Socket和將Socket轉交給SocketProcessor線程處理。另外，我們可以通過在server.xml的Connector配置中增加acceptorThreadCount的值，讓我們同時可以擁有多個Acceptor線程。雖然我們可以修改maxThreads配置把SocketProcessor的線程數設置的很大，但是我們需要區別對待：
原理就講這么多，下面具體分析下Tomcat處理請求的具體實現。
在《TOMCAT源碼分析——請求原理分析（上）》一文中我們曾經介紹過JIoEndpoint的內部類Acceptor，Acceptor實現了Runnable接口。Acceptor作為后臺線程不斷循環，每次循環都會sleep大約1秒鐘（由于是線程級別的，所以并不保證準確），然后接收來自瀏覽器的Socket連接（用戶在瀏覽器輸入HTTP請求地址后，瀏覽器底層實際使用Socket通信的），最后將Socket交給外部類JIoEndpoint的processSocket方法（見代碼清單1）處理。
代碼清單1
/** * Process given socket. */ protected boolean processSocket(Socket socket) { try { SocketWrapper<Socket> wrapper = new SocketWrapper<Socket>(socket); wrapper.setKeepAliveLeft(getMaxKeepAliveRequests()); getExecutor().execute(new SocketProcessor(wrapper)); } catch (RejectedExecutionException x) { log.warn("Socket processing request was rejected for:"+socket,x); return false; } 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; }
根據代碼清單1，JIoEndpoint的processSocket方法的處理步驟如下：
代碼清單2
/** * Max keep alive requests */ private int maxKeepAliveRequests=100; // as in Apache HTTPD server public int getMaxKeepAliveRequests() { return maxKeepAliveRequests; }
代碼清單3
/** * This class is the equivalent of the Worker, but will simply use in an * external Executor thread pool. */ protected class SocketProcessor implements Runnable { protected SocketWrapper<Socket> socket = null; protected SocketStatus status = null; public SocketProcessor(SocketWrapper<Socket> socket) { if (socket==null) throw new NullPointerException(); this.socket = socket; } public SocketProcessor(SocketWrapper<Socket> socket, SocketStatus status) { this(socket); this.status = status; } public void run() { boolean launch = false; try { if (!socket.processing.compareAndSet(false, true)) { log.error("Unable to process socket. Invalid state."); return; } SocketState state = SocketState.OPEN; // Process the request from this socket if ( (!socket.isInitialized()) && (!setSocketOptions(socket.getSocket())) ) { state = SocketState.CLOSED; } socket.setInitialized(true); if ( (state != SocketState.CLOSED) ) { state = (status==null)?handler.process(socket):handler.process(socket,status); } if (state == SocketState.CLOSED) { // Close socket if (log.isTraceEnabled()) { log.trace("Closing socket:"+socket); } try { socket.getSocket().close(); } catch (IOException e) { // Ignore } } else if (state == SocketState.OPEN){ socket.setKeptAlive(true); socket.access(); //keepalive connection //TODO - servlet3 check async status, we may just be in a hold pattern launch = true; } else if (state == SocketState.LONG) { socket.access(); waitingRequests.add(socket); } } finally { socket.processing.set(false); if (launch) getExecutor().execute(new SocketProcessor(socket)); socket = null; } // Finish up this request } }
SocketProcessor線程專門用于處理Acceptor轉交的Socket，其執行步驟如下：
代碼清單4
/** * Set the options for the current socket. */ protected boolean setSocketOptions(Socket socket) { // Process the connection try { // 1: Set socket options: timeout, linger, etc 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) { log.error(sm.getString("endpoint.err.unexpected"), t); // Close the socket return false; } try { // 2: SSL handshake serverSocketFactory.handshake(socket); } catch (Throwable t) { if (log.isDebugEnabled()) { log.debug(sm.getString("endpoint.err.handshake"), t); } // Tell to close the socket return false; } return true; }
代碼清單5
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()); if (soTrafficClass != null) socket.setTrafficClass(soTrafficClass.intValue()); }
以Http11ConnectionHandler為例，我們重點分析它是如何進一步處理Socket的。Http11ConnectionHandler的process方法，見代碼清單6。
代碼清單6
public SocketState process(SocketWrapper<Socket> socket) { return process(socket,SocketStatus.OPEN); } public SocketState process(SocketWrapper<Socket> socket, SocketStatus status) { Http11Processor processor = connections.remove(socket); boolean recycle = true; try { if (processor == null) { processor = recycledProcessors.poll(); } if (processor == null) { processor = createProcessor(); } processor.action(ActionCode.ACTION_START, null); if (proto.isSSLEnabled() && (proto.sslImplementation != null)) { processor.setSSLSupport (proto.sslImplementation.getSSLSupport(socket.getSocket())); } else { processor.setSSLSupport(null); } SocketState state = socket.isAsync()?processor.asyncDispatch(status):processor.process(socket); if (state == SocketState.LONG) { connections.put(socket, processor); socket.setAsync(true); recycle = false; } else { connections.remove(socket); socket.setAsync(false); } return state; } catch(java.net.SocketException e) { // SocketExceptions are normal Http11Protocol.log.debug (sm.getString ("http11protocol.proto.socketexception.debug"), e); } catch (java.io.IOException e) { // IOExceptions are normal Http11Protocol.log.debug (sm.getString ("http11protocol.proto.ioexception.debug"), e); } // Future developers: if you discover any other // rare-but-nonfatal exceptions, catch them here, and log as // above. catch (Throwable 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. Http11Protocol.log.error (sm.getString("http11protocol.proto.error"), e); } finally { // if(proto.adapter != null) proto.adapter.recycle(); // processor.recycle(); if (recycle) { processor.action(ActionCode.ACTION_STOP, null); recycledProcessors.offer(processor); } } return SocketState.CLOSED; }
根據代碼清單6，可見Http11ConnectionHandler的process方法的處理步驟如下：
代碼清單7
protected Http11Processor createProcessor() { Http11Processor processor = new Http11Processor(proto.getMaxHttpHeaderSize(), (JIoEndpoint)proto.endpoint); processor.setAdapter(proto.adapter); processor.setMaxKeepAliveRequests(proto.getMaxKeepAliveRequests()); processor.setKeepAliveTimeout(proto.getKeepAliveTimeout()); processor.setTimeout(proto.getTimeout()); 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()); register(processor); return processor; }
根據之前的分析，我們知道Socket的處理方式有異步和同步兩種，分別調用Http11Processor的asyncDispatch方法或process方法，我們以同步處理為例，來看看接下來的處理邏輯。
Http11Processor的process方法（見代碼清單8）用于同步處理，由于其代碼很多，所以此處在代碼后面追加一些注釋，便于讀者理解。這里面有一些關鍵方法重點拿出來解釋下：
代碼清單8
RequestInfo rp = request.getRequestProcessor(); rp.setStage(org.apache.coyote.Constants.STAGE_PARSE); this.socket = socketWrapper; inputBuffer.setInputStream(socket.getSocket().getInputStream());//設置輸入流 outputBuffer.setOutputStream(socket.getSocket().getOutputStream());//設置輸出流 int keepAliveLeft = maxKeepAliveRequests>0?socketWrapper.decrementKeepAlive():-1;//保持連接遞減 int soTimeout = endpoint.getSoTimeout();//socket超時時間 socket.getSocket().setSoTimeout(soTimeout);//設置超時時間 boolean keptAlive = socketWrapper.isKeptAlive();//是否保持連接 while (started && !error && keepAlive) { // Parsing the request header try { //TODO - calculate timeout based on length in queue (System.currentTimeMills() - wrapper.getLastAccess() is the time in queue) if (keptAlive) {//是否保持連接 if (keepAliveTimeout > 0) { socket.getSocket().setSoTimeout(keepAliveTimeout); } else if (soTimeout > 0) { socket.getSocket().setSoTimeout(soTimeout); } } inputBuffer.parseRequestLine(false);//讀取請求行 request.setStartTime(System.currentTimeMillis()); keptAlive = true; if (disableUploadTimeout) { socket.getSocket().setSoTimeout(soTimeout); } else { socket.getSocket().setSoTimeout(timeout); } inputBuffer.parseHeaders();//解析請求頭 } catch (IOException e) { error = true; break; } catch (Throwable t) { if (log.isDebugEnabled()) { log.debug(sm.getString("http11processor.header.parse"), t); } // 400 - Bad Request response.setStatus(400); adapter.log(request, response, 0); error = true; } if (!error) { // Setting up filters, and parse some request headers rp.setStage(org.apache.coyote.Constants.STAGE_ENDED); try { prepareRequest();//對請求內容增加過濾器——協議、方法、請求頭、host等 } catch (Throwable t) { if (log.isDebugEnabled()) { log.debug(sm.getString("http11processor.request.prepare"), t); } // 400 - Internal Server Error response.setStatus(400); adapter.log(request, response, 0); error = true; } } if (maxKeepAliveRequests > 0 && keepAliveLeft == 0) keepAlive = false; // Process the request in the adapter if (!error) { try { rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE); adapter.service(request, response); //將進一步處理交給CoyoteAdapter // 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 && !error) { // Avoid checking twice. error = response.getErrorException() != null || statusDropsConnection(response.getStatus()); } } catch (InterruptedIOException e) { error = true; } catch (Throwable t) { log.error(sm.getString("http11processor.request.process"), t); // 500 - Internal Server Error response.setStatus(500); adapter.log(request, response, 0); error = true; } } // Finish the handling of the request try { rp.setStage(org.apache.coyote.Constants.STAGE_ENDINPUT); // 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. if(error && !async) inputBuffer.setSwallowInput(false); if (!async) endRequest(); } catch (Throwable t) { log.error(sm.getString("http11processor.request.finish"), t); // 500 - Internal Server Error response.setStatus(500); adapter.log(request, response, 0); error = true; } try { rp.setStage(org.apache.coyote.Constants.STAGE_ENDOUTPUT); } catch (Throwable t) { log.error(sm.getString("http11processor.response.finish"), t); error = true; } // If there was an error, make sure the request is counted as // and error, and update the statistics counter if (error) { response.setStatus(500); } request.updateCounters(); rp.setStage(org.apache.coyote.Constants.STAGE_KEEPALIVE); // Don't reset the param - we'll see it as ended. Next request // will reset it // thrA.setParam(null); // Next request if (!async || error) { inputBuffer.nextRequest(); outputBuffer.nextRequest(); } //hack keep alive behavior break; } rp.setStage(org.apache.coyote.Constants.STAGE_ENDED); if (error) { recycle(); return SocketState.CLOSED; } else if (async) { return SocketState.LONG; } else { if (!keepAlive) { recycle(); return SocketState.CLOSED; } else { return SocketState.OPEN; } }
從代碼清單8可以看出，最后的請求處理交給了CoyoteAdapter，CoyoteAdapter的service方法（見代碼清單9）用于真正處理請求。
代碼清單9
/** * Service method. */ 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; try { // Parse and set Catalina and configuration specific // request parameters req.getRequestProcessor().setWorkerThreadName(Thread.currentThread().getName()); if (postParseRequest(req, request, res, response)) { //check valves if we support async request.setAsyncSupported(connector.getService().getContainer().getPipeline().isAsyncSupported()); // Calling the container 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)) { comet = true; res.action(ActionCode.ACTION_COMET_BEGIN, null); } } else { comet = true; res.action(ActionCode.ACTION_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); } } } AsyncContextImpl asyncConImpl = (AsyncContextImpl)request.getAsyncContext(); if (asyncConImpl!=null && asyncConImpl.getState()==AsyncContextImpl.AsyncState.STARTED) { res.action(ActionCode.ACTION_ASYNC_START, request.getAsyncContext()); async = true; } else if (request.isAsyncDispatching()) { asyncDispatch(req, res, SocketStatus.OPEN); if (request.isAsyncStarted()) { async = true; res.action(ActionCode.ACTION_ASYNC_START, request.getAsyncContext()); } } else if (!comet) { response.finishResponse(); req.action(ActionCode.ACTION_POST_REQUEST , null); } } catch (IOException e) { // Ignore } catch (Throwable t) { log.error(sm.getString("coyoteAdapter.service"), t); } finally { req.getRequestProcessor().setWorkerThreadName(null); // Recycle the wrapper request and response if (!comet && !async) { request.recycle(); response.recycle(); } else { // Clear converters so that the minimum amount of memory // is used by this processor request.clearEncoders(); response.clearEncoders(); } } }
從代碼清單9可以看出，CoyoteAdapter的service方法的執行步驟如下：
將真正的請求處理交給Engine的Pipeline去處理，代碼：connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);
代碼清單10
/** * Parse additional request parameters. */ protected boolean postParseRequest(org.apache.coyote.Request req, Request request, org.apache.coyote.Response res, Response response) throws Exception {// 省略前邊的次要代碼 parsePathParameters(req, request); // URI decoding // %xx decoding of the URL try { 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 if (!checkNormalize(req.decodedURI())) { res.setStatus(400); res.setMessage("Invalid URI character encoding"); connector.getService().getContainer().logAccess( request, response, 0, true); return false; } // Set the remote principal String principal = req.getRemoteUser().toString(); if (principal != null) { request.setUserPrincipal(new CoyotePrincipal(principal)); } // Set the authorization type String authtype = req.getAuthType().toString(); if (authtype != null) { request.setAuthType(authtype); } // Request mapping. MessageBytes serverName; if (connector.getUseIPVHosts()) { serverName = req.localName(); if (serverName.isNull()) { // well, they did ask for it res.action(ActionCode.ACTION_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(); } connector.getMapper().map(serverName, decodedURI, request.getMappingData()); request.setContext((Context) request.getMappingData().context); request.setWrapper((Wrapper) request.getMappingData().wrapper); // 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; } // 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 if (request.getServletContext().getEffectiveSessionTrackingModes() .contains(SessionTrackingMode.URL)) { // Get the session ID if there was one String sessionID = request.getPathParameter( applicationSessionCookieConfig.getSessionUriParamName( request.getContext())); if (sessionID != null) { request.setRequestedSessionId(sessionID); request.setRequestedSessionURL(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 + ";" + ApplicationSessionCookieConfig.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; } // Finally look for session ID in cookies and SSL session parseSessionCookiesId(req, request); parseSessionSslId(request); return true; }
從代碼清單10可以看出，postParseRequest方法的執行步驟如下：
代碼清單11
public void map(MessageBytes host, MessageBytes uri, MappingData mappingData) throws Exception { if (host.isNull()) { host.getCharChunk().append(defaultHostName); } host.toChars(); uri.toChars(); internalMap(host.getCharChunk(), uri.getCharChunk(), mappingData); }
CoyoteAdapter的service方法最后會將請求交給Engine的Pipeline去處理，我將在《Tomcat源碼分析——請求原理分析（下）》一文中具體講解。
如需轉載，請標明本文作者及出處——作者：jiaan.gja，本文原創首發：博客園，原文鏈接：http://www.49028c.com/jiaan-geng/p/4894832.html
新聞熱點
疑難解答
