阅读 75

Tomcat源码学习之Connector(二)

Connector

Connector 用于接收请求并将请求封装成Request 和Response 来具体处理,最底层是使用Socket 来进行连接的, Request 和Response 是按照HTTP 协议来封装的,所以Connector 同时实现了TCP/IP 协议和HTTP 协议, Request 和Response 封装完之后交给Container 进行处理,Container 就是Servlet 的容器, Container 处理完之后返回给Connector,最后Connector 使用Socket 将处理结果返回给客户端,这样整个请求就处理完了。

结构

Connector主要包含三个模块:Http11NioProtocol(Endpoint+processor), Mapper, CoyoteAdapter,http请求在Connector中的流程如下

Connector 中具体是用ProtocolHandler 来处理请求的,不同的ProtocolHandler 代表不同的连接类型,比如, Http11Protocol 使用的是普通Socket 来连接的, Http 11 NioProtocol 使用的是NioSocket 来连接的。

ProtocolHandler 里面有2 个非常重要的组件: Endpoint 、Processor。

Endpoint:用于处理底层Socket 的网络连接。
Processor:用于将Endpoint 接收到的Socket 封装成Request。

也就是说Endpoint用来实现TCP/IP 协议, Processor 用来实现HTTP 协议。

Endpoint 的抽象实现AbstractEndpoint 里面定义的Acceptor 和AsyncTimeout 两个内部类和一个Handler 接口。Acceptor 用于监昕请求, AsyncTimeout 用于检查异步request 的超时,Handler 用于处理接收到的Socket,在内部调用了Processor 进行处理。

Connector类

Connector 类本身的作用主要是在其创建时创建ProtocolHandler,然后在生命周期的相关方法中调用了ProtocolHandler 的相关生命周期方法。

Connector 的使用方法是通过Connector 标签配置在conf/server.xml 文件中,所以Connector 是在Catalina 的load 方法中根据conf/server.xml 配置文件创建Server对象时创建的。Connector 的生命周期方法是在Service 中调用的。如下:

<Service name="Catalina">
    <Connector port="8080" protocol="HTTP/1.1"
               connectionTimeout="20000"
               redirectPort="8443" />
    <Connector port="8009" protocol="AJP/1.3" redirectPort="8443" />
</Service>
复制代码

Connector的创建

Connector 的创建过程主要是初始化ProtocolHandler。server.xrnl 配置文件中Connector 标签的protocol 属性会设置到Connector 构造函数的参数中,它用于指定ProtocolHandler 的类型,Connector 的构造函数代码如下:

public Connector(String protocol) {
    boolean aprConnector = AprLifecycleListener.isAprAvailable() &&
            AprLifecycleListener.getUseAprConnector();
    if ("HTTP/1.1".equals(protocol) || protocol == null) {
        if (aprConnector) {
            protocolHandlerClassName = "org.apache.coyote.http11.Http11AprProtocol";
        } else {
            protocolHandlerClassName = "org.apache.coyote.http11.Http11NioProtocol";
        }
    } else if ("AJP/1.3".equals(protocol)) {
        if (aprConnector) {
            protocolHandlerClassName = "org.apache.coyote.ajp.AjpAprProtocol";
        } else {
            protocolHandlerClassName = "org.apache.coyote.ajp.AjpNioProtocol";
        }
    } else {
        protocolHandlerClassName = protocol;
    }
    // Instantiate protocol handler
    ProtocolHandler p = null;
    try {
        Class<?> clazz = Class.forName(protocolHandlerClassName);
        p = (ProtocolHandler) clazz.getConstructor().newInstance();
    } catch (Exception e) {
        log.error(sm.getString(
                "coyoteConnector.protocolHandlerInstantiationFailed"), e);
    } finally {
        this.protocolHandler = p;
    }
    // Default for Connector depends on this system property
    setThrowOnFailure(Boolean.getBoolean("org.apache.catalina.startup.EXIT_ON_INIT_FAILURE"));
}
复制代码

Apr 是Apache Portable Runtime 的缩写,是Apache 提供的一个运行时环境,如果要使用Apr 需要先安装,安装后Tomcat 可以自己检测出来。如果安装了Apr, 方法会根据配置的HTTP/1.1 属性对应地将protocolHandlerClassName 设置为org.apache.coyote.http11.Http11.AprProtocol ,如果没有安装Apr,会根据配置的HTTP/1.1 属性将protocoHandlerClassName设置为com..apache.coyote.http11.Http11NioProtocol,然后就会根据protocolHandlerClassName 来创建ProtocolHandler。

ProtocolHandler

ProtocolHandler 有一个抽象实现类AbstractProtocol, AbstractProtocol 下面分了三种类型: Ajp 、HTTP 和Spdy 。

Ajp是Apache JServ Protocol 的缩写, Apache 的定向包协议,主要用于与前端服务器(如Apache )进行通信,它是长连接,不需要每次通信都重新建立连接,这样就节省了开销; Spdy 协议Google开发的协议,作用类似HTTP ,比HTTP 效率高,不过这只是Google 制定的企业级协议,使用并不广泛,而且在HTTP/2 协议中已经包含了Spdy 所提供的优势,所以Spdy 协议平常很少使用,不过Tomcat 提供了支持。

