tomcat請(qǐng)求流程源碼解進(jìn)階篇

更新時(shí)間：2023年09月01日 14:10:59 作者：湯卜

這篇文章主要為大家介紹了tomcat請(qǐng)求流程源碼解進(jìn)階,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

Connector的初始化

catalina解析server.xml是通過(guò)degister來(lái)實(shí)現(xiàn)的，degister解析到<Connector標(biāo)簽后做的事情如下代碼所見(jiàn)

ConnectorCreateRule
@Override
    public void begin(String namespace, String name, Attributes attributes)
            throws Exception {
        Service svc = (Service)digester.peek();
        Executor ex = null;
        if ( attributes.getValue("executor")!=null ) {
            ex = svc.getExecutor(attributes.getValue("executor"));
        }
        Connector con = new Connector(attributes.getValue("protocol"));
        if (ex != null) {
            setExecutor(con, ex);
        }
        String sslImplementationName = attributes.getValue("sslImplementationName");
        if (sslImplementationName != null) {
            setSSLImplementationName(con, sslImplementationName);
        }
        digester.push(con);
    }

connector根據(jù)標(biāo)簽屬性，拿到對(duì)應(yīng)的protocol協(xié)議，拿到配在service標(biāo)簽內(nèi)部的線程池，protocol的名稱(chēng)轉(zhuǎn)化成Connector中ProtocolHandler類(lèi)型的成員變量, 后續(xù)將以Http11NioProtocol來(lái)做講解

public Connector(String protocol) {
        setProtocol(protocol);
        // Instantiate protocol handler
        ProtocolHandler p = null;
        try {
            Class<?> clazz = Class.forName(protocolHandlerClassName);
            // 反射調(diào)用ProtocolHandler的構(gòu)造方法的時(shí)候會(huì)做后續(xù)的初始化
            p = (ProtocolHandler) clazz.getConstructor().newInstance();
        } catch (Exception e) {
            log.error(sm.getString("coyoteConnector.protocolHandlerInstantiationFailed"), e);
        } finally {
            this.protocolHandler = p;
        }

ProtocolHandler的構(gòu)造

ProtocolHandler有其抽象方法，Http11NioProtocol構(gòu)造方法中的tcp實(shí)現(xiàn)由NioEndpoint來(lái)做，因此Connector構(gòu)造起來(lái)的時(shí)候，對(duì)應(yīng)的ProtocolHanlder、endpoint的關(guān)聯(lián)關(guān)系已經(jīng)關(guān)聯(lián)好

public AbstractProtocol(AbstractEndpoint<S, ?> endpoint) {
        this.endpoint = endpoint;
        ConnectionHandler<S> cHandler = new ConnectionHandler<>(this);
        setHandler(cHandler);
        getEndpoint().setHandler(cHandler);
        setSoLinger(Constants.DEFAULT_CONNECTION_LINGER);
        setTcpNoDelay(Constants.DEFAULT_TCP_NO_DELAY);
    }

構(gòu)造后的init方法

在解析server.xml的基本層次結(jié)構(gòu)，成員變量填充完整后，需要調(diào)用生命周期的init方法

org/apache/catalina/startup/Catalina.load

getServer().init(); -----> connector.init(); --------------> protocolHandler.init();
----------> endpoint.init(); ------------->

看一下NioEndpoint的init方法

protected void initServerSocket() throws Exception {
        if (getUseInheritedChannel()) {
            // 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"));
            }
        } else {
//    綁定服務(wù)端的傳輸層端口
            serverSock = ServerSocketChannel.open();
            socketProperties.setProperties(serverSock.socket());
            InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset());
            serverSock.socket().bind(addr,getAcceptCount());
        }
        serverSock.configureBlocking(true); //mimic APR behavior
    }

建立Nio的Acceptor線程和Selector事件線程

org.apache.tomcat.util.net.NioEndpoint#startInternal
public void startInternal() throws Exception {
        if (!running) {
            running = true;
            paused = false;
            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());
            }
            // Create worker collection
            if (getExecutor() == null) {
            //   創(chuàng)建Tomcat默認(rèn)線程池，未手動(dòng)配置  線程數(shù)10 - 200
                createExecutor();
            }
            // 這個(gè)是tomcat的連接數(shù)限制器
            initializeConnectionLatch();
            // Start poller thread
            poller = new Poller();
            Thread pollerThread = new Thread(poller, getName() + "-Poller");
            pollerThread.setPriority(threadPriority);
            pollerThread.setDaemon(true);
            pollerThread.start();
            startAcceptorThread();
        }
    }

