前言
前面第一篇介紹了Tomcat的整體架構(gòu)，能讓我們在宏觀上對Tomcat的運(yùn)轉(zhuǎn)流程有一個(gè)認(rèn)識，但原理畢竟只是理論，這篇博客就來從源碼入手，分析Tomcat的的初始化、啟動、各個(gè)組件的構(gòu)建、一次請求的分發(fā)、生命周期的管理等內(nèi)容。
從初始化開始
Tomcat啟動有三種方式，第一種是直接通過org.apache.catalina.startup.Bootstrap類的main方法，這也是最早出現(xiàn)的一種方式，需要自己去配置server.xml文件;第二種是通過內(nèi)嵌的方式,現(xiàn)在比較流行的SpringBoot也就是內(nèi)嵌的org.apache.catalina.startup.Tomcat類;第三種是使用腳本，這里對應(yīng)的是Tomcat根目錄的bin目錄下的startup.sh(Windows下面是startup.bat, 其它一樣)腳本文件開始的，像早期的web項(xiàng)目、ssm這些開發(fā)和部署都是使用的這種方式，這個(gè)腳本文件主要做的事情就是設(shè)置一些環(huán)境變量，然后會調(diào)用catalina.sh文件，這幾種方式做的都是同樣的事，會先獲取一些初始化參數(shù)。其實(shí)不管哪種方式，最終還是都會到Java類，走一樣的流程。Bootstrap類首先會調(diào)用init()方法，看下面的代碼
public void init() throws Exception {
......
Class<?> startupClass = catalinaLoader.loadClass("org.apache.catalina.startup.Catalina");
Object startupInstance = startupClass.getConstructor().newInstance();
String methodName = "setParentClassLoader";
Class<?> paramTypes[] = new Class[1];
paramTypes[0] = Class.forName("java.lang.ClassLoader");
Object paramValues[] = new Object[1];
paramValues[0] = sharedLoader;
Method method =
startupInstance.getClass().getMethod(methodName, paramTypes);
// 調(diào)用Catalina類的setParentClassLoader方法
method.invoke(startupInstance, paramValues);
// 把實(shí)例化的對象賦值給該變量，后面需要用到
catalinaDaemon = startupInstance;
}
Bootstrap類會會通過反射的方式調(diào)用Catalina類的的setParentClassLoader()方法，設(shè)置當(dāng)前Server的擴(kuò)展類加載器，接著看看Bootstrap類main方法接下來的代碼
// daemon就是Bootstrap類的實(shí)例
daemon.load(args);
daemon.start();
這就是在根據(jù)初始化的命令參數(shù)的進(jìn)行后續(xù)的操作, 首先會調(diào)用load()方法, 這個(gè)方法同樣是以反射方式調(diào)用Catalina類的load方法，這是個(gè)比較重要的方法，看看源碼
public void load() {
......
// Create and execute our Digester
Digester digester = createStartDigester();
InputSource inputSource = null;
InputStream inputStream = null;
File file = null;
try {
try {
// 讀取server.xml文件
file = configFile();
inputStream = new FileInputStream(file);
inputSource = new InputSource(file.toURI().toURL().toString());
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug(sm.getString("catalina.configFail", file), e);
}
}
if (inputStream == null) {
try {
inputStream = getClass().getClassLoader()
.getResourceAsStream(getConfigFile());
inputSource = new InputSource
(getClass().getClassLoader()
.getResource(getConfigFile()).toString());
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug(sm.getString("catalina.configFail",
getConfigFile()), e);
}
}
}
// server.xml為空則讀取server-embed.xml文件
if (inputStream == null) {
try {
inputStream = getClass().getClassLoader()
.getResourceAsStream("server-embed.xml");
inputSource = new InputSource
(getClass().getClassLoader()
.getResource("server-embed.xml").toString());
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug(sm.getString("catalina.configFail",
"server-embed.xml"), e);
}
}
}
......
try {
inputSource.setByteStream(inputStream);
digester.push(this);
digester.parse(inputSource);
} catch (SAXParseException spe) {
log.warn("Catalina.start using " + getConfigFile() + ": " +
spe.getMessage());
return;
} catch (Exception e) {
log.warn("Catalina.start using " + getConfigFile() + ": " , e);
return;
}
} finally {
if (inputStream != null) {
try {
inputStream.close();
} catch (IOException e) {
// Ignore
}
}
}
......
// Start the new server
try {
getServer().init();
} catch (LifecycleException e) {
if (Boolean.getBoolean("org.apache.catalina.startup.EXIT_ON_INIT_FAILURE")) {
throw new java.lang.Error(e);
} else {
log.error("Catalina.start", e);
}
}
......
}
看到第一行會調(diào)用createDigester()方法創(chuàng)建一個(gè)Digester對象的實(shí)例，那么這個(gè)對象是干嘛的呢？用來解析xml文件，將xml里面配置的屬性映射到對應(yīng)的Java對象，看看這個(gè)方法
protected Digester createStartDigester() {
// Initialize the digester
Digester digester = new Digester();
......
// Configure the actions we will be using
digester.addObjectCreate("Server",
"org.apache.catalina.core.StandardServer",
"className");
digester.addSetProperties("Server");
// 調(diào)用棧頂對象的setServer()方法注入一個(gè)實(shí)現(xiàn)了Server接口的StandardServer實(shí)現(xiàn)類
digester.addSetNext("Server",
"setServer",
"org.apache.catalina.Server");
digester.addObjectCreate("Server/Listener",
null, // MUST be specified in the element
"className");
digester.addSetProperties("Server/Listener");
digester.addSetNext("Server/Listener",
"addLifecycleListener",
"org.apache.catalina.LifecycleListener");
digester.addObjectCreate("Server/Service",
"org.apache.catalina.core.StandardService",
"className");
......
return digester;
}
這個(gè)方法首先直接實(shí)例化一個(gè)Digester對象, 接著主要是用來解析server.xml文件，該文件不存在時(shí)再獲取其它的配置文件，這里看源碼就很容易清楚了, 這個(gè)方法里面定義了一系列的解析規(guī)則，使用這些規(guī)則經(jīng)能把需要的屬性注入到對應(yīng)的Java對象實(shí)例，再回到load()方法，里面有這么一行
digester.push(this);
Digester對象內(nèi)部維護(hù)了一個(gè)棧，這行代碼就是push當(dāng)前Catalina對象到Digester內(nèi)部維護(hù)的棧的棧頂。以Server的注入為例，這里就是首先用digester.addObjectCreate()方法查看xml文件中Server標(biāo)簽中的className屬性是否為空,為空則實(shí)例化一個(gè)StandardServer對象，并把此對象push到棧頂，然后使用digester.addSetProperties()方法設(shè)置在配置文件中與它相關(guān)的屬性,最后再使用 digester.addSetNext()方法，拿到當(dāng)前棧頂元素的下一層的元素，在這里也就是Catalina對象,然后調(diào)用它的setServer()方法把當(dāng)前棧頂元素當(dāng)作參數(shù)傳入，這樣一來當(dāng)前Catalina的Server已經(jīng)實(shí)例好了。棧頂存放的始終是上一層的實(shí)例化對象, 解析到的當(dāng)前層級的xml標(biāo)簽如果還有下一層，繼續(xù)按照配置好的規(guī)則進(jìn)行解析，層層遞進(jìn)，用這種方式，就可以將Tomcat各個(gè)組件的一些屬性或它的子組件裝配好。再回到load()方法，解析完xml文件后最終會調(diào)用 getServer().init()方法，這個(gè)Server也就是前面實(shí)例化好的StandardServer，直接跟進(jìn)去看看它的init()方法,結(jié)果發(fā)現(xiàn)這個(gè)方法是在它的父類LifecycleBase實(shí)現(xiàn)的，而這個(gè)類看名字就知道是和生命周期有關(guān)的，所以引出了下面的內(nèi)容。
生命周期
其實(shí)在第一篇博客里面已經(jīng)簡單介紹過了，不過這里還要補(bǔ)充一些，前面說到LifecycleBase類，它是實(shí)現(xiàn)了Lifecycle接口，把生命周期的狀態(tài)的轉(zhuǎn)變與維護(hù)、事件的觸發(fā)以及監(jiān)聽器的添加與刪除等公共的邏輯放到這個(gè)類來做，子類就可以在某個(gè)生命周期的時(shí)候?qū)Ｗ?shí)現(xiàn)自己的邏輯，看看這張類圖
為了防止與接口里面的方法重名，所以LifecycleBase類往原來的生命周期方法后面加了internal，子類只要實(shí)現(xiàn)這些方法就可以了，直接來看看具體的代碼吧，還是接著前面的init()方法
public final synchronized void init() throws LifecycleException {
//狀態(tài)檢查,必須是NEW才能繼續(xù)初始化
if (!state.equals(LifecycleState.NEW)) {
invalidTransition(Lifecycle.BEFORE_INIT_EVENT);
}
try {
// 觸發(fā)INITIALIZING的事件監(jiān)聽器
setStateInternal(LifecycleState.INITIALIZING, null, false);
// 抽象方法，給子類實(shí)現(xiàn)
initInternal();
// 觸發(fā)INITIALIZED的事件監(jiān)聽器
setStateInternal(LifecycleState.INITIALIZED, null, false);
} catch (Throwable t) {
handleSubClassException(t, "lifecycleBase.initFail", toString());
}
}
這個(gè)方法邏輯比較簡單，主要邏輯都寫清楚了,setStateInternal()方法會拿到生命狀態(tài)對應(yīng)的事件，然后將事件通知到已經(jīng)注冊到該組件的監(jiān)聽器。initInternal()是抽象方法，給子類實(shí)現(xiàn)，其它生命周期的方法也都是類似的邏輯，其實(shí)這就是設(shè)計(jì)模式中的模板方法模式，把公共邏輯給抽象父類做了，具體的子類再去填充已經(jīng)規(guī)定好的模板，這里也就是xxinternal()方法。接著回到主線，既然該方法是給子類實(shí)現(xiàn)的，我們看到LifecycleBase的子類StandardServer的initInternal()方法
protected void initInternal() throws LifecycleException {
......
// 調(diào)用子組件Service的init()方法
for (int i = 0; i < services.length; i++) {
services[i].init();
}
}
核心功能就是初始化Service，進(jìn)入Service，而Service是我們從Socket連接到Web應(yīng)用真正的管家，管理著Connector和Container兩大核心組件。看看Service的initInternal()方法
protected void initInternal() throws LifecycleException {
super.initInternal();
// 調(diào)用engine的init()方法
if (engine != null) {
engine.init();
}
// 初始化線程池
for (Executor executor : findExecutors()) {
if (executor instanceof JmxEnabled) {
((JmxEnabled) executor).setDomain(getDomain());
}
executor.init();
}
// 初始化映射監(jiān)聽器 Engine->Host的映射
mapperListener.init();
// 初始化connector
synchronized (connectorsLock) {
for (Connector connector : connectors) {
connector.init();
}
}
}
確實(shí)是分別調(diào)用了上層容器(Engine)和連接器(Connector)的初始化方法，同時(shí)初始化了線程池和MapperListener監(jiān)聽器，這幾個(gè)玩意一個(gè)個(gè)來，首先是Engine，調(diào)用了engine的init()方法后進(jìn)入了容器的部分。容器的雖然也是從LifecycleBase繼承下來的，但因?yàn)槿萜骱推渌M件相比也有自己的一部分邏輯，且容器(Container)的子組件較多，所以Tomcat用了一個(gè)ContainerBase類來繼承了LifecycleBase，并實(shí)現(xiàn)了Container接口，簡單描述一下ContainerBase做了哪些事情
- 1.綁定父容器，獲取父容器相關(guān)的信息；管理子容器，實(shí)現(xiàn)了容器及其監(jiān)聽器獲取、添加、刪除，以及容器狀態(tài)的維護(hù)，數(shù)據(jù)銷毀與事件的發(fā)送等通用方法，
- 2.維護(hù)了容器的重要組成元素Pipeline，并提供了操作Valve的通用方法
- 3.后臺任務(wù)線程的管理，可以進(jìn)行周期性的檢查并重載相關(guān)的配置
還是借助類圖，才能有一個(gè)比較清晰的了解，如下
ContainerBase的initInternal()方法會實(shí)例化一個(gè)Executor線程池(默認(rèn)線程數(shù)量是1)，用來啟動、停止子容器。接著看到MapperListener是可以實(shí)現(xiàn)熱部署的，當(dāng)Web應(yīng)用的部署信息發(fā)生變化時(shí)，就會將對應(yīng)的事件通知到該監(jiān)聽器，然后監(jiān)聽器根據(jù)接收到的事件更新Mapper的信息，這里的init()方法是直接用的父類的，把當(dāng)前對象在JMBean里面進(jìn)行注冊，與主題無關(guān)就不多說了，下面來介紹一下線程池。
Tomcat內(nèi)部的線程池
Tomcat的線程池是繼承了jdk的線程池，并實(shí)現(xiàn)了Lifecycle接口，對應(yīng)的類是org.apache.catalina.core.StandardThreadExecutor，它的是內(nèi)部持有一個(gè)ThreadExecutor對象，真正的線程池相關(guān)的邏輯就是這個(gè)定制版的ThreadExecutor來做的，接下來看看org.apache.tomcat.util.threads.ThreadExecutor對比jdk的線程池做了哪些優(yōu)化呢？首先得搞清楚jdk的線程池執(zhí)行的流程是怎樣的，概括來說
- 1.任務(wù)數(shù)量還未達(dá)到coreSize個(gè)時(shí)，來一個(gè)任務(wù)就創(chuàng)建一個(gè)新線程
- 2.再來任務(wù)時(shí)，就把任務(wù)丟到隊(duì)列里面讓所有的線程去搶；如果隊(duì)列滿了就創(chuàng)建臨時(shí)線程
- 3.如果總線程數(shù)達(dá)到maximumPoolSize，則執(zhí)行拒絕策略
而Tomcat的線程池會在上述第3步(當(dāng)然Tomcat也可以配置線程池的核心線程預(yù)啟動)，達(dá)到最大數(shù)后不會直接執(zhí)行拒絕策略，會再嘗試添加任務(wù)到隊(duì)列中，看看代碼
public void execute(Runnable command, long timeout, TimeUnit unit) {
submittedCount.incrementAndGet();
try {
super.execute(command);
} catch (RejectedExecutionException rx) {
if (super.getQueue() instanceof TaskQueue) {
final TaskQueue queue = (TaskQueue)super.getQueue();
try {
// 繼續(xù)嘗試把任務(wù)放到隊(duì)列中
if (!queue.force(command, timeout, unit)) {
submittedCount.decrementAndGet();
// 如果添加任務(wù)失敗 則執(zhí)行拒絕策略
throw new RejectedExecutionException("Queue capacity is full.");
}
} catch (InterruptedException x) {
submittedCount.decrementAndGet();
throw new RejectedExecutionException(x);
}
} else {
submittedCount.decrementAndGet();
throw rx;
}
}
}
可以看到具體的實(shí)現(xiàn)方式是拋出拒絕異常進(jìn)行捕獲后，再次嘗試入隊(duì)失敗再手動拋出異常，這樣做的原因主要是可能再次添加任務(wù)時(shí)有任務(wù)消費(fèi)完了，就能繼續(xù)添加任務(wù)到隊(duì)列中了。這里注意還有submittedCount這么一個(gè)原子變量，用來統(tǒng)計(jì)已經(jīng)提交到任務(wù)隊(duì)列中但還未執(zhí)行的任務(wù)數(shù)量，搞這么一個(gè)變量的原因肯定是和任務(wù)隊(duì)列有關(guān)。首先看看jdk的隊(duì)列LinkedBlockingQueue，如果指定了隊(duì)列的大小還好，如果沒指定默認(rèn)值是Integer.MAX_VALUE，Integer的最大值，當(dāng)任務(wù)數(shù)量過大，核心線程遠(yuǎn)遠(yuǎn)處理不過來的時(shí)候，還不停的添加任務(wù)到隊(duì)列，并且只要沒到隊(duì)列的最大值就不會創(chuàng)建新的臨時(shí)線程來處理，最后導(dǎo)致的結(jié)果可能就是堆內(nèi)存溢出，之前任務(wù)過多時(shí)只是丟掉一些任務(wù)，現(xiàn)在是直接影響到應(yīng)用的正常使用... Tomcat線程池的任務(wù)隊(duì)列org.apache.tomcat.util.threads.TaskQueue就針對這一點(diǎn)進(jìn)行了優(yōu)化，具體實(shí)現(xiàn)還是繼承了LinkedBlockingQueue，只要在關(guān)鍵點(diǎn)做處理就可以，直接看入隊(duì)的代碼
// 這里的parent是ThreadExecutor
public boolean offer(Runnable o) {
// 為空就直接交給父類處理
if (parent==null) return super.offer(o);
//線程池的線程數(shù)量達(dá)到最大時(shí)也直接給父類處理
if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
//如果提交但未執(zhí)行的任務(wù)數(shù)量<當(dāng)前線程池的線程數(shù)量，說明還有空閑線程，或者馬上
// 有空閑線程了，也直接交給父類處理
if (parent.getSubmittedCount()<=(parent.getPoolSize())) return super.offer(o);
// 到這里就說明未執(zhí)行的任務(wù)大于當(dāng)前線程池的線程數(shù)量，如果線程數(shù)量未達(dá)到最大
// 就直接return false 此時(shí)線程池就會創(chuàng)建新的線程
if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
// 默認(rèn)直接丟給父類
return super.offer(o);
}
代碼的邏輯很清晰了，這樣一來就能很好的解決了沒達(dá)到任務(wù)隊(duì)列大小時(shí)無法創(chuàng)建新的線程來進(jìn)行處理的情況。對于jdk的線程池，現(xiàn)在推薦的做法都是建議進(jìn)行顯示的創(chuàng)建來指定相關(guān)的屬性，特別是對于任務(wù)隊(duì)列顯式的指定最大數(shù)量，這樣才能更好在源頭上避免問題的產(chǎn)生，因?yàn)榧词故悄軇?chuàng)建新的線程來進(jìn)行處理，但任務(wù)數(shù)量增加太快時(shí)還是可能出現(xiàn)處理不過來的情況。再回到剛開始說的，使用StandardThreadExecutor類，主要是方便了線程池與Tomcat的組件的生命周期的統(tǒng)一管理，它的init()也與MapperListener一樣，也直接跳過了，注意這里初始化時(shí)StandardThreadExecutor內(nèi)部的線程池和任務(wù)隊(duì)列并沒有實(shí)例化，而是在啟動時(shí)才實(shí)例化的，這里的Executor其實(shí)會通過反射的方式設(shè)置到Connector的ProtocolHandler，也就是前面的createDigester()里面配置了這個(gè)規(guī)則。接下來看看Connector的初始化方法主要做了哪些事
Connector的初始化
Connector作為管理連接、解析參數(shù)，構(gòu)建請求、響應(yīng)的對象，是理解Tomcat的工作流程極其重要的部分，第一篇說過Tomcat其實(shí)是一個(gè)應(yīng)用服務(wù)器+Servlet容器，而Servlet容器是為了遵循JavaWeb開發(fā)的規(guī)范，在其它語言或框架可能就不會管這什么Servlet規(guī)范了，對于任何服務(wù)器框架來說，連接的管理都是最核心的部分，分析的時(shí)候前提是對Tomcat的整體有個(gè)大概了解。直接看到Connector的initInternal()方法
protected void initInternal() throws LifecycleException {
// 初始化適配器
adapter = new CoyoteAdapter(this);
// 將適配器綁定到Connector內(nèi)部的protocolHandler
protocolHandler.setAdapter(adapter);
// 初始化需要解析請求體的請求類型，默認(rèn)有POST 加入到HashSet 方便后續(xù)的判斷
if (null == parseBodyMethodsSet) {
setParseBodyMethods(getParseBodyMethods());
}
......
try {
protocolHandler.init();
} catch (Exception e) {
throw new LifecycleException(
sm.getString("coyoteConnector.protocolHandlerInitializationFailed"), e);
}
}
這里的protocolHandler實(shí)際是在Connector實(shí)例化的時(shí)候在構(gòu)造方法里面利用反射構(gòu)建的，前面說過這個(gè)對象是對應(yīng)用層協(xié)議的抽象(不熟悉的看下第一篇連接器相關(guān)的小節(jié))，這里比較重要的是與一個(gè)Adapter實(shí)例進(jìn)行了綁定，而這個(gè)實(shí)例是用來實(shí)現(xiàn)從連接器到容器的請求、響應(yīng)對象適配工作的，接下來我們可以看到protocolHandler調(diào)用了它的init()方法，這里為了方便還是拿來前面的類圖
我們的目的是搞清楚工作流程，所以以Http11NioProtocolHandler-NioEndpoint這條線來摸清楚流程，繼續(xù)跟進(jìn)去發(fā)現(xiàn)最初的init()方法是在AbstractHttp11Protocol這個(gè)抽象父類，這個(gè)類的構(gòu)造方法如下
public AbstractHttp11Protocol(AbstractEndpoint<S,?> endpoint) {
super(endpoint);
// 設(shè)置連接超時(shí)
setConnectionTimeout(Constants.DEFAULT_CONNECTION_TIMEOUT);
// 實(shí)例化連接處理器
ConnectionHandler<S> cHandler = new ConnectionHandler<>(this);
// 分別與ProtocolHandler的實(shí)現(xiàn)類 以及 Endpoint進(jìn)行綁定
setHandler(cHandler);
getEndpoint().setHandler(cHandler);
}
這里面會調(diào)用父類的構(gòu)造方法，把Endpoint傳進(jìn)去，這是我們處理連接和讀寫事件干事的地方，稍后在詳細(xì)描述，看到接下來幾行代碼會實(shí)例化ConnectorHandler，并且與當(dāng)前Endpoint進(jìn)行綁定，這個(gè)ConnectHandler主要的職責(zé)是用來管理SocketProcessor，后面在源碼中會看到的?；氐街骶€，AbstractHttp11Protocol的init()方法首先會進(jìn)行協(xié)議升級相關(guān)的配置，這個(gè)我們不用管，接著會調(diào)用父類AbstractProtocol的init()方法，看看它的代碼
public void init() throws Exception {
...
String endpointName = getName();
endpoint.setName(endpointName.substring(1, endpointName.length()-1));
endpoint.setDomain(domain);
// 調(diào)用endpoint的init方法
endpoint.init();
}
這里邏輯也比較簡單，所以我們繼續(xù)跟進(jìn)到NioEndPoint的init()方法，這里還是把類圖拿過來
同樣它的init()方法也是在最終的抽象父類AbstractEndpoint，主要有這么一段邏輯
if (bindOnInit) {
bindWithCleanup();
bindState = BindState.BOUND_ON_INIT;
}
bindOnInit指在是否初始化時(shí)進(jìn)行端口的綁定，默認(rèn)是true，它會調(diào)用bindWithCleanup()方法
private void bindWithCleanup() throws Exception {
try {
bind();
} catch (Throwable t) {
// Ensure open sockets etc. are cleaned up if something goes
// wrong during bind
ExceptionUtils.handleThrowable(t);
unbind();
throw t;
}
}
這個(gè)bind()也是一個(gè)抽象模板方法，目的是給子類實(shí)現(xiàn)的，所以我們的init()方法最終到了NioEndpoint的bind()方法，上代碼
public void bind() throws Exception {
//初始化服務(wù)器端Socket
initServerSocket();
//初始化acceptor線程數(shù)量
if (acceptorThreadCount == 0) {
acceptorThreadCount = 1;
}
// 初始化poller線程數(shù)量
if (pollerThreadCount <= 0) {
pollerThreadCount = 1;
}
// 設(shè)置未停止的輪訓(xùn)器數(shù)量
setStopLatch(new CountDownLatch(pollerThreadCount));
// 如果需要ssl就進(jìn)行初始化的配置
initialiseSsl();
// 初始化selectorPool
selectorPool.open();
}
首先看到initServerSocket()方法會先初始化ServerSocket，也就是如下幾行代碼
serverSock = ServerSocketChannel.open();
socketProperties.setProperties(serverSock.socket());
InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset()); serverSock.socket().bind(addr,getAcceptCount());
serverSock.configureBlocking(true);
需要清楚的是對于ServerSocketChannel，Tomcat直接使用的阻塞的方式來監(jiān)聽連接，也無需注冊到Selector，確實(shí)意義不大，反正建立連接后就不歸它管了。接著會初始化Acceptor和Poller的線程數(shù)量，在第一篇里面介紹過，Acceptor是用來接收新的連接的線程，建立連接后就把連接通道Channel交給眾多Poller中的一個(gè)；Poller是用來檢測已經(jīng)建立好的連接的IO事件，對應(yīng)的也就是jdk的selector(多路復(fù)用器)，具體的代碼還在后面，繼續(xù)看到bind()方法的最后一行，會調(diào)用selectorPool的open()方法，那么這個(gè)selectorPool又是個(gè)啥東西呢？selectorPool是org.apache.tomcat.util.net.NioSelectorPool的一個(gè)實(shí)例對象，它的職責(zé)其實(shí)是作為一個(gè)輔助Selector，而前面說到的Poller一般稱被稱為主Selector，后續(xù)在進(jìn)行具體分析。
正式啟動
前面從講了從init()方法開始主要做的事情，接下來就看看Catalina的start()方法，上代碼
public void start() {
......
// Start the new server
getServer().start();
// Register shutdown hook
if (useShutdownHook) {
if (shutdownHook == null) {
shutdownHook = new CatalinaShutdownHook();
}
Runtime.getRuntime().addShutdownHook(shutdownHook);
}
if (await) {
await();
stop();
}
}
代碼精簡后，這里首先調(diào)用了getServer().start()，這里先放一放，接著看看后面的代碼，首先判斷是否useshutdownHook，默認(rèn)為true。接著往Runtime里面添加了了一個(gè)CatalinaShutdownHook，其實(shí)也就是一個(gè)Thread對象實(shí)例，它的里面就是停止和進(jìn)行資源回收的邏輯，這是挺實(shí)用的一個(gè)東西，也就是為了防止應(yīng)用異常停止的時(shí)候資源得不到正確回收，以前寫Java的gui程序的時(shí)候就很容易出現(xiàn)明明調(diào)用了System.exit()，但進(jìn)程還是運(yùn)行著，注冊到Runtime，也就是當(dāng)前應(yīng)用的運(yùn)行環(huán)境后，應(yīng)用停止時(shí)會自動調(diào)用它里面的邏輯，這一點(diǎn)是由JVM來進(jìn)行保證的，JVM的進(jìn)程在完全退出前會自動去執(zhí)行所有注冊了的ShutdownHook的邏輯；此外，后面await()方法還通過啟動一個(gè)Socket服務(wù)端，來監(jiān)聽SHUTDOWN消息，驗(yàn)證通過后進(jìn)入stop流程。接著直接跳到StandardServer的start()方法，前面初始化時(shí)說過，這些生命周期相關(guān)的方法基本的邏輯都交給父類去處理了,子類直接進(jìn)行xxInternal()方法進(jìn)行自己的邏輯，所以我們直接看到startInternal()方法
protected void startInternal() throws LifecycleException {
// 通知已注冊的監(jiān)聽器
fireLifecycleEvent(CONFIGURE_START_EVENT, null);
setState(LifecycleState.STARTING);
// 啟動全局命名服務(wù)
globalNamingResources.start();
// 啟動Service
synchronized (servicesLock) {
for (int i = 0; i < services.length; i++) {
services[i].start();
}
}
}
邏輯很簡單，首先通知已注冊的監(jiān)聽器，包括正常對生命周期進(jìn)行監(jiān)控的，以及需要加載某些配置的。接著會遍歷所有已經(jīng)添加了的Service，并調(diào)用它們的start()方法，這里也直接看到StandardService的startInternal()方法
protected void startInternal() throws LifecycleException {
// 通知監(jiān)聽器當(dāng)前正在啟動
setState(LifecycleState.STARTING);
// 啟動Service層級下的首個(gè)容器Engine
if (engine != null) {
synchronized (engine) {
engine.start();
}
}
// 啟動Executor，此時(shí)線程池與任務(wù)隊(duì)列才實(shí)例化
synchronized (executors) {
for (Executor executor: executors) {
executor.start();
}
}
// 啟動Mapper監(jiān)聽器
mapperListener.start();
// 啟動所有的連接器
synchronized (connectorsLock) {
for (Connector connector: connectors) {
// If it has already failed, don't try and start it
if (connector.getState() != LifecycleState.FAILED) {
connector.start();
}
}
}
容器的啟動
這里首先會啟動Service的首個(gè)子容器Engine，我們繼續(xù)跟進(jìn)StandardEngine的startInternal()方法，不過它除了打印日志以外沒做任何事，是直接調(diào)用了父類ContainerBase的startInternal()方法，直接看代碼
protected synchronized void startInternal() throws LifecycleException {
// 找到所有的子容器 并進(jìn)行啟動
Container children[] = findChildren();
List<Future<Void>> results = new ArrayList<>();
for (int i = 0; i < children.length; i++) {
results.add(startStopExecutor.submit(new StartChild(children[i])));
}
for (Future<Void> result : results) {
result.get();
}
// 調(diào)用管道(Pipeline)中所有閥門(Valve)的start()方法
if (pipeline instanceof Lifecycle) {
((Lifecycle) pipeline).start();
}
setState(LifecycleState.STARTING);
//啟動周期性任務(wù)執(zhí)行的線程
threadStart();
}
這部分是所有容器的公共邏輯，代碼也比較簡單，首先找到所有的子容器并進(jìn)行啟動，只是這里使用的是Future來提交任務(wù)的，與Runnable相比，沒多大區(qū)別，F(xiàn)uture是方便設(shè)置任務(wù)執(zhí)行的結(jié)果的，然后通過get()來獲取，不過這里使用Future也沒看到有什么作用，返回值也是Void，估計(jì)是為了后期的擴(kuò)展?？吹较旅娴?，接著會調(diào)用管道的start()方法，Tomcat的標(biāo)準(zhǔn)實(shí)現(xiàn)類是StandardPipeline，它的start()方法會去遍歷Pipeline內(nèi)所有的Valve，并調(diào)用它們start()方法，接著看到最后一行threadStart()，它會啟動一個(gè)線程執(zhí)行周期性的任務(wù)，對應(yīng)的Runnable就是ContainerBackgroundProcessor，看看它的run方法
while (!threadDone) {
try {
Thread.sleep(backgroundProcessorDelay * 1000L);
} catch (InterruptedException e) {
// Ignore
}
if (!threadDone) {
processChildren(ContainerBase.this);
}
}
只要線程沒結(jié)束，就不停的執(zhí)行周期性的任務(wù)，接著會調(diào)用processChildren()，看看代碼
protected void processChildren(Container container) {
// 調(diào)用容器的任務(wù)處理方法
container.backgroundProcess();
Container[] children = container.findChildren();
for (int i = 0; i < children.length; i++) {
if (children[i].getBackgroundProcessorDelay() <= 0) {
// 遞歸調(diào)用所以子容器
processChildren(children[i]);
}
}
}
這些主要就是調(diào)用了容器的backgroundProcess()，并遞歸的所有對所有子容器進(jìn)行同樣的調(diào)用，所以可以知道了只要我們在容器里面重寫backgroundProcess()方法，就可以做一些周期性執(zhí)行的任務(wù)，并且，例如資源的熱加載，session對象有效期的管理等。
線程池與Mapper的啟動
還是按照StandardService里面代碼的順序，這里線程池對應(yīng)的管理者是StandardExecutor，它的啟動方法只是把內(nèi)部真正的ThreadExectuor和TaskQueue實(shí)例化，接著來看看MapperListener的startInternal()方法
public void startInternal() throws LifecycleException {
// 通知監(jiān)聽器狀態(tài)改變
setState(LifecycleState.STARTING);
// Engine為空直接返回
Engine engine = service.getContainer();
if (engine == null) {
return;
}
// 找到默認(rèn)的Host
findDefaultHost();
// 遞歸的給所有的子容器添加監(jiān)聽器
addListeners(engine);
// 處理子容器，綁定提前配置好的映射
Container[] conHosts = engine.findChildren();
for (Container conHost : conHosts) {
Host host = (Host) conHost;
if (!LifecycleState.NEW.equals(host.getState())) {
// Registering the host will register the context and wrappers
registerHost(host);
}
}
}
Mapper是比較重要的一個(gè)組件，看看上面的代碼，findDefaultHost()首先配置默認(rèn)的主機(jī)名，也就是，默認(rèn)情況下也就是localhost，還是看看圖
這里就很清楚容器的各層組件表示的具體對象了，實(shí)際上這是對于常規(guī)的Servlet項(xiàng)目而言，對于平時(shí)部署Spring的項(xiàng)目而言，Context和Wrapper其實(shí)都是/，把請求和響應(yīng)對象包裝好后在DispatcherServlet的內(nèi)部再對uri處理，最后會找到Controller里面的映射路徑對應(yīng)的方法?；氐缴厦娴拇a，接著addListeners()方法會遞歸的給每個(gè)子容器都添加了MapperListener監(jiān)聽器，所以可以知道子容器的Mapper都是共享的同一個(gè)，繼續(xù)接下來的代碼就是配置默認(rèn)的每層組件到子容器的映射。接下來就看連接器的啟動了
Connector的啟動
Connector的start()方法，里面也是調(diào)用了ProtocolHandler的start()方法，所以直接看到它的子類AbstractProtocolHandler的啟動方法
public void start() throws Exception {
// 啟動Endpoint
endpoint.start();
// 開啟異步超時(shí)線程
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();
}
它的啟動方法主要是做兩件事情，調(diào)用Endpoint的start()，稍后進(jìn)行描述；另外一件事情是開啟了檢測異步Servlet超時(shí)的線程，簡單說一下，跟異步Servlet相關(guān)的processor會放到一個(gè)waitingProcessors集合中，因?yàn)槭褂卯惒降臅r(shí)候，任務(wù)的具體處理交給了Tomcat外部的線程，但連接是由Tomcat進(jìn)行的管理，所以必須保證它的讀寫有一個(gè)時(shí)間限制，不然里連接就可能會一直占用著，由于篇幅有限，這里就不討論異步Servlet有關(guān)的了。
EndPoint
回到Endpoint，AbstarctEndpoint的start()只會判斷一下沒綁定就進(jìn)行下bind()的邏輯，接著就只有一個(gè)給子類的startInternal()方法，所以可以直接看到NioEndPoint的該方法
public void startInternal() throws Exception {
if (!running) {
running = true;
paused = false;
// 實(shí)例化處理器對象緩存棧
processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getProcessorCache());
// 實(shí)例化事件對象緩存棧
eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getEventCache());
// 實(shí)例化NioChannel對象緩存棧
nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getBufferPool());
// 線程池為空則創(chuàng)建新的
if ( getExecutor() == null ) {
createExecutor();
}
// 初始化連接限制計(jì)數(shù)器
initializeConnectionLatch();
// 實(shí)例化并啟動多個(gè)Poller線程
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();
}
// 實(shí)例化并啟動多個(gè)Acceptor線程
startAcceptorThreads();
}
}
到這里就要先好好介紹一下Endpoint里面最重要的幾個(gè)對象了
Acceptor: 用來接收ServerSocketChannel連接請求的對象，會不停的輪詢是否有新的連接加入，也就是調(diào)用serverSocketAccept()方法，然后把連接返回的SocketChannel使用NioChannel包裝，再選擇一個(gè)Poller進(jìn)行注冊，注冊時(shí)會再把NioChannel使用NioSocketWrapper(NioEndPoint中SocketWrapperBase的具體實(shí)現(xiàn)類)進(jìn)行包裝，并生成一個(gè)新的事件PollerEvent，對應(yīng)的也就是jdk的SelectionKey里面的事件類型，，然后加入到Poller的事件隊(duì)列中。
-
Poller: 內(nèi)部持有一個(gè)事件隊(duì)列和多路復(fù)用器(Selector)，內(nèi)部會有一個(gè)while(true)循環(huán)不停的從事件隊(duì)列中取出一個(gè)事件，把事件注冊到持有的Selector，并且會把NioSocketWrapper綁定到對SelectionKey，然后查詢是否有事件就緒，接著會把就緒好的事件去丟到SocketProcessor里面放到線程池去處理，SocketProcessor會取出當(dāng)前的EndPoint中的ConnectionHandler，也就是前面講的在AbstractEndpoint的構(gòu)造方法里面實(shí)例化的，接著它會把NioSocketWrapper丟到Http11Processor里面進(jìn)行處理。
Processor: 可以看看上面的類圖，這里我們關(guān)注點(diǎn)是Http11Processor。這里面會實(shí)例化org.apache.coyote.Request和org.apache.coyote.Response對象，并且持有Http11InputBuffer和Http11OutputBuffer兩個(gè)對象，這兩個(gè)對象會真正的從Socket通道讀取到具體的數(shù)據(jù)，當(dāng)然它的讀取還是通過內(nèi)部持有的NioSocketWrapper，并對請求進(jìn)行初步的處理，也就是解析請求行和請求頭，而請求體是在業(yè)務(wù)層進(jìn)行調(diào)用時(shí)再進(jìn)行讀取和解析。最后會把請求交給Adapter進(jìn)行處理，Adapter會把coyote包下的Request和Response包裝到org.apache.catalina.connector包下的Request和Response，而這兩個(gè)對象是繼承了HttpServletRequest和HttpServletResponse的，再接著請求就會到容器了。
回到上面的代碼首先會初始化化3個(gè)同步棧，processorCache是用來存儲SocketProcessor，evenCache是用來存儲Poller里面的Event，nioChannelCache是用來存儲NioChanel，它們都是SynchronizedStack的實(shí)例，這個(gè)對象是Tomcat用來做對象的緩存池的，也就是說把用完的對象丟到緩存池，下次再需要這些對象時(shí)就可以重用，用的時(shí)候再進(jìn)行重置；緩沖池拿不到了，再創(chuàng)建新的，用完還是丟回緩存池，直到達(dá)到上限。這樣做的好處是減少了常用對象的初始、實(shí)例化的開銷以及減少垃圾收集器對堆內(nèi)存進(jìn)行回收的頻率，但也有一點(diǎn)是會涉及到多個(gè)線程的搶占，有鎖的競爭與釋放的開銷。
接著下面的代碼 initializeConnectionLatch()會初始化一個(gè)限制連接數(shù)量的LimitLatch，Acceptor必須得先拿到鎖才能去監(jiān)聽新連接的到來，否則就一直等待，使用這種方式從上層限制了并發(fā)連接的數(shù)量，保證了正常業(yè)務(wù)處理的進(jìn)行，它的實(shí)現(xiàn)也是基于jdk中的AQS。接下來的代碼就是實(shí)例化并啟動了Poller和Acceptor的線程，到此請求就能正常的到來了。
一次請求的過程
前面寫了那么多，在主線已經(jīng)把初始化和啟動完成了，現(xiàn)在能正常接收并處理請求了，就下來就看看請求是怎么走的，其實(shí)界面介紹Endpoint幾個(gè)組件的時(shí)候已經(jīng)把主要的流程過了一遍，不過還是要來看實(shí)際的代碼，心里才更加踏實(shí)。
Acceptor
要看請求，自然是從連接的到來開始，所以直接看到
Acceptor的run方法
public void run() {
// Loop until we receive a shutdown command
while (endpoint.isRunning()) {
state = AcceptorState.RUNNING;
try {
// 連接計(jì)數(shù)器 達(dá)到最大連接則等待
endpoint.countUpOrAwaitConnection();
U socket = null;
try {
// 從服務(wù)器監(jiān)聽新到來的連接
socket = endpoint.serverSocketAccept();
} catch (Exception ioe) {
// We didn't get a socket
endpoint.countDownConnection();
......
}
if (endpoint.isRunning() && !endpoint.isPaused()) {
// 配置SocketChannel
if (!endpoint.setSocketOptions(socket)) {
endpoint.closeSocket(socket);
}
} else {
endpoint.destroySocket(socket);
}
} catch (Throwable t) {
......
}
}
state = AcceptorState.ENDED;
}
代碼進(jìn)行了精簡，可以看到Acceptor會在一個(gè)while循環(huán)里面，首先看連接數(shù)量是否達(dá)到上線，然后去監(jiān)聽服務(wù)器端的連接請求，連接到來后會獲取到一個(gè)SocketChannel對象，接著會調(diào)用setSocketOptions()方法進(jìn)行參數(shù)的配置，接著會進(jìn)行下一輪循環(huán)，跟進(jìn)去看看
protected boolean setSocketOptions(SocketChannel socket) {
// Process the connection
try {
// 設(shè)置非阻塞
socket.configureBlocking(false);
Socket sock = socket.socket();
// 拿到SocketChannel內(nèi)部的socket進(jìn)行相關(guān)的參數(shù)配置
socketProperties.setProperties(sock);
// 從NioChannel緩沖池拿一個(gè)對象實(shí)例出來
NioChannel channel = nioChannels.pop();
// 為空則創(chuàng)建新的
if (channel == null) {
// 實(shí)例化內(nèi)部NioChannel內(nèi)部的讀寫緩沖區(qū)處理器
SocketBufferHandler bufhandler = new SocketBufferHandler(
socketProperties.getAppReadBufSize(),
socketProperties.getAppWriteBufSize(),
socketProperties.getDirectBuffer());
// 判斷是否開啟ssl
if (isSSLEnabled()) {
channel = new SecureNioChannel(socket, bufhandler, selectorPool, this);
} else {
channel = new NioChannel(socket, bufhandler);
}
} else {
// 不為空就把SocketChannel設(shè)置進(jìn)去
channel.setIOChannel(socket);
// 為防止資源污染 需要進(jìn)行重置 清空內(nèi)部的緩沖區(qū)
channel.reset();
}
// 注冊到眾多Poller中的一個(gè)
getPoller0().register(channel);
} catch (Throwable t) {
......
return false;
}
return true;
}
可以看到，首先會先設(shè)置SocketChannel為非阻塞，然后拿到它內(nèi)部的Socket進(jìn)行基礎(chǔ)的參數(shù)配置，要清楚一點(diǎn)，SocketChannel只是Java Nio出來后為了實(shí)現(xiàn)非阻塞弄出來的一套Api，內(nèi)部實(shí)現(xiàn)還是用的Socket。接著會嘗試從緩沖區(qū)nioChannels從拿到可以復(fù)用的實(shí)例對象，沒有則創(chuàng)建新的，注意創(chuàng)建的時(shí)候需要一個(gè)實(shí)例對象SocketBufferHandler，因?yàn)镹io進(jìn)行讀寫數(shù)據(jù)都是通過Bytebuffer，所以這是為了NioChannel內(nèi)部方便進(jìn)行Bytebuffer的分配、擴(kuò)容、管理使用的。最后看到會把NioChannel注冊到Poller里面，繼續(xù)跟進(jìn)行g(shù)etPoller0().register()方法
public void register(final NioChannel socket) {
// 把NioChannel與當(dāng)前Poller進(jìn)行綁定
socket.setPoller(this);
// 實(shí)例化一個(gè)NioSocketWrapper 把NioChannel再進(jìn)行一層包裝
NioSocketWrapper ka = new NioSocketWrapper(socket, NioEndpoint.this);
// 與NioChannel進(jìn)行綁定
socket.setSocketWrapper(ka);
// 設(shè)置一些參數(shù)...
ka.setPoller(this);
ka.setReadTimeout(getConnectionTimeout());
ka.setWriteTimeout(getConnectionTimeout());
ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
ka.setSecure(isSSLEnabled());
// 從事件隊(duì)列緩沖區(qū)拿到一個(gè)事件
PollerEvent r = eventCache.pop();
// 表明對讀事件感興趣
ka.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
// 給新來的通道事件類型設(shè)置為OP_REGISTER
if ( r==null) r = new PollerEvent(socket,ka,OP_REGISTER);
else r.reset(socket,ka,OP_REGISTER);
// 添加到事件隊(duì)列
addEvent(r);
}
到這里NioChannel又被加了一層包裝NioSocketWrapper，接著會從事件隊(duì)列的緩沖區(qū)eventCache里面拿到一個(gè)事件，并設(shè)置事件類型為OP_REGISTER，接著把事件添加到了Poller內(nèi)部的事件隊(duì)列。
Poller
接著直接看到Poller線程的run方法
public void run() {
// Loop until destroy() is called
while (true) {
boolean hasEvents = false;
try {
// 如果沒關(guān)閉 首先查看是否有事件
if (!close) {
// 看到事件隊(duì)列中是否還有事件
hasEvents = events();
// wakeupCounter原子變量是在添加事件時(shí)進(jìn)行計(jì)數(shù)的
// > 0則表明已經(jīng)有事件注冊到SocketChannel了
if (wakeupCounter.getAndSet(-1) > 0) {
// 立刻喚醒selector
keyCount = selector.selectNow();
} else {
// 到這里表明沒有事件，直接使用selector.select()進(jìn)行阻塞直到超時(shí)
keyCount = selector.select(selectorTimeout);
}
// 更新wakeupCounter為0
wakeupCounter.set(0);
}
// 如果Poller已經(jīng)關(guān)閉了 直接調(diào)用timeout()方法，取消所有已經(jīng)注冊的事件
// 并關(guān)閉已經(jīng)連接的通道
if (close) {
events();
timeout(0, false);
try {
selector.close();
} catch (IOException ioe) {
log.error(sm.getString("endpoint.nio.selectorCloseFail"), ioe);
}
break;
}
} catch (Throwable x) {
ExceptionUtils.handleThrowable(x);
log.error("",x);
continue;
}
// 上次喚醒后沒有事件 再調(diào)用events()看看隊(duì)列是否有事件了
if ( keyCount == 0 ) hasEvents = (hasEvents | events());
// 拿到已經(jīng)有事件就緒的SelectionKey
Iterator<SelectionKey> iterator =
keyCount > 0 ? selector.selectedKeys().iterator() : null;
while (iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
// 拿到SocketChannel的事件注冊到selector時(shí)，返回的SelectionKey綁定的對象
NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
// Attachment may be null if another thread has called
// cancelledKey()
if (attachment == null) {
iterator.remove();
} else {
iterator.remove();
// 綁定的NioSocketWrapper不為空則調(diào)用processKey()方法進(jìn)行事件的處理
processKey(sk, attachment);
}
}
//處理超時(shí)
timeout(keyCount,hasEvents);
}
// 對已經(jīng)停止運(yùn)行的Poller進(jìn)行計(jì)數(shù)
getStopLatch().countDown();
}
主要邏輯都已經(jīng)寫清楚了，同樣Poller也是在一個(gè)while(true)循環(huán)進(jìn)行的，我們前面的時(shí)候說過，調(diào)用getPoller0().register()只是生成新的事件放到事件隊(duì)列里面，并還沒有注冊到selector，粗略一看，Poller的run方法里面有一個(gè)events()方法可能是注冊的地方，跟進(jìn)去看看
public boolean events() {
boolean result = false;
PollerEvent pe = null;
for (int i = 0, size = events.size(); i < size && (pe = events.poll()) != null; i++ ) {
result = true;
try {
pe.run();
// 到這里之前的PollerEvent對象就沒用了 進(jìn)行重置
pe.reset();
// 然后把事件重新丟到緩存隊(duì)列eventsCache中
if (running && !paused) {
eventCache.push(pe);
}
} catch ( Throwable x ) {
log.error("",x);
}
}
return result;
}
發(fā)現(xiàn)這個(gè)方法就是從事件隊(duì)列取出事件，然后調(diào)用了事件的run方法，接著跟進(jìn)去
public void run() {
// 如果事件類型是OP_REGISTER則直接注冊O(shè)P_READ事件
if (interestOps == OP_REGISTER) {
try {
socket.getIOChannel().register(
socket.getPoller().getSelector(), SelectionKey.OP_READ, socketWrapper);
} catch (Exception x) {
log.error(sm.getString("endpoint.nio.registerFail"), x);
}
} else {
// 查詢該SocketChannel是否已經(jīng)在當(dāng)前Poller的selector里面進(jìn)行了注冊
final SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());
try {
// 沒查詢到結(jié)果直接關(guān)閉連接
if (key == null) {
socket.socketWrapper.getEndpoint().countDownConnection();
((NioSocketWrapper) socket.socketWrapper).closed = true;
} else {
// 查詢到結(jié)果則取出key綁定的NioSocketWrapper
final NioSocketWrapper socketWrapper = (NioSocketWrapper) key.attachment();
// 不為空則重新注冊感興趣的事件
if (socketWrapper != null) {
//we are registering the key to start with, reset the fairness counter.
int ops = key.interestOps() | interestOps;
socketWrapper.interestOps(ops);
key.interestOps(ops);
} else {
// 為空則取消已經(jīng)注冊事件到Selector的Socketchannel
socket.getPoller().cancelledKey(key);
}
}
} catch (CancelledKeyException ckx) {
// 捕獲到異常直接取消注冊
try {
socket.getPoller().cancelledKey(key);
} catch (Exception ignore) {}
}
}
}
可以看到代碼，如果感興趣的事件類型是OP_REGISTER的話(而新連接到來，創(chuàng)建PollerEvent對象時(shí)傳入的就是OP_REGISTER)，直接把SocketChannel注冊到Selector，注冊了該Channel的讀事件，并且把NioSocketWrapper綁定到了注冊成功后返回的SelectionKey的attachment中，這樣有事件就緒時(shí)直接把該對象拿出來，進(jìn)行后續(xù)的讀操作，后續(xù)的邏輯都有注釋就不多說了。
現(xiàn)在看到了最終注冊事件的地方，接著回溯到Poller的run方法，可以看到使用selector.select()，等待已連接的通道是否有事件就緒了，然后就是nio標(biāo)準(zhǔn)的寫法，再貼一下代碼
// 拿到已經(jīng)有事件就緒的SelectionKey
Iterator<SelectionKey> iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;
while (iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
// 拿到SocketChannel的事件注冊到selector時(shí)，返回的SelectionKey綁定的對象
NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
// Attachment may be null if another thread has called
// cancelledKey()
if (attachment == null) {
iterator.remove();
} else {
iterator.remove();
// 綁定的NioSocketWrapper不為空則調(diào)用processKey()方法進(jìn)行事件的處理
processKey(sk, attachment);
}
可以看到有事件就緒的通道就會調(diào)用processKey()方法，這個(gè)方法會對已就緒事件的類型進(jìn)行判斷，最后調(diào)用到AbstarcEndpoint里面的processSocket()，直接上代碼
public boolean processSocket(SocketWrapperBase<S> socketWrapper,
SocketEvent event, boolean dispatch) {
try {
if (socketWrapper == null) {
return false;
}
// 從processorCache緩存區(qū)中拿到SocketProcessor
SocketProcessorBase<S> sc = processorCache.pop();
if (sc == null) {
sc = createSocketProcessor(socketWrapper, event);
} else {
sc.reset(socketWrapper, event);
}
// 把該任務(wù)丟到線程池來執(zhí)行
Executor executor = getExecutor();
if (dispatch && executor != null) {
executor.execute(sc);
} else {
sc.run();
}
} catch (RejectedExecutionException ree) {
......
return false;
}
return true;
}
這里就會看到從processorCache里面拿到SocketProcessor對象，并且給它賦值NioSocketWrapper和event后會放到線程池中去處理，這個(gè)SocketProcessor主要做了這么一件事
state = getHandler().process(socketWrapper, event);
getHandler()也就是開始提到的ConnectionHandler，它的process()代碼較多，這里就不貼出來了，總的來說主要就是創(chuàng)建Http11Processor的實(shí)例，并且把它與NioChannel建立映射，如果下次還有這個(gè)連接的請求到來，會從map中復(fù)用同一個(gè)Http11Processor。接著會調(diào)用processor.process()方法，最終會調(diào)用到Http11Processor的service()。
Processor
雖然看代碼實(shí)際點(diǎn)，但這幾個(gè)地方細(xì)節(jié)太多，而且很多與主線無關(guān)的邏輯，就直接總結(jié)一下Http11Processor主要做了以下幾件事
- Http11Processor內(nèi)部持有Http11InputBuffer和Http11OutputBuffer兩個(gè)類，這兩個(gè)類會分別綁定到org.apache.coyote.Request和org.apache.coyote.Response對象，這兩個(gè)類又會綁定了NioSocketWrapper，所有的讀寫還是通過NioSocketWrapper來進(jìn)行的。接著在service()方法會通過Http11InputBuffer讀取連接通道里面的數(shù)據(jù)到緩沖區(qū)，然后進(jìn)行預(yù)處理，解析請求行和請求頭。
- 把解析的結(jié)果填充到request和response實(shí)例對象，接著會調(diào)用getAdapter().service()方法，傳入request和response對象，這個(gè)Adapter對應(yīng)的實(shí)例是CoyoteAdapter，用來將request和response適配到Servlet中的HttpServletRequest和HttpServletResponse接口，Tomcat對應(yīng)的實(shí)現(xiàn)類是org.apache.catalina.connector.Request和org.apache.catalina.connector.Response。
最后還是借一張圖片來看看整個(gè)過程
CoyoteAdapter
Adapter的主要職責(zé)是為了解耦，因?yàn)橥耆梢栽贑onnector(連接器)里面直接調(diào)用方法進(jìn)入Contaienr(容器)的，但由于這兩者之間并無太多關(guān)聯(lián)，對與Connector來說，應(yīng)用層可以有多種協(xié)議，IO模型也可以有多種，但這些東西Container并不關(guān)心，Container只需要上層包裝好的滿足Servlet規(guī)范的接口實(shí)現(xiàn)類即可，所以用了Adapter來連接這兩個(gè)模塊，并且Adapter里面還可以做一些公用的全局管理的邏輯，Adapter的實(shí)現(xiàn)類CoyoteAdapter會對request和response進(jìn)行進(jìn)一步處理，如請求uri是否可以進(jìn)一步處理(也就是使用url拼接的方式進(jìn)行傳參)，解析cookie中的sessionId和ssl的session，并且添加了對異步Servlet的支持以及完成后數(shù)據(jù)緩沖區(qū)的刷新。上面說了中轉(zhuǎn)，所以實(shí)現(xiàn)類里有這么一句邏輯來進(jìn)入到容器
connector.getService().getContainer().getPipeline().getFirst().invoke(
request, response);
也就是調(diào)用Service中的第一個(gè)容器Engine內(nèi)部的Pipeline的第一個(gè)Valve，把reques和response對象傳進(jìn)去，接著每個(gè)容器的的Pipeline中的基礎(chǔ)Valve，也就是StandardxxxValve，都會調(diào)用類似的邏輯進(jìn)入到下一層容器，如果子容器有多個(gè)就會使用之前設(shè)置好的Mapper進(jìn)行路由，直到到達(dá)Wrapper后，就會通過反射來實(shí)例化Servlet對象，過濾器也就是在這一層做的，最終會調(diào)用到Servlet的service()方法，篇幅有限，容器這里就不多費(fèi)口舌了。最后還要了解清楚Endpoint初始化時(shí)提到的NioSelectorPool對象，才算是真正清楚了數(shù)據(jù)讀寫的流程。
NioSelectorPool
前面說過Poller被稱為主selector，而NioSelectorPool被稱為輔助selector，所以最后一個(gè)知識點(diǎn)就是來探索一下NioSelectorPool究竟做了哪些工作。首先要先搞清楚一點(diǎn)，我們前面所說的最終數(shù)據(jù)的讀寫都是使用的NioSocketWrapper，是因?yàn)镹ioSocketWrapper內(nèi)部持有NioChannel，而NioChannel又是對SocketChannel的包裝，所以最終的處理很自然的想到直接在這個(gè)類里面調(diào)用SocketChannel的read()或write()就完事了，但實(shí)際情況可能沒想象中的那么好。想想這么一種情況，前面說過Poller內(nèi)部的selector在檢測到有讀、寫事件就緒的時(shí)候，最終會把對應(yīng)的連接通道讀寫的處理放到線程池中，所以只要讀寫沒完成，當(dāng)前的連接通道會一直占用著線程池中的工作線程。如果數(shù)據(jù)的讀寫能正常進(jìn)行還好，但如果由于網(wǎng)絡(luò)等原因連接通道暫時(shí)不能讀取數(shù)據(jù)，要么我們等待讀寫超時(shí)直接放棄掉讀寫數(shù)據(jù)的連接通道，但如果這個(gè)連接通道馬上可以進(jìn)行數(shù)據(jù)的讀寫處理了，又要重新建立連接，進(jìn)行一系列的調(diào)度；要么等到超時(shí)的時(shí)候，把連接通道重新注冊讀、寫事件到主selector(Poller)，但關(guān)鍵是如果連接還是不能正常使用的話，就會白白的造成selector進(jìn)行空輪詢的消耗，并且即使有讀、寫事件就緒了，還是得經(jīng)歷一系列調(diào)度然后再從主線程(Poller)切換到工作線程(線程池的線程)，增加了線程上下文的切換的消耗。可能聽來沒什么感覺，但這是服務(wù)器端程序，我們希望的是能管理大量的連接和請求的快速響應(yīng)，由于特殊原因造成大量的連接占用了可以正常進(jìn)行數(shù)據(jù)讀寫的連接通道所需的資源，這當(dāng)然不是我們所期望的，所以為了在連接的管理和響應(yīng)之間進(jìn)行協(xié)調(diào)，Tomcat加入了一個(gè)組件NioSelectorPool，用來作channel讀寫超時(shí)的監(jiān)控與連接的輔助管理，之前在Endpoint初始化時(shí)，調(diào)用了NioSelectorPool的open()，可以看看代碼
public void open() throws IOException {
enabled = true;
getSharedSelector();
if (SHARED) {
blockingSelector = new NioBlockingSelector();
blockingSelector.open(getSharedSelector());
}
}
首先會調(diào)用一個(gè)方法getSharedSelector()方法，這個(gè)方法會使用雙重校驗(yàn)鎖的方式實(shí)例化一個(gè)selector單例，接下來會判斷如果是共享(SHARED)，就實(shí)例化一個(gè)NioBlockingSelector對象，這里先說明一下，這個(gè)SHARED表示的含義是NioSelectorPool是否共享同一個(gè)selector(默認(rèn)為true)，如果不共享那么NioSelectorPool內(nèi)部會維護(hù)一個(gè)selector池，selector不夠用時(shí)則創(chuàng)建新的selector。接著后面的代碼，blockingSelector會調(diào)用一個(gè)open()方法，它里面的邏輯就是把當(dāng)前的selector賦值給它內(nèi)部的selector，并且啟動一個(gè)BlockPoller輪詢器線程，這個(gè)BlockPoller與Poller做的事情類似。再來看到NioSocketWrapper，它內(nèi)部也會獲取到EndPoint中NioSelectorPool的實(shí)例，這里以讀操作來分析，它的read()方法最終會調(diào)用到這么一個(gè)方法
private int fillReadBuffer(boolean block, ByteBuffer to) throws IOException {
int nRead;
NioChannel channel = getSocket();
if (block) {
Selector selector = null;
try {
// 獲取到NioSelectorPool中的selector
selector = pool.get();
} catch (IOException x) {
// Ignore
}
try {
NioEndpoint.NioSocketWrapper att = (NioEndpoint.NioSocketWrapper) channel
.getAttachment();
if (att == null) {
throw new IOException("Key must be cancelled.");
}
// 如果是阻塞直接調(diào)用pool里面的read方法，并傳入超時(shí)時(shí)間
nRead = pool.read(to, channel, selector, att.getReadTimeout());
} finally {
if (selector != null) {
pool.put(selector);
}
}
} else {
// 不是阻塞則直接調(diào)用channel的read()方法
nRead = channel.read(to);
if (nRead == -1) {
throw new EOFException();
}
}
return nRead;
}
這里可以看到，首先會進(jìn)行判斷，如果是非阻塞則會直接調(diào)用channel.read()就完事了，但如果是阻塞會調(diào)用到pool里面的read()方法，接著跟進(jìn)去這個(gè)方法
public int read(ByteBuffer buf, NioChannel socket, Selector selector, long readTimeout, boolean block) throws IOException {
if ( SHARED && block ) {
// 如果共享 且是阻塞 則直接調(diào)用BlockingSelector.read()方法
return blockingSelector.read(buf,socket,readTimeout);
}
SelectionKey key = null;
int read = 0;
boolean timedout = false;
int keycount = 1;
long time = System.currentTimeMillis(); //start the timeout timer
try {
while ( (!timedout) ) {
int cnt = 0;
if ( keycount > 0 ) {
// 如果有事件 則直接從SocketChannel里面讀
cnt = socket.read(buf);
if (cnt == -1) {
if (read == 0) {
read = -1;
}
break;
}
// 已讀取數(shù)據(jù)的字節(jié)數(shù)
read += cnt;
// cnt > 0表示讀到數(shù)據(jù)了，再繼續(xù)嘗試讀
if (cnt > 0) continue;
// 如果本次沒讀到數(shù)據(jù)，但之前已經(jīng)讀到了 說明數(shù)據(jù)讀完了 直接跳出循環(huán)
if (cnt==0 && (read>0 || (!block) ) ) break;
}
if ( selector != null ) {
// 重新注冊當(dāng)前channel的讀事件到selector
if (key==null) key = socket.getIOChannel().register(selector, SelectionKey.OP_READ);
else key.interestOps(SelectionKey.OP_READ);
if (readTimeout==0) {
// 如果沒設(shè)置超時(shí)時(shí)間，下一輪會直接退出該循環(huán)
timedout = (read==0);
} else if (readTimeout<0) {
// 如果沒設(shè)置讀取超時(shí)時(shí)間 則直接select()進(jìn)行阻塞等待事件就緒
keycount = selector.select();
} else {
// 同樣select() 但等待超時(shí)時(shí)間后立即返回
keycount = selector.select(readTimeout);
}
}
// 計(jì)算是否超時(shí)
if (readTimeout > 0 && (selector == null || keycount == 0) ) timedout = (System.currentTimeMillis()-time)>=readTimeout;
}
// 超時(shí)了直接拋出異常
if ( timedout ) throw new SocketTimeoutException();
} finally {
if (key != null) {
key.cancel();
if (selector != null) selector.selectNow();//removes the key from this selector
}
}
return read;
}
可以看到，首先到NioSocketWrapper的read()到這里后，會判斷是否是共享并且使用阻塞的方式(這里傳入的直接是true，所以只要判斷SHARED)，是的話直接調(diào)用blockingSelector.read()方法。我們可以先看看不滿足情況下，后面的邏輯，會在一個(gè)while()循環(huán)，首先直接嘗試讀數(shù)據(jù)，如果讀到了數(shù)據(jù)會繼續(xù)嘗試下一次讀，如果沒讀到會重新把當(dāng)前的channel注冊讀事件到當(dāng)前的selector，然后會調(diào)用selector.select()進(jìn)行阻塞，直到有讀事件就緒，就會再去讀數(shù)據(jù)，阻塞的時(shí)間如果達(dá)到超時(shí)時(shí)間就會立即返回，如果channel最后還是沒有事件就緒的話，就會退出循環(huán)，拋出超時(shí)異常?？梢钥吹竭@里selector的用法跟常規(guī)的不一樣，邏輯走到我們肯定是不用共享selector或者使用非阻塞，但之前調(diào)用的時(shí)候就是當(dāng)作阻塞來處理的，所以這里其實(shí)是使用了selector池，也就是說會對每一個(gè)SocketChannel分配一個(gè)selector來進(jìn)行超時(shí)監(jiān)控，有數(shù)據(jù)讀直接就完事了，開始沒讀到數(shù)據(jù)的話也還是會將事件重新注冊到當(dāng)前的selector，最多等待設(shè)置的超時(shí)的時(shí)間，這樣也不關(guān)主selector(Poller)什么事了。接看到進(jìn)入到BlockingSelector的情況，直接看代碼
public int read(ByteBuffer buf, NioChannel socket, long readTimeout) throws IOException {
......
NioSocketWrapper att = (NioSocketWrapper) key.attachment();
int read = 0;
boolean timedout = false;
int keycount = 1; //assume we can read
long time = System.currentTimeMillis(); //start the timeout timer
try {
while(!timedout) {
if (keycount > 0) {
// 直接讀數(shù)據(jù)
read = socket.read(buf);
if (read != 0) {
break;
}
}
try {
if ( att.getReadLatch()==null || att.getReadLatch().getCount()==0) att.startReadLatch(1);
// 添加讀事件到BlockPoller的事件隊(duì)列
poller.add(att,SelectionKey.OP_READ, reference);
if (readTimeout < 0) {
// 沒有設(shè)置超時(shí)則一直等待
att.awaitReadLatch(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
} else {
// 最多等待超時(shí)的時(shí)間
att.awaitReadLatch(readTimeout, TimeUnit.MILLISECONDS);
}
} catch (InterruptedException ignore) {
// Ignore
}
......
if (readTimeout >= 0 && (keycount == 0))
timedout = (System.currentTimeMillis() - time) >= readTimeout;
} //while
if (timedout)
throw new SocketTimeoutException();
} finally {
......
}
return read;
}
可以看到從邏輯上，與NioSelectorPool里面的read()做的事情是類似的，只是這里因?yàn)槭鞘褂玫墓蚕韘elector，所以使用了att.awaitxxxLatch()方法來計(jì)數(shù)，也就是NioSocketWrapper里面的這幾個(gè)方法
protected void awaitLatch(CountDownLatch latch, long timeout, TimeUnit unit) throws InterruptedException {
if ( latch == null ) throw new IllegalStateException("Latch cannot be null");
// Note: While the return value is ignored if the latch does time
// out, logic further up the call stack will trigger a
// SocketTimeoutException
latch.await(timeout,unit);
}
public void awaitReadLatch(long timeout, TimeUnit unit) throws InterruptedException { awaitLatch(readLatch,timeout,unit);}
public void awaitWriteLatch(long timeout, TimeUnit unit) throws InterruptedException { awaitLatch(writeLatch,timeout,unit);}
也就是使用的CountDownLatch來作為超時(shí)的倒計(jì)時(shí)器，各個(gè)NioSocketWrapper相互之間不會受到影響，現(xiàn)在關(guān)鍵是要搞清楚這里阻塞了，在哪里進(jìn)行收到通知解除阻塞。前面使用selector池時(shí)，因?yàn)槊總€(gè)SocketChannel是單獨(dú)的selector監(jiān)控和管理；這里使用共享的方式，關(guān)鍵依賴的就是BlockPoller了，前面說過它與Poller類似，同樣的是從事件隊(duì)列里面取出事件然后在selector里面進(jìn)行注冊，因?yàn)槭枪芾矶鄠€(gè)SocketChannel，必然要查詢有事件就緒的channel，然后分別處理，所以它的run方法里面有這么一段代碼
Iterator<SelectionKey> iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;
while (run && iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
try {
iterator.remove();
sk.interestOps(sk.interestOps() & (~sk.readyOps()));
if ( sk.isReadable() ) {
countDown(attachment.getReadLatch());
}
if (sk.isWritable()) {
countDown(attachment.getWriteLatch());
}
}catch (CancelledKeyException ckx) {
sk.cancel();
countDown(attachment.getReadLatch());
countDown(attachment.getWriteLatch());
}
}
熟悉的代碼，可以看到拿到讀、寫事件就緒的SocketChannel對應(yīng)的SelectionKey后，會先取消事件，防止下次仍然被查詢出來，然后就會根據(jù)事件類型進(jìn)行判斷，最后會調(diào)用countDown()方法來通知之前在NioBlockingSelector.read()里面阻塞的地方，接著循環(huán)開始下一輪，又可以讀數(shù)據(jù)了。到這里，讀寫數(shù)據(jù)的流程就清楚了，這里還有需要注意的點(diǎn)是，NioSocketWrapper的read()方法是阻塞時(shí)才會使用NioSelectorPool進(jìn)行處理，否則直接調(diào)用channel的read()方法；對于write()而言，是直接就丟到NioSelectorPool里面進(jìn)行處理的，因?yàn)榈叫枰憫?yīng)數(shù)據(jù)時(shí)，已經(jīng)經(jīng)歷了連接的建立、數(shù)據(jù)的讀取與解析、業(yè)務(wù)的處理這些過程了，所以寫數(shù)據(jù)對速度更加有要求，一般情況下都能正常寫，但網(wǎng)絡(luò)不通暢或者需要持續(xù)寫數(shù)據(jù)的時(shí)候，如果重新再從Poller里面來進(jìn)行調(diào)度，未免浪費(fèi)時(shí)間，直接在輔助selector內(nèi)部處理了，不需要切換線程，也能讓數(shù)據(jù)響應(yīng)的時(shí)間更快。最后再總結(jié)一下NioSelectorPool的工作方式和作用
- 兩種方式: 第一種使用共享的selector來處理多個(gè)SocketChannel，也就是NioBlockingSelector中啟用單獨(dú)的線程BlockPoller會進(jìn)行通道是否有事件就緒的查詢；第二種是非共享的方式，也就直接在NioSelectorPool里面維護(hù)了一個(gè)selector池，為每個(gè)channel分配一個(gè)selector進(jìn)行超時(shí)監(jiān)控，selector用完則重新丟回selector池。
- 作為輔助的selector，能提高主selector(Poller)處理并發(fā)連接的數(shù)量，對讀寫的超時(shí)進(jìn)行監(jiān)控，網(wǎng)絡(luò)連接較慢時(shí)或者需要持續(xù)讀寫時(shí)能減少從主線程(Poller)到工作線程(線程池中的線程)的線程上下文切換和調(diào)度的開銷。
結(jié)尾
這篇博客從源碼入手，分析了Tomcat的初始化、啟動和從建立連接到數(shù)據(jù)讀寫的過程，比較大的體會魔鬼藏在細(xì)節(jié)中，讀這種源碼時(shí)即使是斷點(diǎn)調(diào)試一步步跟進(jìn)，也會由于分支太多，很容易偏移主線，所以一定要先在宏觀上有一個(gè)整體的把握，再帶著問題去研究具體的模塊才能有比較好的效果。
參考
https://dbaplus.cn/news-134-1990-1.html
https://time.geekbang.org/column/intro/180
http://www.10tiao.com/html/308/201606/2650075890/1.html
https://gearever.iteye.com/blog/1844203
https://www.linuxidc.com/Linux/2015-02/113900p2.htm
