你的老朋友Tomcat😺(三)请求过程的源码分析

Z时代
2024-01-10
分类：技术分享

前面两篇文章介绍了Tomcat的总体架构和核心组件，这一篇通过debug源码的方式来分析一个请求在Tomcat中是如何处理的，在默认的NIO模式下分析。

一、添加自定义Servlet

在项目中创建如图所示的文件夹以及文件：

MyServlet和web.xml如下所示：

public class MyServlet extends HttpServlet {
@Override
protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
PrintWriter pw = resp.getWriter();
pw.println("hello MyServlet");
}
}

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee
http://xmlns.jcp.org/xml/ns/javaee/web-app_4_0.xsd"
version="4.0"
metadata-complete="true">
<description>
Servlet and JSP Examples.
</description>
<display-name>Servlet and JSP Examples</display-name>
<request-character-encoding>UTF-8</request-character-encoding>
<servlet>
<servlet-name>myServlet</servlet-name>
<servlet-class>mywebapp.MyServlet</servlet-class>
</servlet>
<servlet-mapping>
<servlet-name>myServlet</servlet-name>
<url-pattern>/myServlet</url-pattern>
</servlet-mapping>
</web-app>

接下来接可以访问localhost:8080/mywebapp/myServlet路由debug源码了

二、源码分析

2.1 Connector部分分析

在上一节中我们讲到了连接器中比较重要的组件有Endpoint、Processor、Adapter，其中Endpoint中有负责监听socket连接的组件Acceptor和处理socket请求的组件SocketProcessor。首先看Acceptor，它实现了Runnable接口run()方法如下：

这个方法中比较重要的地方有：

endpoint.countUpOrAwaitConnection();它调用的是子组件LimitLatch的countUpOrAwait()方法，而LimitLatch的子类Sync继承了AQS重写了tryAcquireShared()方法限制最大连接数，默认的最大连接数为8*1024；

socket = endpoint.serverSocketAccept();启动之后阻塞在这里等待下一次socket连接

if (!endpoint.setSocketOptions(socket))处理指定连接，将socket移交给指定处理器

NioEndpoint.setSocketOptions()方法用来设置SocketChannel为非阻塞模式以及其他属性，生成一个NioChannel和NioSocketWrapper对象并设置属性，将SocketChannel、NioSocketWrapper设置为NioChannel的属性，SocketWrapperBase这个包装类是为了屏蔽不同I/O模式下的Channel类的差异，再将它们注册到Poller组件poller.register(channel, socketWrapper);也就是封装一个PollerEvent对象加入Poller组件的Queue里。

