Tomcat源码解析系列（十一）ProtocolHandler

前言
上篇文章中讲到了 Connector 的初始化与启动，其中最关键的就是 ProtocolHandler 的初始化与启动。tomcat 中 ProtocolHandler 的默认实现类是 Http11NioProtocol。tomcat9.0.16 中 ProtocolHandler 的实现类中还有一个 Http11Nio2Protocol，两者实现上类似。这两个实现的的父类都是 AbstractHttp11JsseProtocol，AbstractHttp11JsseProtocol 的父类是 AbstractHttp11Protocol，AbstractHttp11Protocol 的父类是 AbstractProtocol。ProtocolHandler来处理网络连接和应用层协议，包含两个重要组件：endpoint和processor，endpoint是通信端点，即通信监听的接口，是具体的socket接受和发送处理器，是对传输层的抽象，processor接受来自endpoint的socket，读取字节流解析成Tomcat的request和response对象，并通过adapter将其提交到容器处理，processor是对应用层协议的抽象。

---

1. Http11NioProtocol 构造方法
在 Connector 的构造方法中，用反射创建了一个 Http11NioProtocol 对象。

public Http11NioProtocol() {
    super(new NioEndpoint());
}

public AbstractHttp11JsseProtocol(AbstractJsseEndpoint<S,?> endpoint) {
    super(endpoint);
}

public AbstractHttp11Protocol(AbstractEndpoint<S,?> endpoint) {
    super(endpoint);
    setConnectionTimeout(Constants.DEFAULT_CONNECTION_TIMEOUT);
    ConnectionHandler<S> cHandler = new ConnectionHandler<>(this);
    setHandler(cHandler);
    getEndpoint().setHandler(cHandler);
}

/**
 * Endpoint that provides low-level network I/O - must be matched to the
 * ProtocolHandler implementation (ProtocolHandler using NIO, requires NIO
 * Endpoint etc.).
 */
private final AbstractEndpoint<S,?> endpoint;

public AbstractProtocol(AbstractEndpoint<S,?> endpoint) {
    this.endpoint = endpoint;
    setConnectionLinger(Constants.DEFAULT_CONNECTION_LINGER);
    setTcpNoDelay(Constants.DEFAULT_TCP_NO_DELAY);
}

public void setConnectionLinger(int connectionLinger) {
    endpoint.setConnectionLinger(connectionLinger);
}

public void setTcpNoDelay(boolean tcpNoDelay) {
    endpoint.setTcpNoDelay(tcpNoDelay);
}

在 Http11NioProtocol 构造方法里，创建了一个 NioEndpoint 对象，Http11Nio2Protocol 与 Http11NioProtocol 的区别主要在这里，Http11Nio2Protocol 在构造方法里创建的是 Nio2Endpoint 对象。这个 NioEndpoint 对象是非常重要的组件，它封装了 tomcat 的线程模型，后面会单独讲解这个类，这里先不多做描述。
在 AbstractProtocol 里把这个 NioEndpoint 对象复制给内部的 Endpoint 类型的属性，然后设置了一些 Endpoint 对象的两个属性，setConnectionLinger 和 setTcpNoDelay 方法就是调用 Endpoint 对象的相关方法。

在 Http11NioProtocol 父类的父类 AbstractHttp11Protocol 里，创建了一个 ConnectionHandler 对象，并调用 setHandler(cHandler) 把这个对象赋值给自己 handler 属性，这个属性在 AbstractHttp11Protocol 的父类 AbstractProtocol 里。

private Handler<S> handler;

protected void setHandler(Handler<S> handler) {
    this.handler = handler;
}

ConnectionHandler 是 AbstractProtocol 类的一个静态内部类，其声明为

rotected static class ConnectionHandler<S> implements AbstractEndpoint.Handler<S>

可以看出 Handler 类是 AbstractEndpoint 里的一个静态的接口类型，这个 Handler 接口定义了一些处理 Socket 事件的方法。
AbstractHttp11Protocol 构造方法里接着调用了 getEndpoint().setHandler(cHandler)，把 ConnectionHandler 对象赋值 NioEndpoint 的 Handler 类型的属性。

---

2. ProtocolHandler#init 方法
ProtocolHandler 定义了 init 和 start 方法，ProtocolHandler 的实现类 AbstractProtocol 及其子类 AbstractHttp11Protocol 实现或重载了 init 方法，AbstractHttp11Protocol 的子类都没有重载 init 方法。

2.1. AbstractHttp11Protocol#init

/**
 * The upgrade protocol instances configured.
 */
private final List<UpgradeProtocol> upgradeProtocols = new ArrayList<>();


@Override
public void init() throws Exception {
    // Upgrade protocols have to be configured first since the endpoint
    // init (triggered via super.init() below) uses this list to configure
    // the list of ALPN protocols to advertise
    for (UpgradeProtocol upgradeProtocol : upgradeProtocols) {
        configureUpgradeProtocol(upgradeProtocol);
    }

    super.init();
}

AbstractHttp11Protocol#init 方法比较简单，就是简单调用了 configureUpgradeProtocol 方法。

/**
 * The protocols that are available via internal Tomcat support for access
 * via HTTP upgrade.
 */
private final Map<String,UpgradeProtocol> httpUpgradeProtocols = new HashMap<>();
/**
 * The protocols that are available via internal Tomcat support for access
 * via ALPN negotiation.
 */
private final Map<String,UpgradeProtocol> negotiatedProtocols = new HashMap<>();
private void configureUpgradeProtocol(UpgradeProtocol upgradeProtocol) {
    // HTTP Upgrade
    String httpUpgradeName = upgradeProtocol.getHttpUpgradeName(getEndpoint().isSSLEnabled());
    boolean httpUpgradeConfigured = false;
    if (httpUpgradeName != null && httpUpgradeName.length() > 0) {
        httpUpgradeProtocols.put(httpUpgradeName, upgradeProtocol);
        httpUpgradeConfigured = true;
        getLog().info(sm.getString("abstractHttp11Protocol.httpUpgradeConfigured",
                getName(), httpUpgradeName));
    }


    // ALPN
    String alpnName = upgradeProtocol.getAlpnName();
    if (alpnName != null && alpnName.length() > 0) {
        if (getEndpoint().isAlpnSupported()) {
            negotiatedProtocols.put(alpnName, upgradeProtocol);
            getEndpoint().addNegotiatedProtocol(alpnName);
            getLog().info(sm.getString("abstractHttp11Protocol.alpnConfigured",
                    getName(), alpnName));
        } else {
            if (!httpUpgradeConfigured) {
                // ALPN is not supported by this connector and the upgrade
                // protocol implementation does not support standard HTTP
                // upgrade so there is no way available to enable support
                // for this protocol.
                getLog().error(sm.getString("abstractHttp11Protocol.alpnWithNoAlpn",
                        upgradeProtocol.getClass().getName(), alpnName, getName()));
            }
        }
    }
}

configureUpgradeProtocol 方法也挺简单的，就是将 UpgradeProtocol 放在 httpUpgradeProtocols 和 negotiatedProtocols 里。

2.2. AbstractProtocol#init

@Override
public void init() throws Exception {
    if (getLog().isInfoEnabled()) {
        getLog().info(sm.getString("abstractProtocolHandler.init", getName()));
        logPortOffset();
    }

    if (oname == null) {
        // Component not pre-registered so register it
        oname = createObjectName();
        if (oname != null) {
            Registry.getRegistry(null, null).registerComponent(this, oname, null);
        }
    }

    if (this.domain != null) {
        rgOname = new ObjectName(domain + ":type=GlobalRequestProcessor,name=" + getName());
        Registry.getRegistry(null, null).registerComponent(
                getHandler().getGlobal(), rgOname, null);
    }

    String endpointName = getName();
    endpoint.setName(endpointName.substring(1, endpointName.length()-1));
    endpoint.setDomain(domain);

    endpoint.init();
}

AbstractProtocol#init 里前面几行是注册一些对象到 MBeanServer 里，最重要的是最后一行的 endpoint.init()，这一行调用了 NioEndpoint 的 init 方法。关于 NioEndpoint 的详细内容将在后续的文章里讲解。

---

3. ProtocolHandler#start 方法
ProtocolHandler 的实现类 AbstractProtocol 实现了 start 方法，AbstractProtocol 的子类并没有重载 start 方法。

@Override
public void start() throws Exception {
    if (getLog().isInfoEnabled()) {
        getLog().info(sm.getString("abstractProtocolHandler.start", getName()));
        logPortOffset();
    }

    endpoint.start();
    monitorFuture = getUtilityExecutor().scheduleWithFixedDelay(
            new Runnable() {
                @Override
                public void run() {
                    if (!isPaused()) {
                        startAsyncTimeout();
                    }
                }
            }, 0, 60, TimeUnit.SECONDS);
}

在 AbstractProtocol#start 方法里先调用了 endpoint.start()，然后使用线程池来定期调用 startAsyncTimeout() 方法。
这里的 getUtilityExecutor() 返回的对象，是在 Connector 的 initInternal 中调用 ProtocolHandler#setUtilityExecutor 设值的，传入的对象 StandardServer 的 utilityExecutorWrapper，这个在之前介绍 Server 的文章里讲到过。
endpoint.start() 会在后续的文章里讲解。

