elasticsearch節(jié)點(diǎn)間通信的基礎(chǔ)transport啟動(dòng)過程
前言
在前一篇中我們分析了cluster的一些元素。接下來的章節(jié)會(huì)對(duì)cluster的運(yùn)作機(jī)制做詳細(xì)分析。本節(jié)先分析一些transport,它是cluster間通信的基礎(chǔ)。它有兩種實(shí)現(xiàn),一種是基于netty實(shí)現(xiàn)nettytransport,主要用于節(jié)點(diǎn)間的通信。另一種是localtransport,主要是用于同一個(gè)jvm上的節(jié)點(diǎn)通信。因?yàn)槭峭粋€(gè)jvm上的網(wǎng)絡(luò)模擬,localtransport實(shí)現(xiàn)上非常簡單,實(shí)際用處也非常有限,這里就不過多說明。這一篇的重點(diǎn)是nettytransport。
transport
transport顧名思義是集群通信的基本通道,無論是集群狀態(tài)信息,還是搜索索引請求信息,都是通過transport傳送。elasticsearch定義了tansport,tansportmessage,tansportchannel,tansportrequest,tansportresponse等所需的所有的基礎(chǔ)接口。這里將以transport為主,分析過程中會(huì)附帶介紹其它接口。首先看一下transport節(jié)點(diǎn)的定義,如下圖所示:

