【Netty源码解读和权威指南】第54篇：Netty在Elasticsearch中的应用——分布式搜索引擎的网络通信

上一篇【第53篇】Netty在Dubbo中的应用——Dubbo网络通信层深度解析
下一篇【第55篇】Netty游戏服务器实战——10万在线游戏服务器架构

一、ES网络通信架构

ES集群各节点的通信： +----------+ +----------+ +----------+ | Node-1 | | Node-2 | | Node-3 | +----------+ +----------+ +----------+ | Transport层 | Transport层 | Transport层 | | (Netty) | (Netty) | (Netty) | +-------------+-------------|-------------| | (节点间RPC) +----------------+ | 协调节点 | +----------------+ | (HTTP) +----------------+ | 客户端(REST) | +----------------+

二、Netty4Transport初始化

// ES的Transport层使用NettypublicclassNetty4TransportextendsTcpTransport{protectedBootstrapcreateClientBootstrap(ThreadContextcontext){Bootstrapb=newBootstrap();b.group(workerGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).option(ChannelOption.SO_KEEPALIVE,true).handler(newChannelInitializer<SocketChannel>(){protectedvoidinitChannel(SocketChannelch){ch.pipeline().addLast("decoder",newNetty4MessageChannelHandler());}});b.validate();returnb;}}

三、节点间RPC调用

// ES节点间通信：TransportService.sendRequest()publicclassTransportService{publicvoidsendRequest(DiscoveryNodenode,Stringaction,TransportRequestrequest,TransportResponseHandlerhandler){// 通过Netty发送请求finalTransport.Connectionconnection=getConnection(node);connection.sendRequest(requestId,action,request,options);}}// 请求-响应匹配（类似RPC）publicclassTransportResponseHandler{publicvoidhandleResponse(TransportResponseresponse){// 处理响应}}

四、ES的自定义协议

ES Transport协议帧： +------+------+------+------+------+------+------+------+ | 'E' | 'S' | 版本 | 保留 | 请求标志 | 状态 | +------+------+------+------+------+------+------+------+ | 请求ID (8B) | +------+------+------+------+------+------+------+------+ | 数据长度 (4B) | +------+------+------+------+------+------+------+------+ | 压缩数据 (variable) | +------+------+------+------+------+------+------+------+

五、HTTP REST API

// ES同时提供HTTP REST接口publicclassNetty4HttpServerTransport{protectedvoidinitChannel(NioSocketChannelch){ch.pipeline().addLast("decoder",newHttpRequestDecoder());ch.pipeline().addLast("encoder",newHttpResponseEncoder());ch.pipeline().addLast("aggregator",newHttpObjectAggregator(104857600));ch.pipeline().addLast("handler",newNetty4HttpRequestHandler());}}

六、总结

上一篇【第53篇】Netty在Dubbo中的应用——Dubbo网络通信层深度解析
下一篇【第55篇】Netty游戏服务器实战——10万在线游戏服务器架构