深入理解Netty核心类及其作用

2023-09-17 05:08:04分类：编程 / Java阅读(6645)评论(0)

本文介绍了Netty框架中的核心类及其作用，包括Channel、EventLoop、ChannelPipeline、ByteBuf等，通过对这些类的深入理解，可以更好地使用Netty进行网络编程开发

MessageToByteEncoder

MessageToByteEncoder是一个抽象编码器，子类可重写encode方法把对象编码为ByteBuf输出。

MessageToByteEncoder继承自ChannelOutboundHandlerAdapter，encode在出站是被调用。

public class MyMessageEncoder extends MessageToByteEncoder { @Override protected void encode(ChannelHandlerContext ctx, MessagePO msg, ByteBuf out) throws Exception { System.out.println("MyMessageEncoder.encode，被调用"); String json = JSONObject.toJSONString(msg); out.writeInt(json.getBytes(StandardCharsets.UTF_8).length); out.writeBytes(json.getBytes(StandardCharsets.UTF_8)); } }

ByteToMessageDecoder

ByteToMessageDecoder是一种ChannelInboundHandler，可以称为解码器，负责将byte字节流(ByteBuf)转换成一种Message，Message是应用可以自己定义的一种Java对象。

ByteToMessageDecoder：用于将字节转为消息，需要检测缓冲区是否有足够的字节。