NettyTransport實(shí)現(xiàn)了該接口。分析NettyTransport前簡單說一下Netty的用法,Netty的使用需要三個(gè)模塊ServerBootStrap,ClientBootStrap(v3.x)及MessageHandler。ServerBootStrap啟動(dòng)服務(wù)器,ClientBootStrap啟動(dòng)客戶端并連接服務(wù)器,MessageHandler是message處理邏輯所在,也就是業(yè)務(wù)邏輯。其它詳細(xì)使用請參考Netty官方文檔。
啟動(dòng)serverBootStrap
NettyTransport每個(gè)在doStart()方法中啟動(dòng)serverBootStrap,和ClientBootStrap,并綁定ip,代碼如下所示:
protected void doStart() throws ElasticsearchException {
clientBootstrap = createClientBootstrap();//根據(jù)配置啟動(dòng)客戶端
……//省略了無關(guān)分代碼
createServerBootstrap(name, mergedSettings);//啟動(dòng)server端
bindServerBootstrap(name, mergedSettings);//綁定ip
}每一個(gè)節(jié)點(diǎn)都需要發(fā)送和接收,因此兩者都需要啟動(dòng),client和server的啟動(dòng)分別在相應(yīng)的方法中,啟動(dòng)過程就是netty的啟動(dòng)過程,有興趣可以去看相應(yīng)方法。bindServerBootstrap(name, mergedSettings)將本地ip和斷開綁定到netty同時(shí)設(shè)定好export host(export host的具體作業(yè)我也看明白也沒有看到相關(guān)的綁定,需要進(jìn)一步研究)。
啟動(dòng)client及server的過程中將messagehandler注入到channelpipeline中。至此啟動(dòng)過程完成,但是client并未連接任何server,連接過程是在節(jié)點(diǎn)啟動(dòng)后,才連接到其它節(jié)點(diǎn)的。
如何連接到node
方法代碼如下所示:
public void connectToNode(DiscoveryNode node, boolean light) {
//transport的模塊必須要啟動(dòng)
if (!lifecycle.started()) {
throw new ElasticsearchIllegalStateException("can't add nodes to a stopped transport");
}
//獲取讀鎖,每個(gè)節(jié)點(diǎn)可以和多個(gè)節(jié)點(diǎn)建立連接,因此這里用讀鎖
globalLock.readLock().lock();
try {
//以node.id為基礎(chǔ)獲取一個(gè)鎖,這保證對(duì)于每個(gè)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,就是對(duì)該節(jié)點(diǎn)只獲取一個(gè)channel,所有類型(5種連接類型下面會(huì)說到)都使用者一個(gè)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();
}
}如果不是輕連接,每個(gè)server和clien之間都有5中連接,著5中連接承擔(dān)著不同的任務(wù)
連接方法的代碼
protected void connectToChannels(NodeChannels nodeChannels, DiscoveryNode node) {
//五種連接方式,不同的連接方式對(duì)應(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);
}
//獲取每個(gè)連接的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é)點(diǎn)建立連接的過程,每一對(duì)client和server間都會(huì)建立一定數(shù)量的不同連接。之所以要區(qū)分連接,是因?yàn)椴煌牟僮飨牡馁Y源不同,請求的頻率也不同。對(duì)于資源消耗少請求頻率高的如ping,可以建立多一些連接,來確保并發(fā)。對(duì)于消耗資源多如bulk操作,則要少建立一些連接,保證機(jī)器不被拖垮。節(jié)點(diǎn)的斷開,這是講相應(yīng)的channel釋放的過程。這里就不再做詳細(xì)說明,可以參考相關(guān)源碼。
總結(jié)
nettytransport的連接過程,啟動(dòng)過程分別啟動(dòng)client和server,同時(shí)將對(duì)于的messagehandler注入,啟動(dòng)多次就是netty的啟動(dòng)過程。然后綁定server ip和斷開。但是這里并沒有連接,連接發(fā)送在節(jié)點(diǎn)啟動(dòng)時(shí),節(jié)點(diǎn)啟動(dòng)會(huì)獲取cluster信息,分別對(duì)集群中的節(jié)點(diǎn)建立上述的5種連接。
這就是NettyTransport的啟動(dòng)和連接過程。transport還有一個(gè)很重要的功能就是發(fā)送request,及如何處理request,這些功能會(huì)在下一篇中分析,希望大家以后多多支持腳本之家!
相關(guān)文章
Maven配置文件settings.xml的實(shí)現(xiàn)
Maven是一個(gè)用于構(gòu)建和管理Java項(xiàng)目的強(qiáng)大工具,它依賴于設(shè)置文件來配置和管理其行為,其中最重要的之一便是settings.xml文件,本文主要介紹了Maven配置文件settings.xml的實(shí)現(xiàn),具有一定的參考價(jià)值,感興趣的可以了解一下2024-01-01
詳解Java正則表達(dá)式中Pattern類和Matcher類
java.util.regex是一個(gè)用正則表達(dá)式所訂制的模式來對(duì)字符串進(jìn)行匹配工作的類庫包。包括兩個(gè)類Pattern和Matcher Pattern,Pattern是一個(gè)正則表達(dá)式經(jīng)編譯后的表現(xiàn)模式。Matcher對(duì)象是一個(gè)狀態(tài)機(jī)器,它依據(jù)Pattern對(duì)象做為匹配模式對(duì)字符串展開匹配檢查。2016-12-12
Java捕獲ThreadPoolExecutor內(nèi)部線程異常的四種方法
這篇文章主要為大家詳細(xì)介紹了Java捕獲ThreadPoolExecutor內(nèi)部線程異常的四種方法,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以參考一下2025-03-03
Java中Bigdecimal類的toString()方法和toPlainString()方法區(qū)別
BigDecimal類有多個(gè)方法可以將其轉(zhuǎn)換為字符串,其中包括toString()和toPlainString(),本文主要介紹了Java中Bigdecimal類的toString()方法和toPlainString()方法區(qū)別,具有一定的參考價(jià)值,感興趣的可以了解一下2024-07-07
java.lang.ExceptionInInitializerError異常的解決方法
這篇文章主要為大家詳細(xì)介紹了java.lang.ExceptionInInitializerError異常的解決方法,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2017-10-10
java.lang.IllegalStateException異常原因和解決辦法
這篇文章主要給大家介紹了關(guān)于java.lang.IllegalStateException異常原因和解決辦法,IllegalStateException是Java標(biāo)準(zhǔn)庫中的一個(gè)異常類,通常表示在不合適或無效的情況下執(zhí)行了某個(gè)方法或操作,需要的朋友可以參考下2023-07-07

