elasticsearch节点的transport请求发送处理分析

elasticsearch节点的transport请求发送处理分析

目录transport请求的发送和处理过程request的发送过程request的接受过程request和response是如何被处理request的处理response的处理过程最后总结

transport请求的发送和处理过程

前一篇分析对nettytransport的启动及连接，本篇主要分析transport请求的发送和处理过程。

cluster中各个节点之间需要相互发送很多信息，如master检测其它节点是否存在，node节点定期检测master节点是否存储，cluster状态的发布及搜索数据请求等等。为了保证信息传输，elasticsearch定义了一个19字节长度的信息头HEADER_SIZE = 2 + 4 + 8 + 1 + 4，以'E'，'S'开头，接着是4字节int信息长度，然后是8字节long型信息id，接着是一个字节的status，最后是4字节int型version。

所有的节点间的信息都是以这19个字节开始。同时elasticsearch对于节点间的所有action都定义 了名字，如对master的周期检测action，internal:discovery/zen/fd/master_ping，每个action对应着相应的messagehandler。接下来会进行详分析。

request的发送过程

代码在nettytransport中如下所示：

public void sendRequest(final DiscoveryNode node, final long requestId, final String action, final TransportRequest request, TransportRequestOptions options) throws IOException, TransportException {

//参数说明：node发送的目的节点，requestId请求id，action action名称，request请求，options包括以下几种操作 RECOVERY,BULK,REG,STATE,PING;

　　　　　Channel targetChannel = nodeChannel(node, options);//获取对应节点的channel，channel在连接节点时初始化完成（请参考上一篇）

if (compress) {

options.withCompress(true);

}

byte status = 0;

　　　　　//设置status 包括以下几种STATUS_REQRES = 1 << 0; STATUS_ERROR = 1 << 1; STATUS_COMPRESS = 1 << 2;

　　　　status = TransportStatus.setRequest(status);

　　　　ReleasableBytesStreamOutput bStream = new ReleasableBytesStreamOutput(bigArrays);//初始写出流

boolean addedReleaseListener = false;

try {

MmMGskUtyW bStream.skip(NettyHeader.HEADER_SIZE);//留出message header的位置

StreamOutput stream = bStream;

// only compress if asked, and, the request is not bytes, since then only

// the header part is compressed, and the "body" can't be extracted as compressed

if (options.compress() && (!(request instanceof BytesTransportRequest))) {

status = TransportStatus.setCompress(status);

stream = CompressorFactory.defaultCompressor().streamOutput(stream);

}

stream = new HandlesStreamOutput(stream);

// we pick the smallest of the 2, to support both backward and forward compatibility

// note, this is the only place we need to do this, since from here on, we use the serialized version

// as the version to use also when the node receiving this request will send the response with

Version version = Version.smallest(this.version, node.version());

stream.setVersion(version);

stream.writeString(transportServiceAdapter.action(action, version));

ReleasableBytesReference bytes;

ChannelBuffer buffer;

// it might be nice to somehow generalize this optimization, maybe a smart "paged" bytes output

// that create paged channel buffers, but its tricky to know when to do it (where this option is

// more explicit).

if (request instanceof BytesTransportRequest) {

BytesTransportRequest bRequest = (BytesTransportRequest) request;

assert node.version().equals(bRequest.version());

bRequest.writeThin(stream);

stream.close();

bytes = bStream.bytes();

ChannelBuffer headerBuffer = bytes.toChannelBuffer();

ChannelBuffer contentBuffer = bRequest.bytes().toChannelBuffer();

buffer = ChannelBuffers.wrappedBuffer(NettyUtils.DEFAULT_GATHERING, headerBuffer, contentBuffer);

} else {

request.writeTo(stream);

stream.close();

bytes = bStream.bytes();

buffer = bytes.toChannelBuffer();

}

NettyHeader.writeHeader(buffer, requestId, status, version);//写信息头

ChannelFuture future = targetChannel.write(buffer);//写buffer同时获取future，发送信息发生在这里

ReleaseChannelFutureListener listener = new ReleaseChannelFutureListener(bytes);

future.addListener(listener);//添加listener

addedReleaseListener = true;

transportServiceAdapter.onRequestSent(node, requestId, action, request, options);

} finally {

if (!addedReleaseListener) {

Releasables.close(bStream.bytes());

}

}

}

以上就是request的发送过程，获取目标node的channel封装请求写入信息头，然后发送并使用listener监听，这里transportRequest是一个抽象类，它继承了TransportMessage同时实现了streamable接口。cluster中对它的实现非常多，各个功能都有相应的request，这里就不一一列举，后面的代码分析中会时常涉及。

request的接受过程

request发送只是transport的一部分功能，有发送就要有接收，这样transport的功能才完整。接下来就是对接收过程的分析。上一篇中简单介绍过netty的使用，message的处理是通过MessageHandler处理，因此nettyTransport的信息处理逻辑都在MessageChannelHandler的messageReceived（）方法中，代码如下所示：