我們來(lái)看一下Poller的實(shí)現(xiàn)和Acctptor的實(shí)現(xiàn)

Poller的實(shí)現(xiàn)

public class Poller implements Runnable {
        private Selector selector;
        // PollerEvent          private NioSocketWrapper socketWrapper;   private int interestOps;
        // 向多路復(fù)用器注冊(cè)socket和需要處理的socket時(shí)間
        private final SynchronizedQueue<PollerEvent> events =
                new SynchronizedQueue<>();
            .........
        public Poller() throws IOException {
            this.selector = Selector.open();
        }
@Override
        public void run() {
            // Loop until destroy() is called
            while (true) {
                boolean hasEvents = false;
                try {
                    if (!close) {
                        hasEvents = events();
                        if (wakeupCounter.getAndSet(-1) > 0) {
                            // If we are here, means we have other stuff to do
                            // Do a non blocking select
                            keyCount = selector.selectNow();
                        } else {
                        // 默認(rèn)阻塞1秒鐘，監(jiān)聽(tīng)需要交給線程池的處理任務(wù)
                            keyCount = selector.select(selectorTimeout);
                        }
                        wakeupCounter.set(0);
                    }
                    ...
                } catch (Throwable x) {
                    ExceptionUtils.handleThrowable(x);
                    log.error(sm.getString("endpoint.nio.selectorLoopError"), x);
                    continue;
                }
                Iterator<SelectionKey> iterator =
                    keyCount > 0 ? selector.selectedKeys().iterator() : null;
                // Walk through the collection of ready keys and dispatch
                // any active event.
                while (iterator != null && iterator.hasNext()) {
                    SelectionKey sk = iterator.next();
                    iterator.remove();
                    NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
                    // Attachment may be null if another thread has called
                    // cancelledKey()
                    if (socketWrapper != null) {
                        // 主角， 任務(wù)處理都在里面
                        processKey(sk, socketWrapper);
                    }
                }
                // Process timeouts
                timeout(keyCount,hasEvents);
            }
.............
        }

Acctptor的實(shí)現(xiàn)

public class Acceptor<U> implements Runnable {
 @Override
    public void run() {
        int errorDelay = 0;
        long pauseStart = 0;
        try {
            // Loop until we receive a shutdown command
            while (!stopCalled) {
            ............
                if (stopCalled) {
                    break;
                }
                state = AcceptorState.RUNNING;
                try {
                    //if we have reached max connections, wait
                    endpoint.countUpOrAwaitConnection();
                    // Endpoint might have been paused while waiting for latch
                    // If that is the case, don't accept new connections
                    if (endpoint.isPaused()) {
                        continue;
                    }
                    U socket = null;
                    try {
                        // 等待新連接
                        socket = endpoint.serverSocketAccept();
                    } catch (Exception ioe) {
                        // We didn't get a socket
                        endpoint.countDownConnection();
                        if (endpoint.isRunning()) {
                            // Introduce delay if necessary
                            errorDelay = handleExceptionWithDelay(errorDelay);
                            // re-throw
                            throw ioe;
                        } else {
                            break;
                        }
                    }
                    // Successful accept, reset the error delay
                    errorDelay = 0;
                    // Configure the socket
                    if (!stopCalled && !endpoint.isPaused()) {
                        // 由endpoint來(lái)處理新連接、把新連接存在map里、向poller注冊(cè)PollerEndpoint
                        // 以后讀寫(xiě)事件就由Poller交給線程池來(lái)管理
                        if (!endpoint.setSocketOptions(socket)) {
                            endpoint.closeSocket(socket);
                        }
                    } else {
                        endpoint.destroySocket(socket);
                    }
                }
                ........
        } finally {
            stopLatch.countDown();
        }
        state = AcceptorState.ENDED;
    }
}

當(dāng)新連接接入時(shí)，會(huì)把新連接注冊(cè)到至底層為Selector的多路復(fù)用器上，Tomcat的Connector擁有了接受新連接和處理socket事件的能力

項(xiàng)目實(shí)戰(zhàn)

依舊使用mytomcat.war，里面有一個(gè)servlet FirstServlet,跟隨tomcat一起啟動(dòng)，容器啟動(dòng)，項(xiàng)目的部署暫且不討論
接下來(lái)我們看一下一個(gè)簡(jiǎn)單的Servlet如何在tomcat中來(lái)流轉(zhuǎn)

tomcat啟動(dòng)時(shí)， Acceptor將阻塞到 socket = endpoint.serverSocketAccept();這一行
在瀏覽器輸入http://localhost:8080/mytomcat/servlet1， Acceptor解除阻塞并且獲取到客戶端socket

來(lái)看一下endpoint.setSocketOptions(socket)做了啥

@Override
    protected boolean setSocketOptions(SocketChannel socket) {
       .........
            NioSocketWrapper newWrapper = new NioSocketWrapper(channel, this);
            channel.reset(socket, newWrapper);
            connections.put(socket, newWrapper);
            socketWrapper = newWrapper;
            // Set socket properties
            // Disable blocking, polling will be used
            socket.configureBlocking(false);
          socketProperties.setProperties(socket.socket());
        // 注冊(cè)新連接
            poller.register(socketWrapper);
            return true;
        ...

再看一下Poller的代碼監(jiān)聽(tīng)了新連接的OP_READ事件

public void register(final NioSocketWrapper socketWrapper) {
            socketWrapper.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
            PollerEvent pollerEvent = createPollerEvent(socketWrapper, OP_REGISTER);
            addEvent(pollerEvent);
        }

這樣，http://localhost:8080/mytomcat/servlet1的請(qǐng)求就會(huì)交給Poller處理，詳細(xì)代碼見(jiàn)Poller的run函數(shù)

public void run() {
            // Loop until destroy() is called
            while (true) {
                boolean hasEvents = false;
                try {
                    if (!close) {
                        hasEvents = events();
                        if (wakeupCounter.getAndSet(-1) > 0) {
                            // If we are here, means we have other stuff to do
                            // Do a non blocking select
                            keyCount = selector.selectNow();
                        } else {
                        // 默認(rèn)阻塞1秒鐘，監(jiān)聽(tīng)需要交給線程池的處理任務(wù)
                            keyCount = selector.select(selectorTimeout);
                        }
                        wakeupCounter.set(0);
                    }
                    ...
                } catch (Throwable x) {
                    ExceptionUtils.handleThrowable(x);
                    log.error(sm.getString("endpoint.nio.selectorLoopError"), x);
                    continue;
                }
                Iterator<SelectionKey> iterator =
                    keyCount > 0 ? selector.selectedKeys().iterator() : null;
                // Walk through the collection of ready keys and dispatch
                // any active event.
                while (iterator != null && iterator.hasNext()) {
                    SelectionKey sk = iterator.next();
                    iterator.remove();
                    NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
                    // Attachment may be null if another thread has called
                    // cancelledKey()
                    if (socketWrapper != null) {
                        // 主角， 任務(wù)處理都在里面
                        processKey(sk, socketWrapper);
                    }
                }
                // Process timeouts
                timeout(keyCount,hasEvents);
            }
.............
        }

processKey的時(shí)候向線程池提交一個(gè)SocketProcessor任務(wù)

多線程任務(wù)的執(zhí)行

經(jīng)過(guò)一系列變換，connecotr對(duì)應(yīng)的Handler為Http11Nio Protocol，將協(xié)議部分的處理交由Http11Processor

CoyoteAdapter

tomcat并非直接把請(qǐng)求封裝為HttpServletRequest對(duì)象和HttpServletResponse對(duì)象，本身connector支持多種協(xié)議，不只使用http協(xié)議。所以tomcat存在比較底層的org.apache.coyote.Request，也存在繼承HttpServletRequest的org.apache.catalina.connector.Request，CoyoteAdapter負(fù)責(zé)做二者的轉(zhuǎn)化并且把轉(zhuǎn)化后的HttpServletRequest對(duì)象和HttpServletResponse對(duì)象交給tomcat容器的Pipeline。
來(lái)看一下CoyoteAdapter的service方法

public void service(org.apache.coyote.Request req, org.apache.coyote.Response res) throws Exception {
        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);
            // 做兩種類(lèi)型request的轉(zhuǎn)換
            request.setResponse(response);
            response.setRequest(request);
        }
        try {
            // Parse and set Catalina and configuration specific
            // 下面有講解這個(gè)方法
            postParseSuccess = postParseRequest(req, request, res, response);
            if (postParseSuccess) {
                // check valves if we support async
                request.setAsyncSupported(connector.getService().getContainer().getPipeline().isAsyncSupported());
                // Calling the container
               //把HttpServletRequest 和HttpServletRequest對(duì)象交給tomcat的職責(zé)鏈Pipeline。
connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);
            }
            ........
    }

我們來(lái)看一下postParseRequest(req, resp)方法

protected boolean postParseRequest(org.apache.coyote.Request req, Request request, org.apache.coyote.Response res,
            Response response) throws IOException, ServletException {
           // 解析出是否要Https
        if (req.scheme().isNull()) {
            // Use connector scheme and secure configuration, (defaults to
            // "http" and false respectively)
            req.scheme().setString(connector.getScheme());
            request.setSecure(connector.getSecure());
        } else {
            // Use processor specified scheme to determine secure state
            request.setSecure(req.scheme().equals("https"));
        }
// 代理相關(guān)
String proxyName = connector.getProxyName();
        int proxyPort = connector.getProxyPort();
        if (proxyPort != 0) {
            req.setServerPort(proxyPort);
        } else if (req.getServerPort() == -1) {
            // Not explicitly set. Use default ports based on the scheme
            if (req.scheme().equals("https")) {
                req.setServerPort(443);
            } else {
                req.setServerPort(80);
            }
        }
        if (proxyName != null) {
            req.serverName().setString(proxyName);
        }
// 預(yù)檢方法請(qǐng)求
// Check for ping OPTIONS * request
        if (undecodedURI.equals("*")) {
            if (req.method().equalsIgnoreCase("OPTIONS")) {
                StringBuilder allow = new StringBuilder();
                allow.append("GET, HEAD, POST, PUT, DELETE, OPTIONS");
                // Trace if allowed
                if (connector.getAllowTrace()) {
                    allow.append(", TRACE");
                }
                res.setHeader("Allow", allow.toString());
                // Access log entry as processing won't reach AccessLogValve
                connector.getService().getContainer().logAccess(request, response, 0, true);
                return false;
            } else {
                response.sendError(400, sm.getString("coyoteAdapter.invalidURI"));
            }
        }
        ........
 while (mapRequired) {
            // This will map the the latest version by default
          // tomcat可以部署多個(gè)應(yīng)用，找到具體在某個(gè)Context、Host、填充到request對(duì)象中。 
connector.getService().getMapper().map(serverName, decodedURI, version, request.getMappingData());
     }
.......

探索一下tomcat的職責(zé)鏈

正式進(jìn)入Tomcat的職責(zé)鏈中，下面我們來(lái)看一下pipeline中有什么
因?yàn)橐延协h(huán)境，本文直接給出默認(rèn)的鏈裝結(jié)構(gòu)

StandardEngineValve----> AccessLogValve----> ErrorReportValve---- > StandardHostValve ----> StandardContextValve -----> StandardWrapperValve---> ApplicationFilterChain ---后續(xù)執(zhí)行指定的Servlet

排除容器本身的節(jié)點(diǎn)StandardEngineValve， StandardHostValve，StandardContextValve三個(gè)節(jié)點(diǎn)。 AccessLogValve，ErrorReportValve， StandardWrapperValve值得我們關(guān)注
AccessLogValve:使用來(lái)記錄訪問(wèn)請(qǐng)求的基本信息，記錄到access.log中等
ErrorReportValve:來(lái)處理后面的錯(cuò)誤，跳到失敗頁(yè)面等
StandardWrapperValve: 如下詳細(xì)解說(shuō)

@Override
    public void invoke(Request request, Response response) throws IOException, ServletException {
        ...........
        // 獲取對(duì)應(yīng)servlet對(duì)應(yīng)的Wrapper， servlet可能還未加載
        StandardWrapper wrapper = (StandardWrapper) getContainer();
        Servlet servlet = null;
        Context context = (Context) wrapper.getParent();
        ........
        try {
            if (!unavailable) {
            // 獲取真正的servlet， 如果未加載初始化，在加載初始化后返回
                servlet = wrapper.allocate();
            }
        } catch (UnavailableException e) {
            container.getLogger().error(sm.getString("standardWrapper.allocateException", wrapper.getName()), e);
            checkWrapperAvailable(response, wrapper);
        } 
        .................
// ApplicationFilterChain本身是servlet規(guī)范中的過(guò)濾器， 持有servelt對(duì)象，使用職責(zé)鏈調(diào)用完成后，再調(diào)用servlet， 至此，tomcat的請(qǐng)求流程到此，后續(xù)交給上層應(yīng)用程序來(lái)處理請(qǐng)求
        ApplicationFilterChain filterChain = ApplicationFilterFactory.createFilterChain(request, wrapper, servlet);
        .........
         filterChain.doFilter(request.getRequest(), response.getResponse());

至此，tomcat講請(qǐng)求處理完后遞交給上層應(yīng)用程序來(lái)處理，并返回結(jié)果，依次寫(xiě)回

本文到此結(jié)束，tomcat的源碼還有很多，如容器之間的關(guān)系，多類(lèi)的加載安全性，動(dòng)態(tài)部署等，有興趣的朋友可以繼續(xù)研究，更多關(guān)于tomcat請(qǐng)求流程的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: