tomcat源码分析-Connector初始化与启动

    一个应用应用服务器的性能很大程度上取决于网络通信模块的实现，因而Connector模块对于tomcat来说是重中之重。从tomcat5开始，默认的连接器实现为Coyote实现(orag.apache.tomcat:tomcat-coyote:7.0.57)，本文基于coyote实现会回答如下两个问题：

•     一个http请求是怎么被tomcat监听到的，会有那些处理；
•     ajp协议干什么用的。

一、Connector配置

    通过对Container的初始化分析，我们很自然的会回过头来看conf/server.xml的配置，其中配置了2个Connector。 

<Connector port="8080" protocol="HTTP/1.1"
               connectionTimeout="20000"
               redirectPort="8443" />
<Connector port="8009" protocol="AJP/1.3" redirectPort="8443" />

为什么会有多个Connector呢？我们部署服务器的时候，通常会有2种方式：

    1 直接部署tomcat，在浏览器中请求http与tomcat直连
    2 部署一个nginx作反向代理，tomcat与nginx直连

这就是上面两种配置，通过协议protocol来区分。所以多个connector的好处是通过不同的协议，是tomcat服务器能够随着http的应用场景，服务器架构的升级而兼容起来。

    好的，现在配置了2个Connector，那么继续思考一下，Connector是通信过程，如果是你你会怎么设计？显然需要做3件事：

（1）监听端口，创建服务端与客户端的链接；
（2）获取到客户端请求的socket数据，并对Socket数据进行解析和包装成Http请求数据格式；
（3）将包装后的数据交给Container处理

通过源码来分析，Connector有两个属性：protocolHandler（协议）和adapter（适配器），其中protocolHandler完成的是步骤(1)(2)，adapter完成的是步骤(3)。

 

二、Connector初始化

    1. Connector构造函数

    在Connector的构造方法中，通过反射生成protocolHandler。

public Connector(String protocol) {
    setProtocol(protocol);
    // Instantiate protocol handler
    try {
        Class<?> clazz = Class.forName(protocolHandlerClassName);
        this.protocolHandler = (ProtocolHandler) clazz.newInstance();
    } catch (Exception e) {
        log.error(sm.getString(
                "coyoteConnector.protocolHandlerInstantiationFailed"), e);
    }
}

public void setProtocol(String protocol) {

        if (AprLifecycleListener.isAprAvailable()) {
            if ("HTTP/1.1".equals(protocol)) {
                setProtocolHandlerClassName
                    ("org.apache.coyote.http11.Http11AprProtocol");
            } else if ("AJP/1.3".equals(protocol)) {
                setProtocolHandlerClassName
                    ("org.apache.coyote.ajp.AjpAprProtocol");
            } else if (protocol != null) {
                setProtocolHandlerClassName(protocol);
            } else {
                setProtocolHandlerClassName
                    ("org.apache.coyote.http11.Http11AprProtocol");
            }
        } else {
            if ("HTTP/1.1".equals(protocol)) {
                setProtocolHandlerClassName
                    ("org.apache.coyote.http11.Http11Protocol");
            } else if ("AJP/1.3".equals(protocol)) {
                setProtocolHandlerClassName
                    ("org.apache.coyote.ajp.AjpProtocol");
            } else if (protocol != null) {
                setProtocolHandlerClassName(protocol);
            }
        }

    }

 协议的设置在conf/server.xml中配置，由setProtocol来赋值，tomcat提供了6种协议：

 

 

由上面6个类可知，对于http/ajp协议，tomcat均提供了三种运行模式，及BIO、NIO、APR，BIO即传统的blocking io，性能是最差的，APR是通过安装apr和native，从操作系统层面解决异步IO问题，能大幅提高性能，最省心也能较大幅度提高性能的是NIO，只需将Connector的protocol配成"org.apache.coyote.http11.Http11NioProtocol"即可。

    为了便于分析，这里只分析Http11Protocol。由类图可以看到，归根结底Http11Protocol是ProtocolHandler的实现，在Http11Protocol的构造方法中，对成员变量endpoint和cHandler进行初始化

public Http11Protocol() {
        endpoint = new JIoEndpoint();
        cHandler = new Http11ConnectionHandler(this);
        ((JIoEndpoint) endpoint).setHandler(cHandler);
        setSoLinger(Constants.DEFAULT_CONNECTION_LINGER);
        setSoTimeout(Constants.DEFAULT_CONNECTION_TIMEOUT);
        setTcpNoDelay(Constants.DEFAULT_TCP_NO_DELAY);
    }

 ，这两个很重要，在后面会讲到。

    继续到Connector代码中，由前面提到的tomcat启动过程知道，会调用Connector两个方法init和start。而端口的绑定和监听则分别在这两个方法中完成。

 

    2. Connector的init()方法

    调用的是Connector的initInternal()方法，主要做了3件事

