Tomcat源码分析 （八）----- HTTP请求处理过程（一）

@override
public void init() throws exception {
    if (getlog().isinfoenabled()) {
        getlog().info(sm.getstring("abstractprotocolhandler.init", getname()));
    }

    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);
    }

    // 1. 设置endpoint的名字，默认为：http-nio-{port}
    string endpointname = getname();
    endpoint.setname(endpointname.substring(1, endpointname.length()-1));
    endpoint.setdomain(domain);
    
    // 2. 初始化endpoint
    endpoint.init();
}

public abstract void bind() throws exception;
public abstract void unbind() throws exception;
public abstract void startinternal() throws exception;
public abstract void stopinternal() throws exception;

public void init() throws exception {
    // 执行bind()方法
    if (bindoninit) {
        bind();
        bindstate = bindstate.bound_on_init;
    }
    if (this.domain != null) {
        // register endpoint (as threadpool - historical name)
        oname = new objectname(domain + ":type=threadpool,name=\"" + getname() + "\"");
        registry.getregistry(null, null).registercomponent(this, oname, null);

        objectname socketpropertiesoname = new objectname(domain +
                ":type=threadpool,name=\"" + getname() + "\",subtype=socketproperties");
        socketproperties.setobjectname(socketpropertiesoname);
        registry.getregistry(null, null).registercomponent(socketproperties, socketpropertiesoname, null);

        for (sslhostconfig sslhostconfig : findsslhostconfigs()) {
            registerjmx(sslhostconfig);
        }
    }
}

@override
public void bind() throws exception {

    if (!getuseinheritedchannel()) {
        serversock = serversocketchannel.open();
        socketproperties.setproperties(serversock.socket());
        inetsocketaddress addr = (getaddress()!=null?new inetsocketaddress(getaddress(),getport()):new inetsocketaddress(getport()));
        //绑定serversocket到指定的ip和端口
        serversock.socket().bind(addr,getacceptcount());
    } else {
        // retrieve the channel provided by the os
        channel ic = system.inheritedchannel();
        if (ic instanceof serversocketchannel) {
            serversock = (serversocketchannel) ic;
        }
        if (serversock == null) {
            throw new illegalargumentexception(sm.getstring("endpoint.init.bind.inherited"));
        }
    }

    serversock.configureblocking(true); //mimic apr behavior

    // 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;
    }
    setstoplatch(new countdownlatch(pollerthreadcount));

    // initialize ssl if needed
    initialisessl();

    selectorpool.open();
}

@override
protected void startinternal() throws lifecycleexception {

    // validate settings before starting
    if (getport() < 0) {
        throw new lifecycleexception(sm.getstring(
                "coyoteconnector.invalidport", integer.valueof(getport())));
    }

    setstate(lifecyclestate.starting);

    try {
        protocolhandler.start();
    } catch (exception e) {
        throw new lifecycleexception(
                sm.getstring("coyoteconnector.protocolhandlerstartfailed"), e);
    }
}

@override
public void start() throws exception {
    if (getlog().isinfoenabled()) {
        getlog().info(sm.getstring("abstractprotocolhandler.start", getname()));
    }

    // 1. 调用`endpoint.start()`方法
    endpoint.start();

    // start async timeout thread
    // 2. 开启异步超时线程，线程执行单元为`asynctimeout`
    asynctimeout = new asynctimeout();
    thread timeoutthread = new thread(asynctimeout, getnameinternal() + "-asynctimeout");
    int priority = endpoint.getthreadpriority();
    if (priority < thread.min_priority || priority > thread.max_priority) {
        priority = thread.norm_priority;
    }
    timeoutthread.setpriority(priority);
    timeoutthread.setdaemon(true);
    timeoutthread.start();
}

public final void start() throws exception {
    // 1. `bind()`已经在`init()`中分析过了
    if (bindstate == bindstate.unbound) {
        bind();
        bindstate = bindstate.bound_on_start;
    }
    startinternal();
}

@override
public void startinternal() throws exception {
    if (!running) {
        running = true;
        paused = false;

        processorcache = new synchronizedstack<>(synchronizedstack.default_size,
                socketproperties.getprocessorcache());
        eventcache = new synchronizedstack<>(synchronizedstack.default_size,
                        socketproperties.geteventcache());
        niochannels = new synchronizedstack<>(synchronizedstack.default_size,
                socketproperties.getbufferpool());

        // create worker collection
        // 2. 创建工作者线程池
        if ( getexecutor() == null ) {
            createexecutor();
        }
        
        // 3. 初始化连接latch，用于限制请求的并发量
        initializeconnectionlatch();

        // start poller threads
        // 4. 开启poller线程。poller用于对接受者线程生产的消息（或事件）进行处理，poller最终调用的是handler的代码
        pollers = new poller[getpollerthreadcount()];
        for (int i=0; i<pollers.length; i++) {
            pollers[i] = new poller();
            thread pollerthread = new thread(pollers[i], getname() + "-clientpoller-"+i);
            pollerthread.setpriority(threadpriority);
            pollerthread.setdaemon(true);
            pollerthread.start();
        }
        // 5. 开启acceptor线程
        startacceptorthreads();
    }
}

