elasticsearch節(jié)點間通信的基礎(chǔ)transport啟動過程
前言
在前一篇中我們分析了cluster的一些元素。接下來的章節(jié)會對cluster的運作機(jī)制做詳細(xì)分析。本節(jié)先分析一些transport，它是cluster間通信的基礎(chǔ)。它有兩種實現(xiàn)，一種是基于netty實現(xiàn)nettytransport，主要用于節(jié)點間的通信。另一種是localtransport，主要是用于同一個jvm上的節(jié)點通信。因為是同一個jvm上的網(wǎng)絡(luò)模擬，localtransport實現(xiàn)上非常簡單，實際用處也非常有限，這里就不過多說明。這一篇的重點是nettytransport。
transport
transport顧名思義是集群通信的基本通道，無論是集群狀態(tài)信息，還是搜索索引請求信息，都是通過transport傳送。elasticsearch定義了tansport，tansportmessage，tansportchannel，tansportrequest，tansportresponse等所需的所有的基礎(chǔ)接口。這里將以transport為主，分析過程中會附帶介紹其它接口。首先看一下transport節(jié)點的定義，如下圖所示：
NettyTransport實現(xiàn)了該接口。分析NettyTransport前簡單說一下Netty的用法，Netty的使用需要三個模塊ServerBootStrap，ClientBootStrap（v3.x）及MessageHandler。ServerBootStrap啟動服務(wù)器，ClientBootStrap啟動客戶端并連接服務(wù)器，MessageHandler是message處理邏輯所在，也就是業(yè)務(wù)邏輯。其它詳細(xì)使用請參考Netty官方文檔。
啟動serverBootStrap
NettyTransport每個在doStart（）方法中啟動serverBootStrap，和ClientBootStrap，并綁定ip，代碼如下所示：
protected void doStart() throws ElasticsearchException {
clientBootstrap = createClientBootstrap();//根據(jù)配置啟動客戶端
……//省略了無關(guān)分代碼
createServerBootstrap(name, mergedSettings);//啟動server端
bindServerBootstrap(name, mergedSettings);//綁定ip
}每一個節(jié)點都需要發(fā)送和接收，因此兩者都需要啟動，client和server的啟動分別在相應(yīng)的方法中，啟動過程就是netty的啟動過程，有興趣可以去看相應(yīng)方法。bindServerBootstrap(name, mergedSettings)將本地ip和斷開綁定到netty同時設(shè)定好export host（export host的具體作業(yè)我也看明白也沒有看到相關(guān)的綁定，需要進(jìn)一步研究）。
啟動client及server的過程中將messagehandler注入到channelpipeline中。至此啟動過程完成，但是client并未連接任何server，連接過程是在節(jié)點啟動后，才連接到其它節(jié)點的。
如何連接到node
方法代碼如下所示：
public void connectToNode(DiscoveryNode node, boolean light) {
//transport的模塊必須要啟動
if (!lifecycle.started()) {
throw new ElasticsearchIllegalStateException("can't add nodes to a stopped transport");
}
//獲取讀鎖，每個節(jié)點可以和多個節(jié)點建立連接，因此這里用讀鎖
globalLock.readLock().lock();
try {
//以node.id為基礎(chǔ)獲取一個鎖，這保證對于每個node只能建立一次連接
connectionLock.acquire(node.id());
try {
if (!lifecycle.started()) {
throw new ElasticsearchIllegalStateException("can't add nodes to a stopped transport");
}
NodeChannels nodeChannels = connectedNodes.get(node);
if (nodeChannels != null) {
return;
}
try {
if (light) {//這里的light，就是對該節(jié)點只獲取一個channel，所有類型（5種連接類型下面會說到）都使用者一個channel
nodeChannels = connectToChannelsLight(node);
} else {
nodeChannels = new NodeChannels(new Channel[connectionsPerNodeRecovery], new Channel[connectionsPerNodeBulk], new Channel[connectionsPerNodeReg], new Channel[connectionsPerNodeState], new Channel[connectionsPerNodePing]);
try {
connectToChannels(nodeChannels, node);
} catch (Throwable e) {
logger.trace("failed to connect to [{}], cleaning dangling connections", e, node);
nodeChannels.close();
throw e;
}
}
// we acquire a connection lock, so no way there is an existing connection
connectedNodes.put(node, nodeChannels);
if (logger.isDebugEnabled()) {
logger.debug("connected to node [{}]", node);
}
transportServiceAdapter.raiseNodeConnected(node);
} catch (ConnectTransportException e) {
throw e;
} catch (Exception e) {
throw new ConnectTransportException(node, "general node connection failure", e);
}
} finally {
connectionLock.release(node.id());
}
} finally {
globalLock.readLock().unlock();
}
}如果不是輕連接，每個server和clien之間都有5中連接，著5中連接承擔(dān)著不同的任務(wù)
連接方法的代碼
protected void connectToChannels(NodeChannels nodeChannels, DiscoveryNode node) {
//五種連接方式，不同的連接方式對應(yīng)不同的集群操作
ChannelFuture[] connectRecovery = new ChannelFuture[nodeChannels.recovery.length];
ChannelFuture[] connectBulk = new ChannelFuture[nodeChannels.bulk.length];
ChannelFuture[] connectReg = new ChannelFuture[nodeChannels.reg.length];
ChannelFuture[] connectState = new ChannelFuture[nodeChannels.state.length];
ChannelFuture[] connectPing = new ChannelFuture[nodeChannels.ping.length];
InetSocketAddress address = ((InetSocketTransportAddress) node.address()).address();
//嘗試建立連接
for (int i = 0; i < connectRecovery.length; i++) {
connectRecovery[i] = clientBootstrap.connect(address);
}
for (int i = 0; i < connectBulk.length; i++) {
connectBulk[i] = clientBootstrap.connect(address);
}
for (int i = 0; i < connectReg.length; i++) {
connectReg[i] = clientBootstrap.connect(address);
}
for (int i = 0; i < connectState.length; i++) {
connectState[i] = clientBootstrap.connect(address);
}
for (int i = 0; i < connectPing.length; i++) {
connectPing[i] = clientBootstrap.connect(address);
}
//獲取每個連接的channel存入到相應(yīng)的channels中便于后面使用。
try {
for (int i = 0; i < connectRecovery.length; i++) {
connectRecovery[i].awaitUninterruptibly((long) (connectTimeout.millis() * 1.5));
if (!connectRecovery[i].isSuccess()) {
throw new ConnectTransportException(node, "connect_timeout[" + connectTimeout + "]", connectRecovery[i].getCause());
}
nodeChannels.recovery[i] = connectRecovery[i].getChannel();
nodeChannels.recovery[i].getCloseFuture().addListener(new ChannelCloseListener(node));
}
for (int i = 0; i < connectBulk.length; i++) {
connectBulk[i].awaitUninterruptibly((long) (connectTimeout.millis() * 1.5));
if (!connectBulk[i].isSuccess()) {
throw new ConnectTransportException(node, "connect_timeout[" + connectTimeout + "]", connectBulk[i].getCause());
}
nodeChannels.bulk[i] = connectBulk[i].getChannel();
nodeChannels.bulk[i].getCloseFuture().addListener(new ChannelCloseListener(node));
}
for (int i = 0; i < connectReg.length; i++) {
connectReg[i].awaitUninterruptibly((long) (connectTimeout.millis() * 1.5));
if (!connectReg[i].isSuccess()) {
throw new ConnectTransportException(node, "connect_timeout[" + connectTimeout + "]", connectReg[i].getCause());
}
nodeChannels.reg[i] = connectReg[i].getChannel();
nodeChannels.reg[i].getCloseFuture().addListener(new ChannelCloseListener(node));
}
for (int i = 0; i < connectState.length; i++) {
connectState[i].awaitUninterruptibly((long) (connectTimeout.millis() * 1.5));
if (!connectState[i].isSuccess()) {
throw new ConnectTransportException(node, "connect_timeout[" + connectTimeout + "]", connectState[i].getCause());
}
nodeChannels.state[i] = connectState[i].getChannel();
nodeChannels.state[i].getCloseFuture().addListener(new ChannelCloseListener(node));
}
for (int i = 0; i < connectPing.length; i++) {
connectPing[i].awaitUninterruptibly((long) (connectTimeout.millis() * 1.5));
if (!connectPing[i].isSuccess()) {
throw new ConnectTransportException(node, "connect_timeout[" + connectTimeout + "]", connectPing[i].getCause());
}
nodeChannels.ping[i] = connectPing[i].getChannel();
nodeChannels.ping[i].getCloseFuture().addListener(new ChannelCloseListener(node));
}
if (nodeChannels.recovery.length == 0) {
if (nodeChannels.bulk.length > 0) {
nodeChannels.recovery = nodeChannels.bulk;
} else {
nodeChannels.recovery = nodeChannels.reg;
}
}
if (nodeChannels.bulk.length == 0) {
nodeChannels.bulk = nodeChannels.reg;
}
} catch (RuntimeException e) {
// clean the futures
for (ChannelFuture future : ImmutableList.<ChannelFuture>builder().add(connectRecovery).add(connectBulk).add(connectReg).add(connectState).add(connectPing).build()) {
future.cancel();
if (future.getChannel() != null && future.getChannel().isOpen()) {
try {
future.getChannel().close();
} catch (Exception e1) {
// ignore
}
}
}
throw e;
}
}以上就是節(jié)點建立連接的過程，每一對client和server間都會建立一定數(shù)量的不同連接。之所以要區(qū)分連接，是因為不同的操作消耗的資源不同，請求的頻率也不同。對于資源消耗少請求頻率高的如ping，可以建立多一些連接，來確保并發(fā)。對于消耗資源多如bulk操作，則要少建立一些連接，保證機(jī)器不被拖垮。節(jié)點的斷開，這是講相應(yīng)的channel釋放的過程。這里就不再做詳細(xì)說明，可以參考相關(guān)源碼。
總結(jié)
nettytransport的連接過程，啟動過程分別啟動client和server，同時將對于的messagehandler注入，啟動多次就是netty的啟動過程。然后綁定server ip和斷開。但是這里并沒有連接，連接發(fā)送在節(jié)點啟動時，節(jié)點啟動會獲取cluster信息，分別對集群中的節(jié)點建立上述的5種連接。
這就是NettyTransport的啟動和連接過程。transport還有一個很重要的功能就是發(fā)送request，及如何處理request，這些功能會在下一篇中分析,希望大家以后多多支持腳本之家！
相關(guān)文章
本文介紹了如何使用MybatisX插件自定義VO對象模板,并提供了一個簡單的示例,首先,文章展示了如何使用FreeMarker語法編寫模板內(nèi)容,接著,詳細(xì)說明了如何配置模板,并通過實際測試驗證了模板的正確性,最后,作者鼓勵大家參考并支持腳本之家2025-01-01
Maven配置文件settings.xml的實現(xiàn)
Maven是一個用于構(gòu)建和管理Java項目的強(qiáng)大工具,它依賴于設(shè)置文件來配置和管理其行為,其中最重要的之一便是settings.xml文件,本文主要介紹了Maven配置文件settings.xml的實現(xiàn),具有一定的參考價值,感興趣的可以了解一下2024-01-01
AOP 從功能的角度來講，可能看作OOP編程方式的一種補(bǔ)充，提供了一種不同的代碼或者系統(tǒng)組織方式，下面這篇文章主要給大家介紹了關(guān)于Spring學(xué)習(xí)教程之AOP模塊的相關(guān)資料，文中通過圖文介紹的非常詳細(xì)，需要的朋友可以參考下2018-05-05
這篇文章主要介紹了如何使用新方式編寫Spring MVC接口,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-09-09
詳解Java正則表達(dá)式中Pattern類和Matcher類
java.util.regex是一個用正則表達(dá)式所訂制的模式來對字符串進(jìn)行匹配工作的類庫包。包括兩個類Pattern和Matcher Pattern，Pattern是一個正則表達(dá)式經(jīng)編譯后的表現(xiàn)模式。Matcher對象是一個狀態(tài)機(jī)器，它依據(jù)Pattern對象做為匹配模式對字符串展開匹配檢查。2016-12-12
Java捕獲ThreadPoolExecutor內(nèi)部線程異常的四種方法
這篇文章主要為大家詳細(xì)介紹了Java捕獲ThreadPoolExecutor內(nèi)部線程異常的四種方法,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以參考一下2025-03-03
Java中Bigdecimal類的toString()方法和toPlainString()方法區(qū)別
BigDecimal類有多個方法可以將其轉(zhuǎn)換為字符串,其中包括toString()和toPlainString(),本文主要介紹了Java中Bigdecimal類的toString()方法和toPlainString()方法區(qū)別,具有一定的參考價值,感興趣的可以了解一下2024-07-07
歡迎來到JAVA反射機(jī)制指南！想要了解如何在JAVA中實現(xiàn)靈活的編程技巧嗎？本指南將帶你揭開JAVA反射機(jī)制的神秘面紗,讓你的編程世界更加精彩！趕緊跟我一起來探索吧！2024-02-02
java.lang.ExceptionInInitializerError異常的解決方法
這篇文章主要為大家詳細(xì)介紹了java.lang.ExceptionInInitializerError異常的解決方法，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-10-10
java.lang.IllegalStateException異常原因和解決辦法
這篇文章主要給大家介紹了關(guān)于java.lang.IllegalStateException異常原因和解決辦法,IllegalStateException是Java標(biāo)準(zhǔn)庫中的一個異常類,通常表示在不合適或無效的情況下執(zhí)行了某個方法或操作,需要的朋友可以參考下2023-07-07