protected void initInternal() throws LifecycleException {

        super.initInternal();

        // step 1. Initialize adapter
        adapter = new CoyoteAdapter(this);
        protocolHandler.setAdapter(adapter);

        // Make sure parseBodyMethodsSet has a default
        if( null == parseBodyMethodsSet ) {
            setParseBodyMethods(getParseBodyMethods());
        }

        if (protocolHandler.isAprRequired() &&
                !AprLifecycleListener.isAprAvailable()) {
            //
        }

        try {
            // step 2
            protocolHandler.init();
        } catch (Exception e) {
            //
        }

        // step 3 Initialize mapper listener
        mapperListener.init();
    }

步骤2中Http11Protocol的init方法，最终会调用到其父类AbstractProtocol的init方法，在这个方法里面对endpoint（Http11Protocol使用的是JIoEndPoint）进行了初始化。

public void init() throws Exception {
        // ...

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

        try {
            endpoint.init();
        } catch (Exception ex) {
            //
        }
    }

 endpoint.init()在AbstractEndpoint中，完成了对需要监听的端口的绑定。

public final void init() throws Exception {
        if (bindOnInit) {
            bind();
            bindState = BindState.BOUND_ON_INIT;
        }
    }

 在JIoEndpoint的bind()中完成了对端口的绑定。

 

对于Http11NioProtocol,在NioEndPoint的bind()方法中我们发现了selector,

@Override
    public void bind() throws Exception {

        serverSock = ServerSocketChannel.open();
        socketProperties.setProperties(serverSock.socket());
        InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
        serverSock.socket().bind(addr,getBacklog());
        serverSock.configureBlocking(true); //mimic APR behavior
        serverSock.socket().setSoTimeout(getSocketProperties().getSoTimeout());

        // Initialize thread count defaults for acceptor, poller
        if (acceptorThreadCount == 0) {
            // FIXME: Doesn't seem to work that well with multiple accept threads
            acceptorThreadCount = 1;
        }
        if (pollerThreadCount <= 0) {
            //minimum one poller thread
            pollerThreadCount = 1;
        }
        stopLatch = new CountDownLatch(pollerThreadCount);

        //...
        selectorPool.open();
    }

 而让我们震惊的是,coyote用了自己搞的一个NioSelectorPool,其中maxSelectors=200与普通IO下Acceptor的数量一样。另外，普通IO下用的处理请求的线程池的核心线程数量是10，max是200.当然，这些值可在配置文件里配。

public class NioSelectorPool {

    public NioSelectorPool() {
    }

    private static final Log log = LogFactory.getLog(NioSelectorPool.class);

    protected static final boolean SHARED =
        Boolean.valueOf(System.getProperty("org.apache.tomcat.util.net.NioSelectorShared", "true")).booleanValue();

    protected NioBlockingSelector blockingSelector;

    protected volatile Selector SHARED_SELECTOR;

    protected int maxSelectors = 200;
    protected long sharedSelectorTimeout = 30000;
    protected int maxSpareSelectors = -1;
    protected boolean enabled = true;
    protected AtomicInteger active = new AtomicInteger(0);
    protected AtomicInteger spare = new AtomicInteger(0);
    protected ConcurrentLinkedQueue<Selector> selectors =
        new ConcurrentLinkedQueue<Selector>();
    //...
}

 

 

三、Connector启动

    Connector的启动会调用start方法，在startInternal方法中，

protected void startInternal() throws LifecycleException {

        // Validate settings before starting
        if (getPort() < 0) {
            //
        }

        setState(LifecycleState.STARTING); // 发送STARTING事件

        try {
            protocolHandler.start(); // 启动端口监听
        } catch (Exception e) {
            //
        }

        mapperListener.start(); // 这个很重要，后面会讲到
    }

 可以看到，start相对init是调用了对应的start方法。其中，protocolHandler.start();即调用了Http11Protocol的start方法。最终调用了调用了JIoEndpoint的startInternal方法，初始化了处理连接请求的线程池（默认最大线程数200个），开启Acceptor线程接收请求。

public void startInternal() throws Exception {

        if (!running) {
            running = true;
            paused = false;

            // Create worker collection
            if (getExecutor() == null) {
                createExecutor();
            }

            initializeConnectionLatch();

            startAcceptorThreads();

            // Start async timeout thread
            Thread timeoutThread = new Thread(new AsyncTimeout(),
                    getName() + "-AsyncTimeout");
            timeoutThread.setPriority(threadPriority);
            timeoutThread.setDaemon(true);
            timeoutThread.start();
        }
    }

    

四、一个http请求在Connector中所经历的代码逻辑

    先来个粗犷的印象，

LifecycleBase implements Lifecycle
start() Lifecycle
  startInternal() LifecycleBase

------------------------------

Catalina#load()
  StandardServer#init()
    StandService[i]#init()
      Container[i]#init()
      Executor[i]#init()
      Connector[i]#init()

Connector#initInternal()
  adapter = new CoyoteAdapter(this)
  AbstractProtocol extends ProtocolHandler.setAdapter(adapter)
  protocolHandler.init()
  mapperListener.init()

AbstractProtocol implements ProtocolHandler#init()
  AbstractEndpoint#init()
    bind()

------------------------------

Catalina#start()
  if (getServer() == null) load();
  StandardServer#start()
    StandardService[i]#start()
      Container#start()
      Executor[i]#start()
      Connector[i]#start()
Connector#startInternal()
  AbstractProtocol implements ProtocolHandler#start()
    AbstractEndpoint#start()
  mapperListener.start()

JIoEndpoint#startInternal()
  createExecutor()
  initializeConnectionLatch()
  startAcceptorThreads()

AbstractEndpoint#startAcceptorThreads()
  acceptors[i] = createAcceptor()
  new Thread(acceptors[i]).start()

JIoEndpoint.Acceptor implements Runnable
  Socket socket = serverSocketFactory.acceptSocket(serverSocket)
  processSocket(socket)
    SocketWrapper<Socket> wrapper = new SocketWrapper<Socket>(socket)
    getExecutor().execute(new SocketProcessor(wrapper))

SocketProcessor implements Runnable
  state = handler.process(socket,status);

Http11Protocol ... extends AbstractProtocol<S>
   Http11ConnectionHandler extends AbstractConnectionHandler implements Handler
AbstractProtocol
   AbstractConnectionHandler implements AbstractEndpoint.Handler
      SocketState process(SocketWrapper<S> wrapper, SocketStatus status)
        state = processor.process(wrapper)

AbstractHttp11Processor
  SocketState process(SocketWrapper<S> socketWrapper)
    adapter.service(request, response)

 在上文中有分析Connector在启动的时候会监听端口。继续以JIoEndpoint为例，在其Accptor类中：

protected class Acceptor extends AbstractEndpoint.Acceptor {
    @Override
    public void run() {
        while (running) {
            //
            try {
                //当前连接数
                countUpOrAwaitConnection();
                Socket socket = null;
                try {
                    //取出队列中的连接请求
                    socket = serverSocketFactory.acceptSocket(serverSocket);
                } catch (IOException ioe) {
                    countDownConnection();
                }
                if (running && !paused && setSocketOptions(socket)) {
                    //处理请求
                    if (!processSocket(socket)) {
                        countDownConnection();
                        closeSocket(socket);
                    }
                } else {
                    countDownConnection();
                    // Close socket right away
                    closeSocket(socket);
                }
            }
            //
        }
    }
}

 在上面的代码中，socket = serverSocketFactory.acceptSocket(serverSocket);与客户端建立连接，将连接的socket交给processSocket(socket)来处理。在processSocket中，对socket进行包装一下交给线程池来处理：

protected boolean processSocket(Socket socket) {
    try {
        SocketWrapper<Socket> wrapper = new SocketWrapper<Socket>(socket);
        wrapper.setKeepAliveLeft(getMaxKeepAliveRequests());
        wrapper.setSecure(isSSLEnabled());
        //交给线程池处理连接
        getExecutor().execute(new SocketProcessor(wrapper));
    }
    //
    return true;
}

 线程池处理的任务SocketProccessor，通过代码分析：

protected class SocketProcessor implements Runnable {

    protected SocketWrapper<Socket> socket = null;
    protected SocketStatus status = null;

    @Override
    public void run() {
        boolean launch = false;
        synchronized (socket) {
            SocketState state = SocketState.OPEN;
            try {
                serverSocketFactory.handshake(socket.getSocket());
            }
            //
            if ((state != SocketState.CLOSED)) {
                //委派给Handler来处理
                if (status == null) {
                    state = handler.process(socket, SocketStatus.OPEN_READ);
                } else {
                    state = handler.process(socket,status);
                }
            }}}
            //
}

 即在SocketProcessor中，将Socket交给handler处理，这个handler就是在Http11Protocol的构造方法中赋值的Http11ConnectionHandler，在该类的父类process方法中通过请求的状态，来创建Http11Processor处理器进行相应的处理，切到Http11Proccessor的父类AbstractHttp11Proccessor中。

public SocketState process(SocketWrapper socketWrapper) {
    RequestInfo rp = request.getRequestProcessor();
    rp.setStage(org.apache.coyote.Constants.STAGE_PARSE);

    // Setting up the I/O
    setSocketWrapper(socketWrapper);
    getInputBuffer().init(socketWrapper, endpoint);
    getOutputBuffer().init(socketWrapper, endpoint);

    while (!getErrorState().isError() && keepAlive && !comet && !isAsync() &&
            upgradeInbound == null &&
            httpUpgradeHandler == null && !endpoint.isPaused()) {
        //
        if (!getErrorState().isError()) {
            // Setting up filters, and parse some request headers
            rp.setStage(org.apache.coyote.Constants.STAGE_PREPARE);
            try {
                //请求预处理
                prepareRequest();
            }
            //
        }
        //
        if (!getErrorState().isError()) {
            try {
                rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
                //交由适配器处理
                adapter.service(request, response);

                if(keepAlive && !getErrorState().isError() && (
                        response.getErrorException() != null ||
                                (!isAsync() &&
                                statusDropsConnection(response.getStatus())))) {
                    setErrorState(ErrorState.CLOSE_CLEAN, null);
                }
                setCometTimeouts(socketWrapper);
            }
        }
    }
    //
}

代码很长，删掉的部分比较多，在这个方法里，可以看到Request和Response的生成，从Socket中获取请求数据，keep-alive处理，数据包装等，最后交给了CoyoteAdapter的service方法。