// Poller
@Override
public void run() {
// Loop until destroy() is called
while (true) {
boolean hasEvents = false;
try {
if (!close) {
hasEvents = events();
if (wakeupCounter.getAndSet(-1) > 0) {
// If we are here, means we have other stuff to do Do a non blocking select
keyCount = selector.selectNow();
} else {
keyCount = selector.select(selectorTimeout);
}
wakeupCounter.set(0);
}
……
} catch (Throwable x) {
……
continue;
}
// Either we timed out or we woke up, process events first
if (keyCount == 0) {
hasEvents = (hasEvents | events());
}
Iterator<SelectionKey> iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;
// Walk through the collection of ready keys and dispatch  any active event.
while (iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
// Attachment may be null if another thread has called cancelledKey()
if (socketWrapper == null) {
iterator.remove();
} else {
iterator.remove();
// 处理发生事件的key
processKey(sk, socketWrapper);
}
}
timeout(keyCount,hasEvents);
}
getStopLatch().countDown();
}
// Poller
protected void processKey(SelectionKey sk, NioSocketWrapper socketWrapper) {
try {
if (close) {
cancelledKey(sk, socketWrapper);
} elseif (sk.isValid() && socketWrapper != null) {
if (sk.isReadable() || sk.isWritable()) {
……
// Read goes before write
if (sk.isReadable()) {
if (socketWrapper.readOperation != null) {
if (!socketWrapper.readOperation.process()) {
closeSocket = true;
}
} elseif (!processSocket(socketWrapper, SocketEvent.OPEN_READ, true)) {
closeSocket = true;
}
}
……
}
} else {
cancelledKey(sk, socketWrapper);
}
} catch (CancelledKeyException ckx) {
……
}
}
// AbstractEndpoint
public boolean processSocket(SocketWrapperBase<S> socketWrapper, SocketEvent event, boolean dispatch) {
try {
if (socketWrapper == null) {
returnfalse;
}
SocketProcessorBase<S> sc = null;
if (processorCache != null) {
sc = processorCache.pop();
}
if (sc == null) {
sc = createSocketProcessor(socketWrapper, event);
} else {
sc.reset(socketWrapper, event);
}
// debug-tomcat9-print
System.out.println("AbstractEndpoint.processSocket(), event:" + event + ", socketProcessor:" + sc.hashCode());
Executor executor = getExecutor();
if (dispatch && executor != null) {
executor.execute(sc);
} else {
sc.run();
}
} catch (RejectedExecutionException ree) {
……
}
returntrue;
}
// SocketProcessor
@Override
protected void doRun() {
// debug-tomcat9-print
System.out.println("SocketProcessor.doRun(), event:" + event + ", Thread:" + Thread.currentThread().getId() + ", socketProcessor:" + this.hashCode());
NioChannel socket = socketWrapper.getSocket();
Poller poller = NioEndpoint.this.poller;
……
try {
int handshake = -1;
try {
if (socket.isHandshakeComplete()) {
handshake = 0;
} elseif (event == SocketEvent.STOP || event == SocketEvent.DISCONNECT ||
event == SocketEvent.ERROR) {
handshake = -1;
} else {
handshake = socket.handshake(event == SocketEvent.OPEN_READ, event == SocketEvent.OPEN_WRITE);
event = SocketEvent.OPEN_READ;
}
} catch (IOException x) {
……
}
if (handshake == 0) {
SocketState state = SocketState.OPEN;
if (event == null) {
state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
} else {
// debug-tomcat9-work-connector
state = getHandler().process(socketWrapper, event);
}
if (state == SocketState.CLOSED) {
poller.cancelledKey(socket.getIOChannel().keyFor(poller.getSelector()), socketWrapper);
}
} elseif (handshake == -1 ) {
getHandler().process(socketWrapper, SocketEvent.CONNECT_FAIL);
poller.cancelledKey(socket.getIOChannel().keyFor(poller.getSelector()), socketWrapper);
} elseif (handshake == SelectionKey.OP_READ){
socketWrapper.registerReadInterest();
} elseif (handshake == SelectionKey.OP_WRITE){
socketWrapper.registerWriteInterest();
}
} catch (CancelledKeyException cx) {
……
} finally {
……
}
}
// AbstractProtocol
@Override
public SocketState process(SocketWrapperBase<S> wrapper, SocketEvent status) {
……
S socket = wrapper.getSocket();
Processor processor = (Processor) wrapper.getCurrentProcessor();
……
try {
if (processor == null) {
……
}
if (processor == null) {
processor = recycledProcessors.pop();
}
if (processor == null) {
processor = getProtocol().createProcessor();
register(processor);
}
processor.setSslSupport(wrapper.getSslSupport(getProtocol().getClientCertProvider()));
wrapper.setCurrentProcessor(processor);
SocketState state = SocketState.CLOSED;
do {
// debug-tomcat9-work-connector
state = processor.process(wrapper, status);
……
} while ( state == SocketState.UPGRADING);
return state;
}
……
}
// AbstractProcessorLight
@Override
public SocketState process(SocketWrapperBase<?> socketWrapper, SocketEvent status)
throws IOException {
// debug-tomcat9-work-connector
SocketState state = SocketState.CLOSED;
Iterator<DispatchType> dispatches = null;
do {
……
} elseif (status == SocketEvent.OPEN_READ) {
state = service(socketWrapper);
} elseif (status == SocketEvent.CONNECT_FAIL) {
……
} while (state == SocketState.ASYNC_END ||
dispatches != null && state != SocketState.CLOSED);
return state;
}
// Http11Processor
@Override
public SocketState service(SocketWrapperBase<?> socketWrapper) throws IOException {
RequestInfo rp = request.getRequestProcessor();
rp.setStage(org.apache.coyote.Constants.STAGE_PARSE);
setSocketWrapper(socketWrapper);
……
// debug-tomcat9-work-connector
getAdapter().service(request, response);
……
}
// CoyoteAdapter
@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().setQueryStringCharset(connector.getURICharset());
}
if (connector.getXpoweredBy()) {
response.addHeader("X-Powered-By", POWERED_BY);
}
boolean async = false;
boolean postParseSuccess = false;
req.getRequestProcessor().setWorkerThreadName(THREAD_NAME.get());
try {
// Parse and set Catalina and configuration specific request parameters
// debug-tomcat9-work
postParseSuccess = postParseRequest(req, request, res, response);
if (postParseSuccess) {
//check valves if we support async
request.setAsyncSupported(connector.getService().getContainer().getPipeline().isAsyncSupported());
// Calling the container
// debug-tomcat9-doc 获取Engine调用pipeline链式处理
connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);
}
……
// 回写request、response
request.finishRequest();
response.finishResponse();
……
}

