Pipeline

节选部分源码、源码版本 Tomcat8.5

处理模式

Pipeline--Valve是一种责任链模式，它和普通责任链模式有两点区别：

• 每个Pipeline都是有特定的Valve，而且是在管道的最后一个执行，这个Valve叫BaseValve，并且BaseValve是不可删除的；
• 在上层容器的管道的BaseValve中会调用下层容器的管道。
• "容器的 过滤器"

Tomcat中的管道

​ Tomcat按照包含关系有4个级别的容器,标准实现分别是：

• StandardEngine
• StandardHost
• StandardContext
• StandardWrapper

请求对象和响应对象分别在这4个容器之间通过管道机制进行传递。

初始化

• 在 server.xml 中 配置 的默认 阀门：

<Engine name="Catalina" defaultHost="localhost"><Host name="localhost"  appBase="webapps"unpackWARs="true" autoDeploy="true"><!-- 默认 Valve --><Valve className="org.apache.catalina.valves.AccessLogValve"directory="logs"prefix="localhost_access_log"suffix=".txt"<!-- maxDays="5" -->pattern="%h %l %u %t &quot;%r&quot; %s %b" /><!-- 自定义 valve --><Valve className="org.apache.catalina.valves.WswAccessValve"/></Host>
</Engine>

自定义Valve

public class WswAccessValve extends ValveBase {@Overridepublic void invoke(Request request, Response response) {String uri = request.getRequestURI();System.out.println("uri = " + uri);getNext().invoke(request, response);}
}

四个基础Valve

public StandardEngine() {super();pipeline.setBasic(new StandardEngineValve());
}public StandardHost() {super();pipeline.setBasic(new StandardHostValve());
}public StandardContext() {super();pipeline.setBasic(new StandardContextValve());
}public StandardWrapper() {super();pipeline.setBasic(new StandardWrapperValve());
}

构造Valve链

// 在处理 server.xml 中配置的 Valve 时
// StandardPipeline
@Override
public void setBasic(Valve valve) {Valve oldBasic = this.basic;if (oldBasic == valve) { return; }// 不会走的if (oldBasic != null) {if (getState().isAvailable() && (oldBasic instanceof Lifecycle)) {((Lifecycle) oldBasic).stop();}if (oldBasic instanceof Contained) {((Contained) oldBasic).setContainer(null);}if (valve == null) { return; }if (valve instanceof Contained) {((Contained) valve).setContainer(this.container);}// 未执行if (getState().isAvailable() && valve instanceof Lifecycle) {((Lifecycle) valve).start();}Valve current = first;while (current != null) {if (current.getNext() == oldBasic) {current.setNext(valve);break;}current = current.getNext();}this.basic = valve;
}@Override
public void addValve(Valve valve) {// 验证 是否可以 绑定 容器（Engine、Host等）if (valve instanceof Contained)((Contained) valve).setContainer(this.container);// 是否有必要调用 start 方法【默认没有调用，不属于LifeCycle】if (getState().isAvailable()) {if (valve instanceof Lifecycle) {((Lifecycle) valve).start();}}// first valve == null，就设置为第一个if (first == null) {first = valve;valve.setNext(basic);} else {Valve current = first;while (current != null) {if (current.getNext() == basic) {// 按照在xml中定义的顺序+basicValve// 组装成链式结构，first + 新增 valve + basicValvecurrent.setNext(valve);valve.setNext(basic);break;}current = current.getNext();}}// [没有执行]container.fireContainerEvent(Container.ADD_VALVE_EVENT, valve);
}

执行流程

• 在收到请求后，调用"容器过滤器"

// org.apache.catalina.connector.CoyoteAdapter
@Override
public void service(org.apache.coyote.Request req,                                                  org.apache.coyote.Response res) {// 此处调用 Pipeline Value 的 invoke 方法。（Engine是最顶层容器）connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);}
}
// StandardPipeline
@Override
public Valve getFirst() {// 如果有注册Valve，就返回if (first != null) {return first;}// 返回注册的 BasicValvereturn basic;
}

• StandardEngineValve

@Override
public final void invoke(Request request, Response response) {// 根据 当前 request 找到合适的 host，通过 MappingData Host host = request.getHost();// 没有找到 host，不走了if (host == null) {response.sendError(HttpServletResponse.SC_BAD_REQUEST,sm.getString("standardEngine.noHost"));return;}// 执行下一个 Valvehost.getPipeline().getFirst().invoke(request, response);
}

• StandardHostValve