public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {

Transports.assertTransportThread();

Object m = e.getMessage();

if (!(m instanceof ChannelBuffer)) {//非buffer之间返回

ctx.sendUpstream(e);

return;

}

　　　　　//解析message头

ChannelBuffer buffer = (ChannelBuffer) m;

int size = buffer.getInt(buffer.readerIndex() - 4);

transportServiceAdapter.received(size + 6);

// we have additional bytes to read, outside of the header

boolean hasMessageBytesToRead = (size - (NettyHeader.HEADER_SIZE - 6)) != 0;

int markedReaderIndex = buffer.readerIndex();

int expectedIndexReader = markedReaderIndex + size;

// netty always copies a buffer, either in NioWorker in its read handler, where it copies to a fresh

// buffer, or in the cumlation buffer, which is cleaned each time

StreamInput streamIn = ChannelBufferStreamInputFactory.create(buffer, size);

　　　　　　//读取信息头中的几个重要元数据

long requestId = buffer.readLong();

byte status = buffer.readByte();

Version version = Version.fromId(buffer.readInt());

StreamInput wrappedStream;

　　　　　　…………

if (TransportStatus.isRequest(status)) {//处理请求

String action = handleRequest(ctx.getChannel(), wrappedStream, requestId, version);

if (buffer.readerIndex() != expectedIndexReader) {

if (buffer.readerIndex() < expectedIndexReader) {

logger.warn("Message not fully read (request) for [{}] and action [{}], resetting", requestId, action);

} else {

logger.warn("Message read past expected size (request) for [{}] and action [{}], resetting", requestId, action);

}

buffer.readerIndex(expectedIndexReader);

}

} else {//处理响应

TransportResponseHandler handler = transportServiceAdapter.onResponseReceived(requestId);

// ignore if its null, the adapter logs it

if (handler != null) {

if (TransportStatus.isError(status)) {

handlerResponseError(wrappedStream, handler);

} else {

handleResponse(ctx.getChannel(), wrappedStream, handler);

}

} else {

// if its null, skip those bytes

buffer.readerIndex(markedReaderIndex + size);

}

…………

wrappedStream.close();

}

以上就是信息处理逻辑，这个方法基础自netty的SimpleChannelUpstreamHandler类。作为MessageHandler会在client和server启动时加入到handler链中，在信息到达后netty会自动调用handler链依次处理。这是netty的内容，就不详细说明，请参考netty文档。

request和response是如何被处理

request的处理

代码如下所示：

protected String handleRequest(Channel channel, StreamInput buffer, long requestId, Version version) throws IOException {

final String action = buffer.readString();//读出action的名字

transportServiceAdapter.onRequestReceived(requestId, action);

final NettyTransportChannel transportChannel = new NettyTransportChannel(transport, transportServiceAdapter, action, channel, requestId, version, profileName);

try {

final TransportRequestHandler handler = transportServiceAdapter.handler(action, version);//获取处理该信息的handler

if (handler == null) {

throw new ActionNotFoundTransportException(action);

}

final TransportRequest request = handler.newInstance();

request.remoteAddress(new InetSocketTransportAddress((InetSocketAddress) channel.getRemoteAddress()));

request.readFrom(buffer);

if (handler.executor() == ThreadPool.Names.SAME) {

//noinspection unchecked

handler.messageReceived(request, transportChannel);//使用该handler处理信息。

} else {

threadPool.executor(handler.executor()).execute(new RequestHandler(handler, request, transportChannel, action));

}

} catch (Throwable e) {

try {

transportChannel.sendResponse(e);

} catch (IOException e1) {

logger.warn("Failed to send error message back to client for action [" + action + "]", e);

logger.warn("Actual Exception", e1);

}

}

return action;

}

几个关键部分在代码中进行了标注。这里仍旧不能看到请求是如何处理的。因为cluster中的请求各种各样，如ping,discovery，index等等，因此不可能使用同一种处理方式。因此request最终又被提交给handler处理。每个功能请求都实现了自己的handler，当请求被提交给handler时会做对应的处理。这里再说一下transportServiceAdapter，消息的处理都是通过它适配转发完成。request的完整处理流程是：messageReceived（）方法收到信息判断是request会将其转发到transportServiceAdapter的handler方法，handler方法查找对应的requesthandler，使用将信息转发给该handler进行处理。这里就不举例说明，在后面的discover分析中我们会看到发现，ping等请求的处理过程。

response的处理过程

response通过handleResponse方法进行处理，代码如下：

protected void handleResponse(Channel channel, StreamInput buffer, final TransportResponseHandler handler) {

final TransportResponse response = handler.newInstance();

response.remoteAddress(new InetSocketTransportAddress((InetSocketAddress) channel.getRemoteAddress()));

response.remoteAddress();

try {

response.readFrom(buffer);

} catch (Throwable e) {

handleException(handler, new TransportSerializationException("Failed to deserialize response of type [" + response.getClass().getName() + "]", e));

return;

}

try {

if (handler.executor() == ThreadPool.Names.SAME) {

//noinspection unchecked

handler.handleResponse(response);//转发给对应的handler

} else {

threadPool.executor(handler.executor()).execute(new ResponseHandler(handler, response));

}

} catch (Throwable e) {

handleException(handler, new ResponseHandlerFailureTransportException(e));

}

}

response的处理过程跟request很类似。每个request都会对应一个handler和一个response的处理handler，会在时候的时候注册到transportService中。请求到达时根据action名称获取到handler处理request，根据requestId获取对应的response handler进行响应。

最后总结

nettyTransport的信息处理过程：信息通过request方法发送到目标节点，目标节点的messagehandler会受到该信息，确定是request还是response，将他们分别转发给transportServiceAdapter，TransportServiceAdapter会查询到对应的handler，信息最终会被转发给对应的handler处理并反馈。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。