tomcat請(qǐng)求流程源碼解進(jìn)階篇
Connector的初始化
catalina解析server.xml是通過degister來實(shí)現(xiàn)的,degister解析到<Connector標(biāo)簽后做的事情如下代碼所見
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的名稱轉(zhuǎn)化成Connector中ProtocolHandler類型的成員變量, 后續(xù)將以Http11NioProtocol來做講解
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來做,因此Connector構(gòu)造起來的時(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(); } }
我們來看一下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)聽需要交給線程池的處理任務(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來處理新連接、把新連接存在map里、向poller注冊(cè)PollerEndpoint // 以后讀寫事件就由Poller交給線程池來管理 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)目的部署暫且不討論
接下來我們看一下一個(gè)簡(jiǎn)單的Servlet如何在tomcat中來流轉(zhuǎn)
tomcat啟動(dòng)時(shí), Acceptor將阻塞到 socket = endpoint.serverSocketAccept();這一行
在瀏覽器輸入http://localhost:8080/mytomcat/servlet1, Acceptor解除阻塞并且獲取到客戶端socket
來看一下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)聽了新連接的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ì)代碼見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)聽需要交給線程池的處理任務(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)過一系列變換,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。
來看一下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); // 做兩種類型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); } ........ }
我們來看一下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é)鏈中,下面我們來看一下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:使用來記錄訪問請(qǐng)求的基本信息,記錄到access.log中等
ErrorReportValve:來處理后面的錯(cuò)誤,跳到失敗頁面等
StandardWrapperValve: 如下詳細(xì)解說
@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ī)范中的過濾器, 持有servelt對(duì)象,使用職責(zé)鏈調(diào)用完成后,再調(diào)用servlet, 至此,tomcat的請(qǐng)求流程到此,后續(xù)交給上層應(yīng)用程序來處理請(qǐng)求 ApplicationFilterChain filterChain = ApplicationFilterFactory.createFilterChain(request, wrapper, servlet); ......... filterChain.doFilter(request.getRequest(), response.getResponse());
至此,tomcat講請(qǐng)求處理完后遞交給上層應(yīng)用程序來處理,并返回結(jié)果,依次寫回
本文到此結(jié)束,tomcat的源碼還有很多,如容器之間的關(guān)系,多類的加載安全性,動(dòng)態(tài)部署等,有興趣的朋友可以繼續(xù)研究,更多關(guān)于tomcat請(qǐng)求流程的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
Spring Boot 中使用 JSON Schema 校驗(yàn)復(fù)雜JSO
在數(shù)據(jù)交換領(lǐng)域,JSON Schema 以其強(qiáng)大的標(biāo)準(zhǔn)化能力,為定義和規(guī)范 JSON 數(shù)據(jù)的結(jié)構(gòu)與規(guī)則提供了有力支持,下面給大家介紹Spring Boot 中使用 JSON Schema 校驗(yàn)復(fù)雜JSON數(shù)據(jù)的過程,感興趣的朋友跟隨小編一起看看吧2024-08-08SpringBoot2整合activiti6環(huán)境搭建過程解析
這篇文章主要介紹了SpringBoot2整合activiti6環(huán)境搭建過程解析,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-11-11Spring思維導(dǎo)圖助你輕松學(xué)習(xí)Spring
這篇文章主要為大家詳細(xì)介紹了Spring思維導(dǎo)圖,幫助你輕松學(xué)習(xí)Spring的相關(guān)資料,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2018-03-03SpringBoot用JdbcTemplates訪問Mysql實(shí)例代碼
本篇文章主要介紹了SpringBoot用JdbcTemplates訪問Mysql實(shí)例代碼,非常具有實(shí)用價(jià)值,需要的朋友可以參考下2017-05-05Java concurrency集合之CopyOnWriteArraySet_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理
CopyOnWriteArraySet基于CopyOnWriteArrayList實(shí)現(xiàn),其唯一的不同是在add時(shí)調(diào)用的是CopyOnWriteArrayList的addIfAbsent(若沒有則增加)方法2017-06-06Java調(diào)用Zookeeper的實(shí)現(xiàn)步驟
本文主要介紹了Java調(diào)用Zookeeper的實(shí)現(xiàn)步驟,文中通過示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2021-08-08RocketMQ?Broker實(shí)現(xiàn)高可用高并發(fā)的消息中轉(zhuǎn)服務(wù)
RocketMQ消息代理(Broker)是一種高可用、高并發(fā)的消息中轉(zhuǎn)服務(wù),能夠接收并存儲(chǔ)生產(chǎn)者發(fā)送的消息,并將消息發(fā)送給消費(fèi)者。它具有多種消息存儲(chǔ)模式和消息傳遞模式,支持水平擴(kuò)展和故障轉(zhuǎn)移等特性,可以為分布式應(yīng)用提供可靠的消息傳遞服務(wù)2023-04-04