protected final void startacceptorthreads() {
    int count = getacceptorthreadcount();
    acceptors = new acceptor[count];

    for (int i = 0; i < count; i++) {
        acceptors[i] = createacceptor();
        string threadname = getname() + "-acceptor-" + i;
        acceptors[i].setthreadname(threadname);
        thread t = new thread(acceptors[i], threadname);
        t.setpriority(getacceptorthreadpriority());
        t.setdaemon(getdaemon());
        t.start();
    }
}

protected void processkey(selectionkey sk, niosocketwrapper attachment) {
    try {
        if ( close ) {
            cancelledkey(sk);
        } else if ( sk.isvalid() && attachment != null ) {
            if (sk.isreadable() || sk.iswritable() ) {
                if ( attachment.getsendfiledata() != null ) {
                    processsendfile(sk,attachment, false);
                } else {
                    unreg(sk, attachment, sk.readyops());
                    boolean closesocket = false;
                    // 1. 处理读事件，比如生成request对象
                    // read goes before write
                    if (sk.isreadable()) {
                        if (!processsocket(attachment, socketevent.open_read, true)) {
                            closesocket = true;
                        }
                    }
                    // 2. 处理写事件，比如将生成的response对象通过socket写回客户端
                    if (!closesocket && sk.iswritable()) {
                        if (!processsocket(attachment, socketevent.open_write, true)) {
                            closesocket = true;
                        }
                    }
                    if (closesocket) {
                        cancelledkey(sk);
                    }
                }
            }
        } else {
            //invalid key
            cancelledkey(sk);
        }
    } catch ( cancelledkeyexception ckx ) {
        cancelledkey(sk);
    } catch (throwable t) {
        exceptionutils.handlethrowable(t);
        log.error("",t);
    }
}

public boolean processsocket(socketwrapperbase<s> socketwrapper,
        socketevent event, boolean dispatch) {
    try {
        if (socketwrapper == null) {
            return false;
        }
        // 1. 从`processorcache`里面拿一个`processor`来处理socket，`processor`的实现为`socketprocessor`
        socketprocessorbase<s> sc = processorcache.pop();
        if (sc == null) {
            sc = createsocketprocessor(socketwrapper, event);
        } else {
            sc.reset(socketwrapper, event);
        }
        // 2. 将`processor`放到工作线程池中执行
        executor executor = getexecutor();
        if (dispatch && executor != null) {
            executor.execute(sc);
        } else {
            sc.run();
        }
    } catch (rejectedexecutionexception ree) {
        getlog().warn(sm.getstring("endpoint.executor.fail", socketwrapper) , ree);
        return false;
    } catch (throwable t) {
        exceptionutils.handlethrowable(t);
        // this means we got an oom or similar creating a thread, or that
        // the pool and its queue are full
        getlog().error(sm.getstring("endpoint.process.fail"), t);
        return false;
    }
    return true;
}

@override
protected socketprocessorbase<niochannel> createsocketprocessor(
        socketwrapperbase<niochannel> socketwrapper, socketevent event) {
    return new socketprocessor(socketwrapper, event);
}

/**
 * this class is the equivalent of the worker, but will simply use in an
 * external executor thread pool.
 */
protected class socketprocessor extends socketprocessorbase<niochannel> {

    public socketprocessor(socketwrapperbase<niochannel> socketwrapper, socketevent event) {
        super(socketwrapper, event);
    }