poller线程不断循环取出events中的事件，进行处理，注册channel到selector上并监听OP_READ事件；selector对有事件发生的SelectionKey调用processKey(SelectionKey sk, NioSocketWrapper socketWrapper)方法；

对读写事件的key将对应的socketWrapper调用processSocket(SocketWrapperBase<T> socketWrapper, SocketEvent event, boolean dispatch)方法处理；

根据socketWrapper和event创建一个SocketProcessor对象(其父类SocketProcessorBase实现了Runnable接口)提交给线程池executor执行；

SocketProcessor类重写了doRun()方法，握手完成的socketWrapper且event不为null则调用getHandler().process(socketWrapper, event)也就是ConnectionHandler中的process(SocketWrapperBase<S> wrapper, SocketEvent status)方法，在这个方法中拿到一个Processor(Http11Processor)处理应用层协议相关问题，最后调用到Http11Processor.service(SocketWrapperBase<S> socketWrapper)方法；这个方法中解析了HTTP请求头相关信息并调用了getAdapter().service(request, response)；

CoyoteAdapter.service(org.apache.coyote.Request req, org.apache.coyote.Response res)方法将Tomcat的request、response对象转换为Servlet规范的request、response对象；然后调用postParseRequest(req, request, res, response)记录access日志、解析请求URI和后面的参数、后续还有解析session，调用了connector.getService().getMapper().map(serverName, decodedURI, version, request.getMappingData())处理请求映射 (获取 host, context, wrapper)；

2.2 Container部分分析