@Override
public final void invoke(Request request, Response response) {Context context = request.getContext();// 未找到匹配的 项目if (context == null) {response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,sm.getString("standardHost.noContext"));return;}// debug 后 都为 falseboolean asyncAtStart = request.isAsync();boolean asyncDispatching = request.isAsyncDispatching();try {context.bind(Globals.IS_SECURITY_ENABLED, MY_CLASSLOADER);// 见下面的说明if (!asyncAtStart && !context.fireRequestInitEvent(request.getRequest())) {return;}try {if (!asyncAtStart || asyncDispatching) {context.getPipeline().getFirst().invoke(request, response);} else {if (!response.isErrorReportRequired()) {throw new IllegalStateException("standardHost.asyncStateError");}}} catch (Throwable t) {if (!response.isErrorReportRequired()) {request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, t);throwable(request, response, t);}}response.setSuspended(false);// 是否有错误信息Throwable t = (Throwable) request.getAttribute(RequestDispatcher.ERROR_EXCEPTION);// 判断 上下文状态是否可用if (!context.getState().isAvailable()) {return;}// 查找（如果找到则呈现）应用程序级别错误页面if (response.isErrorReportRequired()) {if (t != null) {throwable(request, response, t);} else {status(request, response);}}// 调用销毁 request 的方法,见说明if (!request.isAsync() && !asyncAtStart) {context.fireRequestDestroyEvent(request.getRequest());}} finally {if (ACCESS_SESSION) {request.getSession(false);}context.unbind(Globals.IS_SECURITY_ENABLED, MY_CLASSLOADER);}
}

• 在 Listener 章节中 提到 ：ServletRequestListener 的 requestInitialized 方法判断
• Line18：context.fireRequestInitEvent(request.getRequest())

// 说明
@Override
public boolean fireRequestInitEvent(ServletRequest request) {Object instances[] = getApplicationEventListeners();if ((instances != null) && (instances.length > 0)) {ServletRequestEvent event =new ServletRequestEvent(getServletContext(), request);for (int i = 0; i < instances.length; i++) {if (instances[i] == null) { continue; }if (!(instances[i] instanceof ServletRequestListener))continue;ServletRequestListener listener =(ServletRequestListener) instances[i];try {listener.requestInitialized(event);} catch (Throwable t) {request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, t);return false;}}}return true;
}
// 获取所有的 applicationEventListener
// 判断是否是 ServletRequestListener 类型
// 调用 listener.requestInitialized(event); 完成初始化

• Line51：context.fireRequestDestroyEvent(request.getRequest());

@Override
public boolean fireRequestDestroyEvent(ServletRequest request) {Object instances[] = getApplicationEventListeners();if ((instances != null) && (instances.length > 0)) {ServletRequestEvent event =new ServletRequestEvent(getServletContext(), request);for (int i = 0; i < instances.length; i++) {int j = (instances.length -1) -i;if (instances[j] == null) { continue; }if (!(instances[j] instanceof ServletRequestListener))continue;ServletRequestListener listener =(ServletRequestListener) instances[j];try {listener.requestDestroyed(event);} catch (Throwable t) {request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, t);return false;}}}return true;
}
// 和初始化方法极其相似

• StandardContextValve

@Override
public final void invoke(Request request, Response response) {// 测试的请求路径【/index.html】MessageBytes requestPathMB = request.getRequestPathMB();// 不允许直接访问 WEB-INF or META-INF，在这里判断的if ((requestPathMB.startsWithIgnoreCase("/META-INF/", 0))|| (requestPathMB.equalsIgnoreCase("/META-INF"))|| (requestPathMB.startsWithIgnoreCase("/WEB-INF/", 0))|| (requestPathMB.equalsIgnoreCase("/WEB-INF"))) {response.sendError(HttpServletResponse.SC_NOT_FOUND);return;}// 根据当前 request 选择合适的 wrapper【servlet】Wrapper wrapper = request.getWrapper();// 判断状态if (wrapper == null || wrapper.isUnavailable()) {response.sendError(HttpServletResponse.SC_NOT_FOUND);return;}// 发送确认请求，HTTP/1.1 100try {response.sendAcknowledgement();} catch (IOException ioe) {return;}// 判断是否支持异步if (request.isAsyncSupported()) {request.setAsyncSupported(wrapper.getPipeline().isAsyncSupported());}// 调用 Wrapper 的 pipeline 处理wrapper.getPipeline().getFirst().invoke(request, response);
}

• Line6：重点

• StandardWrapperValve

@Override
public final void invoke(Request request, Response response)throws IOException, ServletException {boolean unavailable = false;requestCount.incrementAndGet();StandardWrapper wrapper = (StandardWrapper) getContainer();Servlet servlet = null;Context context = (Context) wrapper.getParent();// Check for the application being marked unavailableif (!context.getState().isAvailable()) {response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,sm.getString("standardContext.isUnavailable"));unavailable = true;}// 分配 servlet 实例处理此次请求if (!unavailable) {// 会判断  Servlet 是否有  // ******* SingleThreadModel 接口 ******************servlet = wrapper.allocate();}// 创建 FilterApplicationFilterChain filterChain =ApplicationFilterFactory.createFilterChain(request, wrapper, servlet);// Call the filter chain for this requestif ((servlet != null) && (filterChain != null)) {if (request.isAsyncDispatching()) {request.getAsyncContextInternal().doInternalDispatch();} else {// 此处 调用 过滤器方法filterChain.doFilter(request.getRequest(),response.getResponse());}} else {if (request.isAsyncDispatching()) {request.getAsyncContextInternal().doInternalDispatch();} else {filterChain.doFilter(request.getRequest(), response.getResponse());}}// 释放 filterChainif (filterChain != null) {filterChain.release();}// 释放 instanceif (servlet != null) {wrapper.deallocate(servlet);}// 如果不可用，卸载 & 释放 instanceif ((servlet != null) &&(wrapper.getAvailable() == Long.MAX_VALUE)) {wrapper.unload();}
}

• Line25：关联 Filter 章节