以默认配置中的Http11NioProtocol为例来分析,它使用HTTP11协议, TCP 层使用NioSocket 来传输数据。构造方法如下:

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

可见在构造方法中创建了endpoint。

Endpoint

Endpoint 用于处理具体连接和传输数据, NioEndpoint 继承自org.apache.tomcat. util.net.AbstractEndpoint,在NioEndpoint 中新增了Poller 和SocketProcessor 内部类, NioEndpoint 中处理请求的具体流程如图:

其主要调用bind()方法进行初始化:

public void bind() throws Exception {
    initServerSocket();

    setStopLatch(new CountDownLatch(1));

    // Initialize SSL if needed
    initialiseSsl();

    selectorPool.open(getName());
}
复制代码
protected void initServerSocket() throws Exception {
    if (!getUseInheritedChannel()) {
    	// 初始化socket
        serverSock = ServerSocketChannel.open();
        socketProperties.setProperties(serverSock.socket());
        InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset());
        serverSock.socket().bind(addr,getAcceptCount());
    } else {
        // Retrieve the channel provided by the OS
        Channel ic = System.inheritedChannel();
        ...
    }
    serverSock.configureBlocking(true); //mimic APR behavior
}
复制代码

初始化主要是建立socket连接,接着看生命周期的startInternal方法:

public void startInternal() throws Exception {

	// 创建socketProcessor
	if (socketProperties.getProcessorCache() != 0) {
    processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
            socketProperties.getProcessorCache());
	}
	if (socketProperties.getEventCache() != 0) {
    	eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
            	socketProperties.getEventCache());
	}
	if (socketProperties.getBufferPool() != 0) {
    	nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
            	socketProperties.getBufferPool());
	}
    ...
    
    // 启动poller线程
    poller = new Poller();
    Thread pollerThread = new Thread(poller, getName() + "-ClientPoller");
    pollerThread.setPriority(threadPriority);
    pollerThread.setDaemon(true);
    pollerThread.start();
    startAcceptorThread();
    
}
复制代码

可以发现它创建了一个Poller以及启动Poller线程与Acceptor线程来处理请求。同时它还初始化了SocketProcessor,这是它的内部类,Poller收到请求后就会交由它处理。而SocketProcessor又会将请求传递到Handler,如下:

state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
复制代码

最后由handler将请求传递给processor。

Processor

Processor的结构如下:

当Handler建立好socket连接后,请求将会交由Processor处理,AbstractProcessor构造方法如下:

protected AbstractProcessor(Adapter adapter, Request coyoteRequest, Response coyoteResponse) {
    this.adapter = adapter;
    asyncStateMachine = new AsyncStateMachine(this);
    request = coyoteRequest;
    response = coyoteResponse;
    response.setHook(this);
    request.setResponse(response);
    request.setHook(this);
    userDataHelper = new UserDataHelper(getLog());
}
复制代码

之后它将调用Adapter 将请求传递到Container 中,最后对处理的结果进行了处理,如有没有启动异步处理、处理过程巾有没有抛出异常等。主要体现在Service方法中:

public SocketState service(SocketWrapperBase<?> socketWrapper)
    throws IOException {
    
    ...
    
    // Process the request in the adapter
    if (getErrorState().isIoAllowed()) {
    
        rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
        getAdapter().service(request, response);
        
     	...
    }
    
    // Finish the handling of the request
    ...
}
复制代码

Adaper

Adapter 只有一个实现类,那就是org.apache.catalina.connector 包下的CoyoteAdapter 类。

Processor 在其process 方法中会调用Adapter 的service 方法来处理请求, Adapter 的service 方法主要是调用Container 管道中的invoke方法来处理请求,在处理之前对Request和Response做了处理,将原来创建的org.apache.coyote 包下的Request 和Response 封装成了org.apache.catal ina.connector 的Request 和Response ,并在处理完成后判断再启动了Comet(长连接推模式)和是否启动了异步请求,并作出相应处理。调用Container 管道的相应代码片段如下:

connector.getService().getContainer().getPipeline().getFirst().invoke(        request, response);
复制代码

我们知道,tomcat有四个Container,采用了责任链的设计模式,每一个Container定义了一个Pipeline,每一个Pipeline又定义了多个Valve,代表需要处理的任务。Pipeline就像是每个容器的逻辑总线,在Pipeline上按照配置的顺序,加载各个Valve。通过Pipeline完成各个Valve之间的调用,各个Valve实现具体的应用逻辑。

有了责任链设计模式的概念,http请求由Connector转发至Container,在Container中的流程就比较清晰了,如下:

最终将Response返回给Connector完成一次http的请求。

Mapper

在Tomcat中,当一个请求到达时,该请求最终由哪个Servlet来处理是靠Mapper路由映射器完成的。Mapper由Service管理。