public class MyMessageDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { System.out.println("MyMessageDecoder.decode，被调用"); while (in.readableBytes() >= 4){ int num = in.readInt(); System.out.println("解码出一个整数："+num); out.add(num); } } } 
ReplayingDecoder
ReplayingDecoder：继承自ByteToMessageDecoder，不需要检测缓冲区是否有足够的字节，但是ReplayingDecoder的速度略慢于ByteToMessageDecoder，而且并不是所有的ByteBuf都支持。
项目复杂度高用ReplayingDecoder，否则使用ByteToMessageDecoder。
public class MyMessageDecoder extends ReplayingDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { System.out.println("MyMessageDecoder.decode，被调用"); int length = in.readInt(); byte[] content = new byte[length]; in.readBytes(content); String json = new String(content,StandardCharsets.UTF_8); MessagePO po = JSONObject.parseObject(json,MessagePO.class); out.add(po); } } 
MessageToMessageEncoder
用于从一种消息编码为另外一种消息，例如从POJO到POJO，是一种ChannelOutboundHandler
MessageToMessageDecoder
从一种消息解码为另一种消息，例如POJO到POJO，是一种ChannelInboundHandler
MessageToMessageCodec
整合了MessageToMessageEncoder 和 MessageToMessageDecoder
public class RequestMessageCodec extends MessageToMessageCodec { @Override protected void encode(ChannelHandlerContext ctx, RequestData msg, List out) throws Exception { System.out.println("RequestMessageCodec.encode 被调用 " + msg); String json = JSONObject.toJSONString(msg); out.add(json); } @Override protected void decode(ChannelHandlerContext ctx, String msg, List out) throws Exception { System.out.println("RequestMessageCodec.decode 被调用 " + msg); RequestData po = JSONObject.parseObject(msg, RequestData.class); out.add(po); } } 
ChannelInitializer
ChannelInitializer是一种特殊的ChannelInboundHandler，可以通过一种简单的方式（调用initChannel方法）来初始化Channel。
通常在Bootstrap.handler(ChannelHandler) ， ServerBootstrap.handler(ChannelHandler) 和 ServerBootstrap.childHandler(ChannelHandler)中给Channel设置ChannelPipeline。
注意：当initChannel被执行完后，会将当前的handler从Pipeline中移除。
Bootstrap bootstrap = new Bootstrap().group(group)//设置线程组 .channel(NioSocketChannel.class)//设置客户端通道的实现类 .handler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new NettyClientHandler());//加入自己的处理器 } }); 
ServerBootstrap bootstrap = new ServerBootstrap().group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class)//使用NioServerSocketChannel作为服务器的通道实现 .option(ChannelOption.SO_BACKLOG, 128)//设置线程队列等待连接的个数 .childOption(ChannelOption.SO_KEEPALIVE, true)//设置保持活动连接状态 //      .handler(null)//该Handler对应bossGroup .childHandler(new ChannelInitializer() {//给workerGroup的EventLoop对应的管道设置处理器 @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new NettyServerHandler()); } }); 
SimpleChannelInboundHandler
SimpleChannelInboundHandler继承自ChannelInboundHandlerAdapter，可以通过泛型来规定消息类型。
处理入站的数据我们只需要实现channelRead0方法。
SimpleChannelInboundHandler在接收到数据后会自动release掉数据占用的Bytebuffer资源，ChannelInboundHandlerAdapter不会自动释放。
public class MyClientHandler extends SimpleChannelInboundHandler { @Override protected void channelRead0(ChannelHandlerContext ctx, MessagePO msg) throws Exception { System.out.println("收到服务端消息：" + msg); } } 
DefaultEventLoopGroup
在向pipline中添加ChannelHandler时，可以提供一个新的线程组，Handler业务会在该线程中执行。
当加ChannelHandler需要执行多线程并发业务时，DefaultEventLoopGroup可以派上大用处。
如果没有设置DefaultEventLoopGroup，默认使用的是EventLoopGroup workerGroup = new NioEventLoopGroup();
DefaultEventLoopGroup businessGroup = new DefaultEventLoopGroup(100); ... addLast(businessGroup, new MyNettyServerHandler()) 
/** * 读取客户端发送过来的消息 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf byteBuf = (ByteBuf) msg; System.out.println("收到客户信息：" + byteBuf.toString(CharsetUtil.UTF_8)); System.out.println("客户端地址：" + ctx.channel().remoteAddress()); System.out.println("处理线程：" + Thread.currentThread().getName()); ctx.executor().parent().execute(()->{ try { System.out.println("parent().execute Thread = " + Thread.currentThread().getName()); TimeUnit.SECONDS.sleep(2L); System.out.println("parent任务执行完成1"); } catch (InterruptedException e) { e.printStackTrace(); } }); ctx.executor().parent().execute(()->{ try { System.out.println("parent().execute Thread = " + Thread.currentThread().getName()); TimeUnit.SECONDS.sleep(2L); System.out.println("parent任务执行完成2"); } catch (InterruptedException e) { e.printStackTrace(); } }); ctx.executor().parent().execute(()->{ try { System.out.println("parent().execute Thread = " + Thread.currentThread().getName()); TimeUnit.SECONDS.sleep(2L); System.out.println("parent任务执行完成3"); } catch (InterruptedException e) { e.printStackTrace(); } }); }
以上代码执行日志如下：
收到客户信息：Hello 服务端
客户端地址：/127.0.0.1:60345
处理线程：defaultEventLoopGroup-4-1
parent().execute Thread = defaultEventLoopGroup-4-2
parent().execute Thread = defaultEventLoopGroup-4-3
程序继续~~ defaultEventLoopGroup-4-1
parent().execute Thread = defaultEventLoopGroup-4-4
parent任务执行完成1
parent任务执行完成3
parent任务执行完成2
EventLoop定时任务
可以在Handler中通过方法ctx.channel().eventLoop().schedule()添加定时任务
ctx.channel().eventLoop().schedule(()->{ try { System.out.println("Thread.currentThread().getName() = " + Thread.currentThread().getName()); TimeUnit.SECONDS.sleep(2L); System.out.println("定时任务执行完成"); } catch (InterruptedException e) { e.printStackTrace(); } },10L,TimeUnit.SECONDS); 
到此这篇关于深入理解Netty核心类及其作用的文章就介绍到这了,更多相关Netty核心类内容请搜索0133技术站以前的文章或继续浏览下面的相关文章希望大家以后多多支持0133技术站！
以上就是深入理解Netty核心类及其作用的详细内容，更多请关注0133技术站其它相关文章！
赞(0) 打赏
未经允许不得转载：0133技术站首页 » Java
相关文章
编程
JavaC语言新手学堂数据库ASP编程PHP编程XML/RSSFlex正则表达式R语言汇编语言其他教程移动

置顶推荐
猜你喜欢