// Mapper
public void map(MessageBytes host, MessageBytes uri, String version,
MappingData mappingData) throws IOException {
if (host.isNull()) {
String defaultHostName = this.defaultHostName;
if (defaultHostName == null) {
return;
}
host.getCharChunk().append(defaultHostName);
}
host.toChars();
uri.toChars();
internalMap(host.getCharChunk(), uri.getCharChunk(), version, mappingData);
}
// Mapper
private final void internalMap(CharChunk host, CharChunk uri,
String version, MappingData mappingData) throws IOException {
if (mappingData.host != null) {
throw new AssertionError();
}
// Virtual host mapping
MappedHost[] hosts = this.hosts;
// debug-tomcat9-doc 从hosts数组中根据主机名找到MappedHost
MappedHost mappedHost = exactFindIgnoreCase(hosts, host);
if (mappedHost == null) {
// 特殊处理后再次寻找
……
}
mappingData.host = mappedHost.object;
if (uri.isNull()) {
return;
}
uri.setLimit(-1);
// Context mapping
ContextList contextList = mappedHost.contextList;
MappedContext[] contexts = contextList.contexts;
……
// 找到context设置到mappingData
mappingData.context = contextVersion.object;
mappingData.contextSlashCount = contextVersion.slashCount;
// Wrapper mapping
if (!contextVersion.isPaused()) {
// debug-tomcat9-doc 从contextVersion的wrappers数组中根据主机名找到MappedWrapper
internalMapWrapper(contextVersion, uri, mappingData);
}
}
// Mapper
// 根据hostName算出在hosts数组中位置下标，找到Host
private static final <T, E extends MapElement<T>> E exactFindIgnoreCase(E[] map, CharChunk name) {
int pos = findIgnoreCase(map, name);
if (pos >= 0) {
E result = map[pos];
if (name.equalsIgnoreCase(result.name)) {
return result;
}
}
return null;
}
// 继续执行CoyoteAdpter中service()方法的connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);
// StandardEngineValve
@Override
public final void invoke(Request request, Response response) throws IOException, ServletException {
// 前面request中的mappingData已获取到host信息
Host host = request.getHost();
if (host == null) {
// HTTP 0.9 or HTTP 1.0 request without a host when no default host
// is defined. This is handled by the CoyoteAdapter.
return;
}
if (request.isAsyncSupported()) {
request.setAsyncSupported(host.getPipeline().isAsyncSupported());
}
// Ask this Host to process this request
host.getPipeline().getFirst().invoke(request, response);
}
//
public final void invoke(Request request, Response response)
throws IOException, ServletException {
……
StandardWrapper wrapper = (StandardWrapper) getContainer();
Servlet servlet = null;
Context context = (Context) wrapper.getParent();
……
try {
if (!unavailable) {
// debug-tomcat9-doc 分配servlet处理请求
servlet = wrapper.allocate();
}
} catch (UnavailableException e) {
……
} catch (ServletException e) {
container.getLogger().error(sm.getString("standardWrapper.allocateException",
wrapper.getName()), StandardWrapper.getRootCause(e));
throwable = e;
exception(request, response, e);
} catch (Throwable e) {
……
}
……
// debug-tomcat9-doc 获取请求path
MessageBytes requestPathMB = request.getRequestPathMB();
DispatcherType dispatcherType = DispatcherType.REQUEST;
if (request.getDispatcherType()==DispatcherType.ASYNC) dispatcherType = DispatcherType.ASYNC;
request.setAttribute(Globals.DISPATCHER_TYPE_ATTR,dispatcherType);
request.setAttribute(Globals.DISPATCHER_REQUEST_PATH_ATTR, requestPathMB);
// debug-tomcat9-doc 创建filterChain
ApplicationFilterChain filterChain = ApplicationFilterFactory.createFilterChain(request, wrapper, servlet);
Container container = this.container;
try {
if ((servlet != null) && (filterChain != null)) {
// Swallow output if needed
if (context.getSwallowOutput()) {
try {
SystemLogHandler.startCapture();
if (request.isAsyncDispatching()) {
request.getAsyncContextInternal().doInternalDispatch();
} else {
// debug-tomcat9-doc 开始执行filter链,最后到servlet中
filterChain.doFilter(request.getRequest(), response.getResponse());
}
} finally {
String log = SystemLogHandler.stopCapture();
if (log != null && log.length() > 0) {
context.getLogger().info(log);
}
}
} else {
if (request.isAsyncDispatching()) {
request.getAsyncContextInternal().doInternalDispatch();
} else {
filterChain.doFilter
(request.getRequest(), response.getResponse());
}
}
}
} catch (ClientAbortException | CloseNowException e) {
……
} finally {
// Release the filter chain (if any) for this request
if (filterChain != null) {
filterChain.release();
}
// Deallocate the allocated servlet instance
try {
if (servlet != null) {
wrapper.deallocate(servlet);
}
} catch (Throwable e) {
……
}
……
}
}

在Mapper组件的map(MessageBytes host, MessageBytes uri, String version, MappingData mappingData)方法中简单处理了host和uri然后调用internalMap();

internalMap()方法根据host和uri找到Host和Context以及Wrapper；(这些映射关系在容器启动的时候都缓存在Mapper组件中了)

开始调用Tomcat容器, 首先调用StandardEngine容器中的管道PipeLine中的第一个Valve(StandardEngineValve), 传入request与response来处理所有逻辑，之后经过一系列的Valve(StandardHostValve、StandardWrapperValve等)可以理解为容器级别的过滤器

StandardWrapperValve的invoke()方法中进行servlet的分配，创建请求的ApplicationFilterChain，对象包装请求的Servlet对象及一些过滤器Filter对象，执行 filterChain 链, 在链的末尾是servlet调用service(request, response)方法