Tomcat中管道

Posted 2021-02-02 wansw

处理模式

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 className="org.apache.catalina.valves.AccessLogValve" directory="logs"
               prefix="localhost_access_log" suffix=".txt"
               pattern="%h %l %u %t &quot;%r&quot; %s %b" />
    </Host>
</Engine>

public StandardEngine() {
    super();
    // pipeline 设置 basicValve
    pipeline.setBasic(new StandardEngineValve());
    setJvmRoute(System.getProperty("jvmRoute"));
    // 设置等待时间 10
    backgroundProcessorDelay = 10;
}

public StandardHost() {
    super();
    // 设置 BasicValve
    pipeline.setBasic(new StandardHostValve());
}

public StandardContext() {
    super();
    // 设置 BasicValve
    pipeline.setBasic(new StandardContextValve());
    broadcaster = new NotificationBroadcasterSupport();
    if (!Globals.STRICT_SERVLET_COMPLIANCE) {
        resourceOnlyServlets.add("jsp");
    }
}

public StandardWrapper() {
    super();
    swValve = new StandardWrapperValve();
    // 设置 BasicValve
    pipeline.setBasic(swValve);
    broadcaster = new NotificationBroadcasterSupport();
}

// 在处理 server.xml 中配置的 Valve 时
// StandardPipeline
@Override
public void addValve(Valve valve) {
    // 验证 是否可以 绑定 容器（Engine、Host等）
    if (valve instanceof Contained)
        ((Contained) valve).setContainer(this.container);
    // 是否有必要调用 start 方法
    if (getState().isAvailable()) {
        if (valve instanceof Lifecycle) {
            ((Lifecycle) valve).start();
        }
    }
    // first valve == null，就设置为第一个
    if (first == null) {
        first = valve;
        valve.setNext(basic);
    } else {
        // 组装成链式结构，first + 新增 valve + basicValve
        Valve current = first;
        while (current != null) {
            if (current.getNext() == basic) {
                current.setNext(valve);
                valve.setNext(basic);
                break;
            }
            current = current.getNext();
        }
    }
    container.fireContainerEvent(Container.ADD_VALVE_EVENT, valve);
}

注册的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;
    }
    // 返回注册的 BasicValve
    return 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",
                          request.getServerName()));
        return;
    }
    // 判断是否 支持异步
    if (request.isAsyncSupported()) {
        request.setAsyncSupported(host.getPipeline().isAsyncSupported());
    }
    // 执行Host的Pipeline
    host.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;
    }

    // 判断是否 支持异步
    if (request.isAsyncSupported()) {
        request.setAsyncSupported(context.getPipeline().isAsyncSupported());
    }

    boolean 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 {
                // Make sure this request/response is here because an error
                // report is required.
                if (!response.isErrorReportRequired()) {
                    throw new IllegalStateException(sm.getString("standardHost.asyncStateError"));
                }
            }
        } catch (Throwable t) {
            ExceptionUtils.handleThrowable(t);
            container.getLogger().error("Exception Processing " + request.getRequestURI(), t);
            // If a new error occurred while trying to report a previous
            // error allow the original error to be reported.
            if (!response.isErrorReportRequired()) {
                request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, t);
                throwable(request, response, t);
            }
        }

        // Now that the request/response pair is back under container
        // control lift the suspension so that the error handling can
        // complete and/or the container can flush any remaining data
        response.setSuspended(false);

        Throwable t = (Throwable) request.getAttribute(RequestDispatcher.ERROR_EXCEPTION);

        // Protect against NPEs if the context was destroyed during a
        // long running request.
        if (!context.getState().isAvailable()) {
            return;
        }

        // Look for (and render if found) an application level error page
        if (response.isErrorReportRequired()) {
            if (t != null) {
                throwable(request, response, t);
            } else {
                status(request, response);
            }
        }

        if (!request.isAsync() && !asyncAtStart) {
            context.fireRequestDestroyEvent(request.getRequest());
        }
    } finally {
        if (ACCESS_SESSION) {
            request.getSession(false);
        }
        context.unbind(Globals.IS_SECURITY_ENABLED, MY_CLASSLOADER);
    }
}

// StandardContextValve
@Override
public final void invoke(Request request, Response response) {
    // 不允许直接访问 WEB-INF or META-INF
    MessageBytes requestPathMB = request.getRequestPathMB();
    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、MappingData 设置好了。
    Wrapper wrapper = request.getWrapper();
    if (wrapper == null || wrapper.isUnavailable()) {
        response.sendError(HttpServletResponse.SC_NOT_FOUND);
        return;
    }

    // 判断是否 支持异步
    if (request.isAsyncSupported()) {
        request.setAsyncSupported(wrapper.getPipeline().isAsyncSupported());
    }
    // 调用 Wrapper 的 pipeline 处理
    wrapper.getPipeline().getFirst().invoke(request, response);
}

// StandardWrapperValve
@Override
public final void invoke(Request request, Response response)
    throws IOException, ServletException {

    // 分配 servlet 实例处理此次请求
    if (!unavailable) {
        // 会判断  Servlet 是否有  
        // ******* SingleThreadModel 接口 ******************
        servlet = wrapper.allocate();
    }
   
    // 创建 Filter
    ApplicationFilterChain filterChain =
        ApplicationFilterFactory.createFilterChain(request, wrapper, servlet);

    // Call the filter chain for this request
    if ((servlet != null) && (filterChain != null)) {
        if (request.isAsyncDispatching()) {
            request.getAsyncContextInternal().doInternalDispatch();
        } else {
            // 此处 先调用 过滤器方法，再走service 方法
            filterChain.doFilter(request.getRequest(),
                                 response.getResponse());
        }
    } else {
        if (request.isAsyncDispatching()) {
            request.getAsyncContextInternal().doInternalDispatch();
        } else {
            filterChain.doFilter
                (request.getRequest(), response.getResponse());
        }
    }

    // 释放 filterChain
    if (filterChain != null) {
        filterChain.release();
    }

    // 释放 instance
    if (servlet != null) {
        wrapper.deallocate(servlet);
    }

    // 如果不可用，卸载 & 释放 instance
    if ((servlet != null) &&
        (wrapper.getAvailable() == Long.MAX_VALUE)) {
        wrapper.unload();
    }
}

@Override
public void doFilter(ServletRequest request, ServletResponse response) {
    // 省略部分代码
    internalDoFilter(request,response);
}

private void internalDoFilter(ServletRequest request,
                              ServletResponse response) {
    // 省略部分代码，此处转交给DispatcherServlet处理
    // We fell off the end of the chain -- call the servlet instance
    servlet.service(request, response);
}