    @override
    protected void dorun() {
        niochannel socket = socketwrapper.getsocket();
        selectionkey key = socket.getiochannel().keyfor(socket.getpoller().getselector());

        try {
            int handshake = -1;

            try {
                if (key != null) {
                    if (socket.ishandshakecomplete()) {
                        // no tls handshaking required. let the handler
                        // process this socket / event combination.
                        handshake = 0;
                    } else if (event == socketevent.stop || event == socketevent.disconnect ||
                            event == socketevent.error) {
                        // unable to complete the tls handshake. treat it as
                        // if the handshake failed.
                        handshake = -1;
                    } else {
                        handshake = socket.handshake(key.isreadable(), key.iswritable());
                        // the handshake process reads/writes from/to the
                        // socket. status may therefore be open_write once
                        // the handshake completes. however, the handshake
                        // happens when the socket is opened so the status
                        // must always be open_read after it completes. it
                        // is ok to always set this as it is only used if
                        // the handshake completes.
                        event = socketevent.open_read;
                    }
                }
            } catch (ioexception x) {
                handshake = -1;
                if (log.isdebugenabled()) log.debug("error during ssl handshake",x);
            } catch (cancelledkeyexception ckx) {
                handshake = -1;
            }
            if (handshake == 0) {
                socketstate state = socketstate.open;
                // process the request from this socket
                // 将处理逻辑交给`handler`处理，当event为null时，则表明是一个`open_read`事件
                if (event == null) {
                    state = gethandler().process(socketwrapper, socketevent.open_read);
                } else {
                    state = gethandler().process(socketwrapper, event);
                }
                if (state == socketstate.closed) {
                    close(socket, key);
                }
            } else if (handshake == -1 ) {
                close(socket, key);
            } else if (handshake == selectionkey.op_read){
                socketwrapper.registerreadinterest();
            } else if (handshake == selectionkey.op_write){
                socketwrapper.registerwriteinterest();
            }
        } catch (cancelledkeyexception cx) {
            socket.getpoller().cancelledkey(key);
        } catch (virtualmachineerror vme) {
            exceptionutils.handlethrowable(vme);
        } catch (throwable t) {
            log.error("", t);
            socket.getpoller().cancelledkey(key);
        } finally {
            socketwrapper = null;
            event = null;
            //return to cache
            if (running && !paused) {
                processorcache.push(this);
            }
        }
    }
}

@override
public socketstate process(socketwrapperbase<s> wrapper, socketevent status) {
    try {
     
        if (processor == null) {
            processor = getprotocol().createprocessor();
            register(processor);
        }

        processor.setsslsupport(
                wrapper.getsslsupport(getprotocol().getclientcertprovider()));

        // associate the processor with the connection
        connections.put(socket, processor);

        socketstate state = socketstate.closed;
        do {
            // 关键的代码，终于找到你了
            state = processor.process(wrapper, status);

        } while ( state == socketstate.upgrading);
        return state;
    } 
    catch (throwable e) {
        exceptionutils.handlethrowable(e);
        // any other exception or error is odd. here we log it
        // with "error" level, so it will show up even on
        // less-than-verbose logs.
        getlog().error(sm.getstring("abstractconnectionhandler.error"), e);
    } finally {
        containerthreadmarker.clear();
    }

    // make sure socket/processor is removed from the list of current
    // connections
    connections.remove(socket);
    release(processor);
    return socketstate.closed;
}

protected http11processor createprocessor() {                          
    // 构建 http11processor
    http11processor processor = new http11processor(
            proto.getmaxhttpheadersize(), (jioendpoint)proto.endpoint, // 1. http header 的最大尺寸
            proto.getmaxtrailersize(),proto.getmaxextensionsize());
    processor.setadapter(proto.getadapter());
    // 2. 默认的 keepalive 情况下, 每个 socket 处理的最多的 请求次数
    processor.setmaxkeepaliverequests(proto.getmaxkeepaliverequests());
    // 3. 开启 keepalive 的 timeout
    processor.setkeepalivetimeout(proto.getkeepalivetimeout());      
    // 4. http 当遇到文件上传时的 默认超时时间 (300 * 1000)    
    processor.setconnectionuploadtimeout(
            proto.getconnectionuploadtimeout());                      
    processor.setdisableuploadtimeout(proto.getdisableuploadtimeout());
    // 5. 当 http 请求的 body size超过这个值时, 通过 gzip 进行压缩
    processor.setcompressionminsize(proto.getcompressionminsize());  
    // 6. http 请求是否开启 compression 处理    
    processor.setcompression(proto.getcompression());                  
    processor.setnocompressionuseragents(proto.getnocompressionuseragents());
    // 7. http body里面的内容是 "text/html,text/xml,text/plain" 才会进行 压缩处理
    processor.setcompressablemimetypes(proto.getcompressablemimetypes());
    processor.setrestricteduseragents(proto.getrestricteduseragents());
    // 8. socket 的 buffer, 默认 9000
    processor.setsocketbuffer(proto.getsocketbuffer());       
    // 9. 最大的 post 处理尺寸的大小 4 * 1000    
    processor.setmaxsavepostsize(proto.getmaxsavepostsize());          
    processor.setserver(proto.getserver());
    processor.setdisablekeepalivepercentage(
            proto.getdisablekeepalivepercentage());                    
    register(processor);                                               
    return processor;
}

public abstract class abstractprocessorlight implements processor {

    @override
    public socketstate process(socketwrapperbase<?> socketwrapper, socketevent status)
            throws ioexception {

        socketstate state = socketstate.closed;
        iterator<dispatchtype> dispatches = null;
        do {
            if (dispatches != null) {
                dispatchtype nextdispatch = dispatches.next();
                state = dispatch(nextdispatch.getsocketstatus());
            } else if (status == socketevent.disconnect) {
                // do nothing here, just wait for it to get recycled
            } else if (isasync() || isupgrade() || state == socketstate.async_end) {
                state = dispatch(status);
                if (state == socketstate.open) {
                    // there may be pipe-lined data to read. if the data isn't
                    // processed now, execution will exit this loop and call
                    // release() which will recycle the processor (and input
                    // buffer) deleting any pipe-lined data. to avoid this,
                    // process it now.
                    state = service(socketwrapper);
                }
            } else if (status == socketevent.open_write) {
                // extra write event likely after async, ignore
                state = socketstate.long;
            } else if (status == socketevent.open_read){
                // 调用`service()`方法
                state = service(socketwrapper);
            } else {
                // default to closing the socket if the socketevent passed in
                // is not consistent with the current state of the processor
                state = socketstate.closed;
            }

            if (getlog().isdebugenabled()) {
                getlog().debug("socket: [" + socketwrapper +
                        "], status in: [" + status +
                        "], state out: [" + state + "]");
            }

            if (state != socketstate.closed && isasync()) {
                state = asyncpostprocess();
                if (getlog().isdebugenabled()) {
                    getlog().debug("socket: [" + socketwrapper +
                            "], state after async post processing: [" + state + "]");
                }
            }

            if (dispatches == null || !dispatches.hasnext()) {
                // only returns non-null iterator if there are
                // dispatches to process.
                dispatches = getiteratorandcleardispatches();
            }
        } while (state == socketstate.async_end ||
                dispatches != null && state != socketstate.closed);

        return state;
    }
}

@override
public void service(org.apache.coyote.request req, org.apache.coyote.response res)
        throws exception {

    // 1. 根据coyote框架的request和response对象，生成connector的request和response对象（是httpservletrequest和httpservletresponse的封装）
    request request = (request) req.getnote(adapter_notes);
    response response = (response) res.getnote(adapter_notes);

    if (request == null) {
        // create objects
        request = connector.createrequest();
        request.setcoyoterequest(req);
        response = connector.createresponse();
        response.setcoyoteresponse(res);

        // link objects
        request.setresponse(response);
        response.setrequest(request);

        // set as notes
        req.setnote(adapter_notes, request);
        res.setnote(adapter_notes, response);

        // set query string encoding
        req.getparameters().setquerystringcharset(connector.geturicharset());
    }

    // 2. 补充header
    if (connector.getxpoweredby()) {
        response.addheader("x-powered-by", powered_by);
    }

    boolean async = false;
    boolean postparsesuccess = false;

    req.getrequestprocessor().setworkerthreadname(thread_name.get());

    try {
        // parse and set catalina and configuration specific
        // request parameters
        // 3. 解析请求，该方法会出现代理服务器、设置必要的header等操作
        // 用来处理请求映射 (获取 host, context, wrapper, uri 后面的参数的解析, sessionid )
        postparsesuccess = postparserequest(req, request, res, response);
        if (postparsesuccess) {
            //check valves if we support async
            request.setasyncsupported(
                    connector.getservice().getcontainer().getpipeline().isasyncsupported());
            // calling the container
            // 4. 真正进入容器的地方，调用engine容器下pipeline的阀门
            connector.getservice().getcontainer().getpipeline().getfirst().invoke(
                    request, response);
        }
        if (request.isasync()) {
            async = true;
            readlistener readlistener = req.getreadlistener();
            if (readlistener != null && request.isfinished()) {
                // possible the all data may have been read during service()
                // method so this needs to be checked here
                classloader oldcl = null;
                try {
                    oldcl = request.getcontext().bind(false, null);
                    if (req.sendalldatareadevent()) {
                        req.getreadlistener().onalldataread();
                    }
                } finally {
                    request.getcontext().unbind(false, oldcl);
                }
            }

            throwable throwable =
                    (throwable) request.getattribute(requestdispatcher.error_exception);

            // if an async request was started, is not going to end once
            // this container thread finishes and an error occurred, trigger
            // the async error process
            if (!request.isasynccompleting() && throwable != null) {
                request.getasynccontextinternal().seterrorstate(throwable, true);
            }
        } else {
            //5. 通过request.finishrequest 与 response.finishresponse(刷outputbuffer中的数据到浏览器) 来完成整个请求
            request.finishrequest();
            //将 org.apache.catalina.connector.response对应的 outputbuffer 中的数据 刷到 org.apache.coyote.response 对应的 internaloutputbuffer 中, 并且最终调用 socket对应的 outputstream 将数据刷出去( 这里会组装 http response 中的 header 与 body 里面的数据, 并且刷到远端 )
            response.finishresponse();
        }

    } catch (ioexception e) {
        // ignore
    } finally {
        atomicboolean error = new atomicboolean(false);
        res.action(actioncode.is_error, error);

        if (request.isasynccompleting() && error.get()) {
            // connection will be forcibly closed which will prevent
            // completion happening at the usual point. need to trigger
            // call to oncomplete() here.
            res.action(actioncode.async_post_process,  null);
            async = false;
        }

        // access log
        if (!async && postparsesuccess) {
            // log only if processing was invoked.
            // if postparserequest() failed, it has already logged it.
            context context = request.getcontext();
            // if the context is null, it is likely that the endpoint was
            // shutdown, this connection closed and the request recycled in
            // a different thread. that thread will have updated the access
            // log so it is ok not to update the access log here in that
            // case.
            if (context != null) {
                context.logaccess(request, response,
                        system.currenttimemillis() - req.getstarttime(), false);
            }
        }

        req.getrequestprocessor().setworkerthreadname(null);

        // recycle the wrapper request and response
        if (!async) {
            request.recycle();
            response.recycle();
        }
    }
}

