Spark最一开始使用Akka作为内部通信部件，在Spark1.3为了解决大数据的传输问题，引入Netty通信框架，到了1.6版本Spark已经可以配置使用Akka或者Netty了，到Spark2就已经完全抛弃Akka了，全部使用Netty了。
应用场景：分布式部署，各个节点之间需要远程服务调用，高性能的 RPC 框架必不可少，Netty 作为异步高性能的通信框架，往往作为基础通信组件被这些 RPC 框架使用，比如阿里分布式服务框架Dubbo的RPC框架使用 Dubbo 协议进行节点间通信，Dubbo 协议默认使用
Netty 作为基础通信组件，用于实现各进程节点之间的内部通信。Netty 的异步高性能、高可靠性和高成熟度的优点，使它在通信行业得到了大量的应用
2019-10-31_204953.png
- RpcEndpoint：RPC端点 ，Spark针对于每个节点（Client/Master/Worker）都称之一个Rpc端点 ,且都实现RpcEndpoint接口，内部根据不同端点的需求，设计不同的消息和不同的业务处理，如果需要发送（询问）则调用Dispatcher。
- RpcEnv：RPC上下文环境，每个Rpc端点运行时依赖的上下文环境称之为RpcEnv
- Dispatcher：消息分发器，针对于RPC端点需要发送消息或者从远程RPC接收到的消息，分发至对应的指令收件箱/发件箱。如果指令接收方是自己存入收件箱，如果指令接收方为非自身端点，则放入发件箱。
- Inbox：指令消息收件箱，一个本地端点对应一个收件箱，Dispatcher在每次Inbox存入消息时，都将对应EndpointData加入内部待Receiver Queue中，另外Dispatcher创建时会启动一个单独线程进行轮询Receiver Queue，进行收件箱消息消费。
- OutBox：指令消息发件箱，一个远程端点对应一个发件箱，当消息放入Outbox后，紧接着将消息通过TransportClient发送出去。消息放入发件箱以及发送过程是在同一个线程中进行，这样做的主要原因是远程消息分为RpcOutboxMessage,OneWayOutboxMessage两种消息，而针对于需要应答的消息直接发送且需要得到结果进行处理。
- TransportClient：Netty通信客户端，根据OutBox消息的receiver信息，请求对应远程TransportServer。
- TransportServer：Netty通信服务端，一个RPC端点一个TransportServer,接受远程消息后调用Dispatcher分发消息至对应收发件箱。
注意：
TransportClient与TransportServer通信虚线表示两个RpcEnv之间的通信，图示没有单独表达式一个Outbox一个TransportClient，图示没有单独表达式一个RpcEnv中存在两个RpcEndpoint,一个代表本身启动的RPC端点，另外一个为 RpcEndpointVerifier。
实现Netty模型的Server和Client端的通信
import io.netty.bootstrap.ServerBootstrap
import io.netty.channel.ChannelInitializer
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioServerSocketChannel
class NettyServer{
def bind(host:String,port:Int): Unit = {
// 配置服务器线程组 // 用于服务器接收客服端连接的
val group = new NioEventLoopGroup()
// 用户进行网络读写 val loopGroup = new NioEventLoopGroup()
// 是Netty用户启动NIO服务端的辅助类
val bootstrap = new ServerBootstrap()
bootstrap.group(group,loopGroup)
.channel(classOf[NioServerSocketChannel])
// 绑定i/o事件
.childHandler(new ChannelInitializer[SocketChannel] {
override def initChannel(c: SocketChannel): Unit = {
c.pipeline().addLast( new ServerHandler )
}
})
// 绑定端口
val channelFuture = bootstrap.bind(host,port).sync()
// 等待服务关闭
channelFuture.channel().closeFuture().sync()
// 退出 释放线程
group.shutdownGracefully()
loopGroup.shutdownGracefully()
}
}
object NettyServer {
def main(args: Array[String]): Unit = {
val host = "127.0.0.1" val port = "8888".toInt
val server = new NettyServer server.bind(host,port)
}
}
import io.netty.buffer.{ByteBuf, Unpooled}
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
/**
*建立和客户端连接
*/
class ServerHandler extends ChannelInboundHandlerAdapter{
/**
* 当客户端连接上了server后使用此方法
* @param ctx
*/
override def channelActive(ctx: ChannelHandlerContext): Unit = {
println("连接成功")
Thread.sleep(2000)
}
/**
* 接收客户端发来的消息
* @param ctx
* @param msg
*/ override
def channelRead(ctx: ChannelHandlerContext, msg: scala.Any): Unit = {
println("接收到客户端发来的消息")
val byteBuf = msg.asInstanceOf[ByteBuf]
val bytes = new Array[Byte](byteBuf.readableBytes())
byteBuf.readBytes(bytes)
val message = new String(bytes,"UTF-8")
println(message)
val back = " Client 再见！！！"
val resp = Unpooled.copiedBuffer(back.getBytes("UTF-8"))
ctx.write(resp)
}
/**
* 将消息队列汇总的数据写入到SocketChannel 并发送给对方
*/
override def channelReadComplete(ctx: ChannelHandlerContext): Unit = {
ctx.flush()
}
}
import io.netty.bootstrap.Bootstrap
import io.netty.channel.ChannelInitializer
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioSocketChannel
class NettyClient{
def bind(host:String,port:Int): Unit = {
// 配置服务器线程组
// 用于服务器接收客服端连接的
val group = new NioEventLoopGroup()
// 创建客户端的辅助类
val bootstrap = new Bootstrap bootstrap.group(group)
.channel(classOf[NioSocketChannel])
// 绑定i/o事件
.handler(new ChannelInitializer[SocketChannel] {
override def initChannel(c: SocketChannel): Unit = {
c.pipeline().addLast( new ClientHandler )
}
})
// 绑定端口
val channelFuture = bootstrap.connect(host,port).sync()
// 等待服务关闭
channelFuture.channel().closeFuture().sync()
// 退出 释放线程
group.shutdownGracefully()
}
}
object NettyClient {
def main(args: Array[String]): Unit = {
val host = "127.0.0.1"
val port = "8888".toInt
val server = new NettyClient
server.bind(host,port)
}
}
import io.netty.buffer.{ByteBuf, Unpooled}
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} class ClientHandler extends ChannelInboundHandlerAdapter{
override def channelActive(ctx: ChannelHandlerContext): Unit = {
println("发送消息请求")
val content = "Hello Server"
ctx.writeAndFlush(Unpooled.copiedBuffer(content.getBytes("UTF-8")))
}
override def channelRead(ctx: ChannelHandlerContext, msg: scala.Any): Unit = {
println("接收到server发来的消息")
val byteBuf = msg.asInstanceOf[ByteBuf]
val bytes = new Array[Byte](byteBuf.readableBytes())
byteBuf.readBytes(bytes)
val message = new String(bytes,"UTF-8")
println(message)
}
/**
* 将消息队列汇总的数据写入到SocketChannel 并发送给对方
*/
override def channelReadComplete(ctx: ChannelHandlerContext): Unit = {
// 是否循环
// channelActive(ctx)
ctx.flush()
}
}
