网站访问量怎么增加,企业站seo点击软件,建筑导航网站,高级网页设计师前言
学习Spark源码绕不开通信#xff0c;Spark通信是基于Netty实现的#xff0c;所以先简单学习总结一下Netty。
Spark 通信历史
最开始: Akka Spark 1.3#xff1a; 开始引入Netty#xff0c;为了解决大块数据#xff08;如Shuffle#xff09;的传输问题 Spark 1.6Spark通信是基于Netty实现的所以先简单学习总结一下Netty。
Spark 通信历史
最开始: Akka Spark 1.3 开始引入Netty为了解决大块数据如Shuffle的传输问题 Spark 1.6支持配置使用 Akka 或者 Netty Spark 2完全废弃Akka,全部使用Netty Akka 是一个用 Scala 编写的库用于简化编写容错的、高可伸缩性的 Java 和 Scala 的 Actor 模型应用。 Spark 借鉴Akka 通过 Netty 实现了类似的简约版的Actor 模型 Netty
Server 主要代码
// 创建ServerBootstrap实例服务器启动对象
ServerBootstrap bootstrap new ServerBootstrap();ChannelFuture channelFuture bootstrap.bind(8888).sync();
// 等待服务器关闭
channelFuture.channel().closeFuture().sync();主要是启动 ServerBootstrap、绑定端口、等待关闭。
Client 主要代码
// 创建Bootstrap实例客户端启动对象
Bootstrap bootstrap new Bootstrap();
// 连接服务端
ChannelFuture channelFuture bootstrap.connect(localhost, 8888).sync();Server 添加 Handler
bootstrap.childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {socketChannel.pipeline().addLast(new ServerHandler());}
});bootstrap.handler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {socketChannel.pipeline().addLast(new ClientHandler());}
});这里的 ServerHandler 和 ClientHandler 都是自己实现的类处理具体的逻辑。 如channelActive 建立连接时发消息给服务器channelRead 读取数据时调用处理读取数据的逻辑。给服务器或者客户端发消息可以用 writeAndFlush 方法。
完整代码
地址https://gitee.com/dongkelun/java-learning/tree/master/netty-learning/src/main/java/com/dkl/java/demo
NettyServer
package com.dkl.java.demo;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;public class NettyServer {public static void main(String[] args) {try {bind();} catch (InterruptedException e) {throw new RuntimeException(e);}}public static void bind() throws InterruptedException {// 创建boss线程组用于接收连接EventLoopGroup bossGroup new NioEventLoopGroup(1);// 创建worker线程组用于处理连接上的I/O操作含有子线程NioEventGroup个数为CPU核数大小的2倍EventLoopGroup workerGroup new NioEventLoopGroup();try {// 创建ServerBootstrap实例服务器启动对象ServerBootstrap bootstrap new ServerBootstrap();// 使用链式编程配置参数// 将boss线程组和worker线程组暂存到ServerBootstrapbootstrap.group(bossGroup, workerGroup);// 设置服务端Channel类型为NioServerSocketChannel作为通道实现bootstrap.channel(NioServerSocketChannel.class);bootstrap.childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {// 添加ServerHandler到ChannelPipeline对workerGroup的SocketChannel客户端设置处理器socketChannel.pipeline().addLast(new ServerHandler());}});// 设置启动参数初始化服务器连接队列大小。服务端处理客户端连接请求是顺序处理一个时间内只能处理一个客户端请求// 当有多个客户端同时来请求时未处理的请求先放入队列中bootstrap.option(ChannelOption.SO_BACKLOG, 1024);// 绑定端口并启动服务器bind方法是异步的sync方法是等待异步操作执行完成返回ChannelFuture异步对象ChannelFuture channelFuture bootstrap.bind(8888).sync();// 等待服务器关闭channelFuture.channel().closeFuture().sync();} finally {// 优雅地关闭boss线程组bossGroup.shutdownGracefully();// 优雅地关闭worker线程组workerGroup.shutdownGracefully();}}
}ServerHandler
package com.dkl.java.demo;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;public class ServerHandler extends ChannelInboundHandlerAdapter {/*** 当 Channel 已经注册到它的 EventLoop 并且能够处理 I/O 时被调用** param ctx* throws Exception*/Overridepublic void channelRegistered(ChannelHandlerContext ctx) throws Exception {System.out.println(执行 channelRegistered);}/*** 当 Channel 从它的 EventLoop 注销并且无法处理任何 I/O 时被调* 用** param ctx* throws Exception*/Overridepublic void channelUnregistered(ChannelHandlerContext ctx) throws Exception {System.out.println(执行 channelUnregistered);}/*** 当 Channel 处于活动状态时被调用Channel 已经连接/绑定并且已经就绪** param ctx* throws Exception*/Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println(执行 channelActive);}/*** 当 Channel 离开活动状态并且不再连接它的远程节点时被调用** param ctx* throws Exception*/Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.out.println(执行 channelInactive);}/*** 当从 Channel 读取数据时被调用** param ctx* param msg* throws Exception*/Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println(执行 channelRead);// 处理接收到的数据ByteBuf byteBuf (ByteBuf) msg;try {// 将接收到的字节数据转换为字符串String message byteBuf.toString(CharsetUtil.UTF_8);// 打印接收到的消息System.out.println(Server端收到客户消息: message);// 发送响应消息给客户端ctx.writeAndFlush(Unpooled.copiedBuffer(我是服务端我收到你的消息啦~, CharsetUtil.UTF_8));} finally {// 释放ByteBuf资源ReferenceCountUtil.release(byteBuf);}}/*** 当 Channel 上的一个读操作完成时被调用对通道的读取完成的事件或通知。当读取完成可通知发送方或其他的相关方告诉他们接受方读取完成** param ctx* throws Exception*/Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {System.out.println(执行 channelReadComplete);}/*** 当 ChannelnboundHandler.fireUserEventTriggered()方法被调用时被* 调用** param ctx* param evt* throws Exception*/Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {System.out.println(执行 userEventTriggered);}/*** 当 Channel 的可写状态发生改变时被调用。可以通过调用 Channel 的 isWritable()方法* * 来检测 Channel 的可写性。与可写性相关的阈值可以通过* * Channel.config().setWriteHighWaterMark()和 Channel.config().setWriteLowWaterMark()方法来* * 设置** param ctx* throws Exception*/Overridepublic void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {System.out.println(执行 channelWritabilityChanged);}Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println(执行 exceptionCaught);// 异常处理cause.printStackTrace();ctx.close();}
}
NettyClient
package com.dkl.java.demo;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;public class NettyClient {public static void main(String[] args) {start();}public static void start() {// 创建EventLoopGroup用于处理客户端的I/O操作EventLoopGroup groupThread new NioEventLoopGroup();try {// 创建Bootstrap实例客户端启动对象Bootstrap bootstrap new Bootstrap();bootstrap.group(groupThread);// 设置服务端Channel类型为NioSocketChannel作为通道实现bootstrap.channel(NioSocketChannel.class);// 设置客户端处理bootstrap.handler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {socketChannel.pipeline().addLast(new ClientHandler());}});// 绑定端口ChannelFuture channelFuture bootstrap.connect(localhost, 8888).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {throw new RuntimeException(e);} finally {// 优雅地关闭线程groupThread.shutdownGracefully();}}
}ClientHandler
package com.dkl.java.demo;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;public class ClientHandler extends ChannelInboundHandlerAdapter {Overridepublic void channelActive(ChannelHandlerContext ctx) {// 连接建立时的处理发送请求消息给服务器ctx.writeAndFlush(Unpooled.copiedBuffer(你好服务端我是客户端测试通道连接, CharsetUtil.UTF_8));}Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {// 处理接收到的数据ByteBuf byteBuf (ByteBuf) msg;try {// 将接收到的字节数据转换为字符串String message byteBuf.toString(CharsetUtil.UTF_8);// 打印接收到的消息System.out.println(受到服务端响应的消息: message);// TODO: 对数据进行业务处理} finally {// 释放ByteBuf资源ReferenceCountUtil.release(byteBuf);}}Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {// 异常处理cause.printStackTrace();ctx.close();}
}运行截图 handler 执行顺序
Server 端
连接时执行 channelRegistered
执行 channelActive
执行 channelRead
执行 channelReadComplete断开连接时执行 channelReadComplete
(强制中断 Client 连接
执行 exceptionCaught
执行 userEventTriggered (exceptionCaught 中 ctx.close()) 触发
)
执行 channelInactive
执行 channelUnregisteredchannelReadComplete 中 ctx.close(); 触发
执行 channelInactive
执行 channelUnregisteredClient 端
执行 channelRegistered
执行 channelActive
执行 channelRead
执行 channelReadCompleteSpark 对应位置
Spark版本3.2.3Server: org.apache.spark.network.server.TransportServer.initClient: org.apache.spark.network.client.TransportClientFactory.createClient
