广州网站建设厂家,wordpress 不用插件代码高亮,专业教学资源库网站建设工作,fineui 如何做网站继续Netty第二部的内容
一、ChannelHandler
1、ChannelHandler接口
ChannelHandler是Netty的主要组件#xff0c;处理所有的入站和出站数据的应用程序逻辑的容器#xff0c;可以应用在数据的格式转换、异常处理、数据报文统计等
继承ChannelHandler的两个子接口#xff…继续Netty第二部的内容
一、ChannelHandler
1、ChannelHandler接口
ChannelHandler是Netty的主要组件处理所有的入站和出站数据的应用程序逻辑的容器可以应用在数据的格式转换、异常处理、数据报文统计等
继承ChannelHandler的两个子接口
ChannelInboundHandler处理入站数据以及各种状态变化
ChannelOutboundHandler处理出站数据并且允许拦截所有的操作
2、ChannelInboundHandler接口
channelRegistered当Channel已经注册到它的EventLoop并且能够处理I/O时被调用channelUnregistered当Channel从它的EventLoop注销并且无法处理任何I/O时被调用channelActive当Channel处于活动状态时被调用Channel已经连接/绑定并且已经就绪channelInactive当 Channel离开活动状态并且不再连接它的远程节点时被调用channelReadComplete当Channel上的一个读操作完成时被调用channelRead 当从Channel读取数据时被调用channelWritabilityChanged当Channel的可写状态发生改变时被调用。可以通过调用Channel的 isWritable()方法来检测Channel的可写性。与可写性相关的阈值可以通过 Channel.config().setWriteHighWaterMark()和Channel.config().setWriteLowWaterMark()方法来设置userEventTriggered当ChannelnboundHandler.fireUserEventTriggered()方法被调用时被调用。
3、ChannelOutboundHandler接口
出站操作和数据将由 ChannelOutboundHandler 处理。它的方法将被 Channel、ChannelPipeline 以及 ChannelHandlerContext 调用。
所有由 ChannelOutboundHandler 本身所定义的方法
bind(ChannelHandlerContext,SocketAddress,ChannelPromise) 当请求将 Channel 绑定到本地地址时被调用connect(ChannelHandlerContext,SocketAddress,SocketAddress,ChannelPromise) 当请求将 Channel 连接到远程节点时被调用disconnect(ChannelHandlerContext,ChannelPromise) 当请求将 Channel 从远程节点断开时被调用close(ChannelHandlerContext,ChannelPromise) 当请求关闭 Channel 时被调用deregister(ChannelHandlerContext,ChannelPromise) 当请求将 Channel 从它的 EventLoop 注销时被调用read(ChannelHandlerContext) 当请求从 Channel 读取更多的数据时被调用flush(ChannelHandlerContext) 当请求通过 Channel 将入队数据冲刷到远程节点时被调用write(ChannelHandlerContext,Object,ChannelPromise) 当请求通过 Channel 将数据写到 远程节点时被调用
4、ChannelHandler的适配器
ChannelInboundHandlerAdapter处理入站和ChannelOutboundHandlerAdapter处理出站是Netty提供的ChannelHandler基类降低了ChannelHandler实现的复杂度 为什么ChannelOutboundHandlerAdapter会有read()方法
ChannelOutboundHandler.read()是主动触发读事件的或者是处理读事件的前置处理器调用ChannelOutboundHandler.read()方法时Channel会向Selector读取数据读取数据之后会交给ChannelPipeline这时候ChannelPipeline就会触发ChannelInboundHandler.channelRead()读事件
如何在一个ChannelHandler实现同时实现入栈出战
继承ChannelDuplexHandler也可以同时实现ChannelOutboundHandler, ChannelInboundHandler这两个接口
5、共享Handler
ChannelHandlerAdapter提供了isSharable()方法如果实现ChannelHandler的类上指定了Sharable注解isSharable()方法就会返回true反之则否
每个SocketChannel都有自己的ChannelPipeline同时SocketChannel会对应一个EventLoop也就是说只会一个线程来处理ChannelHandler实例之间是完全独立只要不是共享了全局变量ChannelHandler是线程安全的
如何通过共享Handler实现包统计
继承ChannelDuplexHandler类同时类上加上Sharable注解注解指定该实例为static保证只有一个添加到ChannelPipeline中重写read()和flush()方法就可以具体发包收包记录了
6、资源管理和SimpleChannelInboundHandler
在NIO实现中需要依靠创建buffer来进行Channel之间的数据交换
Netty也是同样的设计在read网络数据时由Netty创建Bufferwrite时会拿到这个Buffer写到网络中outBoundHandler处理了write()操作并丢弃了数据而read()需要继续往下一个Handler传递所以没有相关处理
可能产生内存泄露的情况
没有调用相关的fireChannelRead方法也不释放Buffer执行channelRead方法抛出异常导致fireChannelRead方法未执行同时也不释放Buffer
正常执行fireChannelRead方法入站往后传递BufferNetty都会释放BufferNetty提供了SimpleChannelInboundHandler类来支持这种情况继承SimpleChannelInboundHandler类重写channelRead0方法即使不抛出异常也不调用fireChannelRead方法最终Netty也会帮我们做Buffer的释放
Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {boolean release true;try {if (acceptInboundMessage(msg)) {SuppressWarnings(unchecked)I imsg (I) msg;channelRead0(ctx, imsg);} else {release false;ctx.fireChannelRead(msg);}} finally {if (autoRelease release) {ReferenceCountUtil.release(msg);}}
}/*** Is called for each message of type {link I}.** param ctx the {link ChannelHandlerContext} which this {link SimpleChannelInboundHandler}* belongs to* param msg the message to handle* throws Exception is thrown if an error occurred*/
protected abstract void channelRead0(ChannelHandlerContext ctx, I msg) throws Exception;
二、Netty内置通信传输模式
NIOio.netty.channel.socket.nio 使用java.nio.channels包作为基础——基于选择器的方式Epollio.netty.channel.epoll 由JNI驱动的epoll()和非阻塞 IO。这个传输支持只有在Linux上可用的多种特性如SO_REUSEPORT比NIO传输更快而且是完全非阻塞的。将NioEventLoopGroup替换为EpollEventLoopGroup并且将NioServerSocketChannel.class替换为EpollServerSocketChannel.class即可。OIOio.netty.channel.socket.oio 使用java.net包作为基础——使用阻塞流Localio.netty.channel.local 可以在VM内部通过管道进行通信的本地传输Embeddedio.netty.channel.embedded Embedded 传输允许使用ChannelHandler而又不需要一个真正的基于网络的传输。在测试ChannelHandler实现时非常有用
三、引导Bootstrap
ServerBootstrap将绑定到一个端口因为服务器必须要监听连接而Bootstrap则是由想要连接到远程节点的客户端应用程序所使用的。
引导一个客户端只需要一个EventLoopGroup但是一个ServerBootstrap则需要两个也可以是同一个实例
Override
public ServerBootstrap group(EventLoopGroup group) {return group(group, group);
}/*** Set the {link EventLoopGroup} for the parent (acceptor) and the child (client). These* {link EventLoopGroup}s are used to handle all the events and IO for {link ServerChannel} and* {link Channel}s.*/
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {super.group(parentGroup);if (this.childGroup ! null) {throw new IllegalStateException(childGroup set already);}this.childGroup ObjectUtil.checkNotNull(childGroup, childGroup);return this;
}
与ServerChannel相关联的EventLoopGroup将分配一个负责为传入连接请求创建Channel的EventLoop。一旦连接被接受第二个EventLoopGroup就会给它的Channel分配 一个EventLoop。
四、ChannelInitializer
ChannelInitializer是ChannelInboundHandlerAdapter的子类当中的initChannel()方法提供了一种将多个ChannelHandler添加到一个ChannelPipeline中的简便方法。你只需要简单地向Bootstrap或ServerBootstrap的实例提供你的ChannelInitializer实现即可并且一旦Channel被注册到了它的EventLoop之后就会调用你的initChannel()版本。在该方法返回之后ChannelInitializer的实例将会从ChannelPipeline中移除它自己。
/*** This method will be called once the {link Channel} was registered. After the method returns this instance* will be removed from the {link ChannelPipeline} of the {link Channel}.** param ch the {link Channel} which was registered.* throws Exception is thrown if an error occurs. In that case it will be handled by* {link #exceptionCaught(ChannelHandlerContext, Throwable)} which will by default close* the {link Channel}.*/
protected abstract void initChannel(C ch) throws Exception;SuppressWarnings(unchecked)
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {if (initMap.add(ctx)) { // Guard against re-entrance.try {initChannel((C) ctx.channel());} catch (Throwable cause) {// Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...).// We do so to prevent multiple calls to initChannel(...).exceptionCaught(ctx, cause);} finally {ChannelPipeline pipeline ctx.pipeline();if (pipeline.context(this) ! null) {pipeline.remove(this);}}return true;}return false;
}
五、ChannelOption
ChannelOption.SO_BACKLOG对应的是 tcp/ip 协议 listen 函数中的 backlog 参数从Linux2.2开始backlog的参数行为在Linux2.2中发生了变化现在它指定等待接受的完全建立的套接字的队列长度而不是不完整的连接请求的数量ChannelOption.SO_REUSEADDR对应于套接字选项中的SO_REUSEADDR这个参数表示允许重复使用本地地址和端口ChannelOption.SO_KEEPALIVE对应于套接字选项中的 SO_KEEPALIVE如果在两小时内没有数据的通信时TCP会自动发送一 个活动探测数据报文确定连接状态ChannelOption.SO_SNDBUF和ChannelOption.SO_RCVBUF对应于套接字选项中的 SO_SNDBUF和 SO_RCVBUF这两个参数用于操作接 收缓冲区和发送缓冲区的大小ChannelOption.SO_LINGER 对应于套接字选项中的 SO_LINGER保证TCP四次回收最后一次发送close()时数据传输完毕ChannelOption.TCP_NODELAY对应于套接字选项中的TCP_NODELAY该参数的使用与Nagle算法有关Nagle算法是将小的数据包组装为更大的帧然后进行发送而不是输入一次发送一次因此在数据包不足的时候会等待其他数据的到了组装成大的数据包进行发送虽然该方式有效提高网络的有效负载但是却造成了延时而该参数的作用就是禁止使用Nagle算法使用于小数据即时传输于TCP_NODELAY相对应的是TCP_CORK该选项是需要等到发送的数据量最大的时候一次性发送数据适用于文件传输。
六、ByteBuf
ByteBuf 维护了两个不同的索引名称以 read 或者 write 开头的 ByteBuf 方法将会 推进其对应的索引而名称以 set 或者 get 开头的操作则不会。 如果打算读取字节直到 readerIndex 达到和 writerIndex 同样的值时会发生什么。在那 时你将会到达“可以读取的”数据的末尾。就如同试图读取超出数组末尾的数据一样试 图读取超出该点的数据将会触发一个 IndexOutOf-BoundsException。 可以指定 ByteBuf 的最大容量。试图移动写索引即 writerIndex超过这个值将会触发 一个异常。默认的限制是 Integer.MAX_VALUE。
七、粘包/半包问题
1、TCP粘包/半包发生的原因
分包产生的原因就简单的多就是一个数据包被分成了多次接收。 更具体的原因至少包括
应用程序写入数据的字节大小大于套接字发送缓冲区的大小进行 MSS大小的TCP分段。MSS是最大报文段长度的缩写。MSS是TCP报文段中的 数据字段的最大长度。数据字段加上TCP首部才等于整个的TCP报文段。所以MSS并不是TCP报文段的最大长度而是MSSTCP报文段长度-TCP首部长度。 假设客户端分别发送了两个数据包D1和D2给服务端由于服务端一次读取到的字节 数是不确定的故可能存在以下 4 种情况。
服务端分两次读取到了两个独立的数据包分别是D1和D2没有粘包和拆包服务端一次接收到了两个数据包D1和D2粘合在一起被称为TCP粘包服务端分两次读取到了两个数据包第一次读取到了完整的D1包和D2包的部分内容第二次读取到了D2包的剩余内容这被称为TCP拆包服务端分两次读取到了两个数据包第一次读取到了D1包的部分内容D1_1第 二次读取到了D1包的剩余内容D1_2和D2包的整包。
如果此时服务端TCP接收滑窗非常小而数据包D1和D2比较大很有可能会发生第五种可能即服务端分多次才能将D1和D2包接收完全期间发生多次拆包。
2、解决TCP粘包/半包问题
由于底层的 TCP 无法理解上层的业务数据所以在底层是无法保证数据包不被拆分和重 组的这个问题只能通过上层的应用协议栈设计来解决根据业界的主流协议的解决方案 可以归纳如下。
在包尾增加分割符比如回车换行符进行分割例如 FTP 协议 参见 cn.tuling.nettybasic.splicing.linebase回车换行符进行分割和 cn.tuling.nettybasic.splicing.delimiter自定义分割符下的代码消息定长例如每个报文的大小为固定长度 200 字节如果不够空位补空格 参见 cn.tuling.nettybasic.splicing.fixed 下的代码将消息分为消息头和消息体消息头中包含表示消息总长度或者消息体长度 的字段通常设计思路为消息头的第一个字段使用 int32 来表示消息的总长度使用 LengthFieldBasedFrameDecoder
4、分析channelRead和channelReadComplete
Netty是在读到完整的业务请求报文后才调用一次业务ChannelHandler的channelRead方法如果一个业务消息被TCP协议栈发送了N次则服务端的channelReadComplete方法就会被调用N次。
九、编码器和解码器
将字节解码为消息——ByteToMessageDecoder将一种消息类型解码为另一种——MessageToMessageDecoder。
十、序列化问题
1、Java序列化的缺点
无法跨语言序列化后的码流太大序列化性能太低
2、序列化框架
具体可以参考几种Java常用序列化框架的选型与对比-阿里云开发者社区