protected void startAsyncTimeout() {
    if (asyncTimeoutFuture == null || (asyncTimeoutFuture != null && asyncTimeoutFuture.isDone())) {
        if (asyncTimeoutFuture != null && asyncTimeoutFuture.isDone()) {
            // There was an error executing the scheduled task, get it and log it
            try {
                asyncTimeoutFuture.get();
            } catch (InterruptedException | ExecutionException e) {
                getLog().error(sm.getString("abstractProtocolHandler.asyncTimeoutError"), e);
            }
        }
        asyncTimeoutFuture = getUtilityExecutor().scheduleAtFixedRate(
                new Runnable() {
                    @Override
                    public void run() {
                        long now = System.currentTimeMillis();
                        for (Processor processor : waitingProcessors) {
                            processor.timeoutAsync(now);
                        }
                    }
                }, 1, 1, TimeUnit.SECONDS);
    }
}

可以看出 startAsyncTimeout 方法的作用是定期调用 waitingProcessors 里的 Processor 对象的 timeoutAsync 方法来处理一些超时的请求。
Processor 也是 tomcat 用来处理请求的一个关键组件。上文中提到的 ConnectionHandler 就是使用 Processor 来具体处理请求的。Processor 将会在后续的文章中介绍。

请求到达Poller处理，最终是由Processor来进行处理，为了说明这中间过程所涉及的部分，先整理下在tomcat服务的各个组成部分：ProtocolHandler、Endpoint、Endpoint.Handler、Processor

它们之间的引用关系如下

[ProtocolHandler]     <---------   Connector

org.apache.coyote.ProtocolHandler
org.apache.coyote.AbstractProtocol
org.apache.coyote.http11.AbstractHttp11Protocol
org.apache.coyote.http11.AbstractHttp11JsseProtocol
org.apache.coyote.http11.Http11NioProtocol    *

[Endpoint]           <---------   ProtocolHandler(Http11NioProtocol)
 
org.apache.tomcat.util.net.AbstractEndpoint 
org.apache.tomcat.util.net.NioEndpoint      *

处理连接数的控制、连接的建立等工作，主要包括Acceptor和Poller两大部分，将建立好的连接交由Endpoint的Handler去处理

[Endpoint.Handler]     <---------   Endpoint(NioEndpoint)

     AbstractEndpoint.Handler
     AbstractConnectionHandler    < -- > AbstractProtocol
     Http11ConnectionHandler      < -- >  Http11NioProtocol    *

 缓存连接和Processor的关系，根据连接寻找Processor来处理

[Processor]      <---------   Endpoint.Handler

org.apache.coyote.Processor
org.apache.coyote.AbstractProcessor
org.apache.coyote.http11.AbstractHttp11Processor
org.apache.coyote.http11.Http11NioProcessor    *

（process方法）处理http请求中的相关业务，服务状态，协议解析，请求握手，内容解压等，构造适合Adapter处理request和response对象，然后调用Adater进行业务数据处理。

Acceptor和Poller都是属于Endpoint的内部组成部分，所以，这里是socket连接有数据到达时，被交到Endpoint.Handler中，经过缓存加速，找到对应的Processor之后，交由Processor来处理。

Processor处理完http协议相关的内容后，交由Adapter来处理业务。

Q&A

1、在什么时候读取请求数据？

     Java Servlet开发模型中，由开发者自定义的servlet来通过InputStream读取数据，通过OutputStream写入数据。所以，以上的这些部分，都不涉及到具体业务数据的读写。具体的数据读写时机，在Adapter之后。

2、请求如何被转到用户自定义的Servlet上的？

     具体的实现，在Adapter之后的部分，这里并不涉及与此相关的部分。

3、服务的连接数是如何控制的？

     Endpoint内部的Acceptor来控制，Endpoint提供了setMaxConnection方法来设置tomcat所能支持的最大连接数。此方法会设置一个Acceptor用的一个LimitLatch锁。通过LimitLatch来控制连接的数量。当有新连接加入时，会增加计算器；连接关闭时，会将计数器减一。

4、tomcat服务内在哪个地方使用了线程池？
   
     Endpoint内部的Poller用来选出有数据到达的socket连接，并将连接交给Endpoint.Handler来进行处理。这个传递事件被封装成了一个SocketProcessor任务，通过执行SocketProcessor任务来完成。
     SocketProcessor的执行时再线程池中实现的，也就是Endpoint的Poller线程，将事件转成SocketProcessor任务后，放在Endpoint的ThreadPool中完成。

小结
本文介绍了 ProtocolHandler 的初始化和启动，ProtocolHandler的默认实现类是 Http11NioProtocol。Http11NioProtocol 有一个非常重要的 NioEndpoint 对象，ProtocolHandler 的 init 和 start 方法中最关键的就是调用这个 NioEndpoint 对象的 init 和 start 方法。此外，在 AbstractHttp11Protocol 构造方法里创建了一个也是非常重要的 ConnectionHandler 对象，这个对象是用来处理请求，ConnectionHandler 使用 Processor 对象来具体处理请求。
原文链接：https://blog.csdn.net/yanlinwang/article/details/46537735

https://blog.csdn.net/yanlinwang/article/details/46537735?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromBaidu-9.nonecase&depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromBaidu-9.nonecase