protected boolean postparserequest(org.apache.coyote.request req, request request,
        org.apache.coyote.response res, response response) throws ioexception, servletexception {
    // 省略部分代码
    messagebytes decodeduri = req.decodeduri();

    if (undecodeduri.gettype() == messagebytes.t_bytes) {
        // copy the raw uri to the decodeduri
        decodeduri.duplicate(undecodeduri);

        // parse the path parameters. this will:
        //   - strip out the path parameters
        //   - convert the decodeduri to bytes
        parsepathparameters(req, request);

        // uri decoding
        // %xx decoding of the url
        try {
            req.geturldecoder().convert(decodeduri, false);
        } catch (ioexception ioe) {
            res.setstatus(400);
            res.setmessage("invalid uri: " + ioe.getmessage());
            connector.getservice().getcontainer().logaccess(
                    request, response, 0, true);
            return false;
        }
        // normalization
        if (!normalize(req.decodeduri())) {
            res.setstatus(400);
            res.setmessage("invalid uri");
            connector.getservice().getcontainer().logaccess(
                    request, response, 0, true);
            return false;
        }
        // character decoding
        converturi(decodeduri, request);
        // check that the uri is still normalized
        if (!checknormalize(req.decodeduri())) {
            res.setstatus(400);
            res.setmessage("invalid uri character encoding");
            connector.getservice().getcontainer().logaccess(
                    request, response, 0, true);
            return false;
        }
    } else {
        /* the uri is chars or string, and has been sent using an in-memory
            * protocol handler. the following assumptions are made:
            * - req.requesturi() has been set to the 'original' non-decoded,
            *   non-normalized uri
            * - req.decodeduri() has been set to the decoded, normalized form
            *   of req.requesturi()
            */
        decodeduri.tochars();
        // remove all path parameters; any needed path parameter should be set
        // using the request object rather than passing it in the url
        charchunk uricc = decodeduri.getcharchunk();
        int semicolon = uricc.indexof(';');
        if (semicolon > 0) {
            decodeduri.setchars
                (uricc.getbuffer(), uricc.getstart(), semicolon);
        }
    }

    // request mapping.
    messagebytes servername;
    if (connector.getuseipvhosts()) {
        servername = req.localname();
        if (servername.isnull()) {
            // well, they did ask for it
            res.action(actioncode.req_local_name_attribute, null);
        }
    } else {
        servername = req.servername();
    }

    // version for the second mapping loop and
    // context that we expect to get for that version
    string version = null;
    context versioncontext = null;
    boolean maprequired = true;

    while (maprequired) {
        // this will map the the latest version by default
        connector.getservice().getmapper().map(servername, decodeduri,
                version, request.getmappingdata());
        // 省略部分代码
    }
    // 省略部分代码
}