存储结构

MapElement

protected abstract static class MapElement<T> {

    public final String name;   // 名字

    public final T object;      // 对应的对象,如host, context, wrapper

}
复制代码

MappedHost

protected static final class MappedHost extends MapElement<Host> {

    // host包含的context列表,即MappedContext数组的包装
    public volatile ContextList contextList;
}
复制代码

MappedContext

protected static final class MappedContext extends MapElement<Void> {

    // 一个Context可能会对应许多个不同的版本的context,一般情况下是1个
    public volatile ContextVersion[] versions;

}
复制代码

其中ContextVersion包含了Context下的所有Servlet,有多种映射方式,如精确的map,通配符的map,扩展名的map,如下:

protected static final class ContextVersion extends MapElement<Context> {

    // 对wrapper的精确的map
    public MappedWrapper[] exactWrappers = new MappedWrapper[0];

    // 基于通配符的map
    public MappedWrapper[] wildcardWrappers = new MappedWrapper[0];

    // 基于扩展名的map
    public MappedWrapper[] extensionWrappers = new MappedWrapper[0];
}
复制代码

MappedWrapper

protected static class MappedWrapper extends MapElement<Wrapper> {
    public final boolean jspWildCard;
    public final boolean resourceOnly;
}
复制代码

Mapper类

public final class Mapper {

    // host数组,host里面又包括了context和wrapper数组
    volatile MappedHost[] hosts = new MappedHost[0];

    // 下面三个是添加host, context, wrapper的函数,都是同步的
    // 而且保证添加后是有序的
    public synchronized void addHost(String name, String[] aliases, Host host) { }

    public void addContextVersion(String hostName, Host host, String path,
            String version, Context context, String[] welcomeResources,
            WebResourceRoot resources, Collection<WrapperMappingInfo> wrappers) {

    }

    protected void addWrapper(ContextVersion context, String path,
            Wrapper wrapper, boolean jspWildCard, boolean resourceOnly) {

    }

    // 根据name,查找一个MapElement(host, context, 或者wrapper)
    private static final <T> int find(MapElement<T>[] map, CharChunk name,
            int start, int end) {

        // ...省略

        // 核心是二分法
        while (true) {
            i = (b + a) >>> 1;
            int result = compare(name, start, end, map[i].name);
            if (result == 1) {
                a = i;
            } else if (result == 0) {
                return i;
            } else {
                b = i;
            }
            if ((b - a) == 1) {
                int result2 = compare(name, start, end, map[b].name);
                if (result2 < 0) {
                    return a;
                } else {
                    return b;
                }
            }
        }
    }
}
复制代码

体现

下面介绍在一次http请求中,哪个环节调用了Mapper作路由映射,已经路由映射的过程。

http请求经过http11Processor解析之后,会调用CoyoteAdapterservice()函数转发给Container,我们来详细看一下这一步。

public void service(org.apache.coyote.Request req, org.apache.coyote.Response res)
        throws Exception {
    ...
    
    try {
        // 解析并配置请求参数
        postParseSuccess = postParseRequest(req, request, res, response);
        
        if (postParseSuccess) {
            //check valves if we support async
            request.setAsyncSupported(
                 connector.getService().getContainer().getPipeline().isAsyncSupported());
            // 调用container
            connector.getService().getContainer().getPipeline().getFirst().invoke(
                    request, response);
        }
    }catch{
    	....
    }
    
    ...
}
复制代码

我们进入postParseRequest方法:

protected boolean postParseRequest(org.apache.coyote.Request req, Request request,
        org.apache.coyote.Response res, Response response) throws IOException, ServletException {
    
    ...
    
    while (mapRequired) {
        // This will map the the latest version by default
        connector.getService().getMapper().map(serverName, decodedURI,
                version, request.getMappingData());
		
		...
    }
    
    ...
}
复制代码

CoyoteAdapter会获取Connector中的Service中的Mapper,然后调用map()方法。

public final class Mapper {

    public void map(MessageBytes host, MessageBytes uri, String version,
            MappingData mappingData) throws IOException {
        
        ...
        
        // 调用私有方法internalMap,传入host, uri, version, 结果将会保存在mappingData
        internalMap(host.getCharChunk(), uri.getCharChunk(), version,
                mappingData);
    }

    private final void internalMap(CharChunk host, CharChunk uri,
            String version, MappingData mappingData) throws IOException {

        ...

        // 查找映射的host
        MappedHost[] hosts = this.hosts;
        // 跟mapper的find方法一样,采用二分法查找
        MappedHost mappedHost = exactFindIgnoreCase(hosts, host);
        mappingData.host = mappedHost.object;
        
        ...
        
         // 查找映射的context
        contextVersion = exactFind(contextVersions, version);


        // 查找映射的wrapper
        if (!contextVersion.isPaused()) {
            internalMapWrapper(contextVersion, uri, mappingData);
        }
    }
}
复制代码
关注下面的标签,发现更多相似文章
评论