Java进阶-Netty-基础
一、NIO模型
image.png
image.png
二、服务端启动流程
//两大线程组
//bossGroup表示监听端口，accept 新连接的线程组
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
//workerGroup表示处理每一条连接的数据读写的线程组
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
//引导类ServerBootstrap，这个类将引导我们进行服务端的启动工作
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap
.group(bossGroup, workerGroup) //给引导类配置两大线程组
.channel(NioServerSocketChannel.class) //指定服务端的IO模型为NIO
.childHandler(new ChannelInitializer<NioSocketChannel>() { //定义后续每条连接的数据读写，业务处理逻辑
protected void initChannel(NioSocketChannel ch) {
}
});
bind(serverBootstrap, 1000); //绑定端口
/***** 绑定方法 ******/
private static void bind(final ServerBootstrap serverBootstrap, final int port) {
serverBootstrap.bind(port).addListener(new GenericFutureListener<Future<? super Void>>() {
public void operationComplete(Future<? super Void> future) {
if (future.isSuccess()) {
System.out.println("端口[" + port + "]绑定成功!");
} else {
System.err.println("端口[" + port + "]绑定失败!");
bind(serverBootstrap, port + 1);
}
}
});
}
服务端启动其他方法：
serverBootstrap.handler(new ChannelInitializer<NioServerSocketChannel>() {
protected void initChannel(NioServerSocketChannel ch) {
System.out.println("服务端启动中");
}
})
- childHandler()用于指定处理新连接数据的读写处理逻辑
- handler()用于指定在服务端启动过程中的一些逻辑（通常不用）
serverBootstrap.attr(AttributeKey.newInstance("serverName"), "nettyServer")
- attr()方法可以给服务端的channel，也就是NioServerSocketChannel指定一些自定义属性，然后我们可以通过channel.attr()取出这个属性
serverBootstrap.childAttr(AttributeKey.newInstance("clientKey"), "clientValue")
- childAttr()可以给每一条连接指定自定义属性，然后后续我们可以通过channel.attr()取出该属性
serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024)
- option()给服务端channel设置一些属性
serverBootstrap
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childOption(ChannelOption.TCP_NODELAY, true)
- childOption()可以给每条连接设置一些TCP底层相关的属性
三、客户端启动流程
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap
.group(workerGroup) //指定线程模型
.channel(NioSocketChannel.class) //指定 IO 类型为 NIO
.handler(new ChannelInitializer<SocketChannel>() { //给引导类指定一个handler，这里主要就是定义连接的业务处理逻辑
@Override
public void initChannel(SocketChannel ch) {
}
});
//建立连接
connect(bootstrap, "127.0.0.1", 1000, MAX_RETRY);
private static void connect(Bootstrap bootstrap, String host, int port, int retry) {
bootstrap.connect(host, port).addListener(future -> {
if (future.isSuccess()) {
System.out.println("连接成功!");
} else if (retry == 0) {
System.err.println("重试次数已用完，放弃连接！");
} else {
// 第几次重连
int order = (MAX_RETRY - retry) + 1;
// 本次重连的间隔
int delay = 1 << order;
System.err.println(new Date() + ": 连接失败，第" + order + "次重连……");
/*
* bootstrap.config() 这个方法返回的是 BootstrapConfig，他是对 Bootstrap 配置参数的抽象
* .group() 返回的是配置的线程模型 workerGroup
*/
bootstrap.config().group().schedule(() -> connect(bootstrap, host, port, retry - 1), delay, TimeUnit
.SECONDS);
}
});
}
bootstrap.attr(AttributeKey.newInstance("clientName"), "nettyClient")
- attr()方法可以给客户端Channel，也就是NioSocketChannel绑定自定义属性，然后我们可以通过channel.attr()取出这个属性
Bootstrap
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.TCP_NODELAY, true)
- option()方法可以给连接设置一些TCP底层相关的属性
四、数据传输载体ByteBuf
客户端和服务端的逻辑处理是均是在启动的时候，通过给逻辑处理链pipeline添加逻辑处理器，来编写数据的读写逻辑。
客户端连接成功之后会回调到逻辑处理器的channelActive方法，而不管是服务端还是客户端，收到数据之后都会调用到channelRead方法。
写数据调用writeAndFlush方法，客户端与服务端交互的二进制数据载体为ByteBuf，ByteBuf通过连接的内存管理器创建，字节数据填充到ByteBuf之后才能写到对端。
ByteBuf结构：
image.png
ByteBuf是一个字节容器，容器里面的的数据分为三个部分：
- 第一个部分是已经丢弃的字节，这部分数据是无效的
- 第二部分是可读字节，这部分数据是ByteBuf的主体数据
- 最后一部分的数据是可写字节，所有写到ByteBuf的数据都会写到这一段
ByteBuf里面总共有writerIndex-readerIndex个字节可读。Netty使用ByteBuf这个数据结构可以有效地区分可读数据和可写数据，读写之间相互没有冲突。
Netty使用了堆外内存，而堆外内存是不被jvm直接管理的，申请到的内存无法被垃圾回收器直接回收，需要手动回收。
在一个函数体里面，只要增加了引用计数（包括ByteBuf的创建和手动调用retain()方法），就必须调用release()方法。
五、通信协议编解码
通信协议设计：
image.png
登录流程：
image.png
channel的attr()的实际用法：可以通过给channel绑定属性来设置某些状态，获取某些状态，不需要额外的map来维持。
六、pipeline与channelHandler
通过责任链设计模式来组织代码逻辑，并且能够支持逻辑的动态添加和删除。
image.png
一条连接对应着一个Channel，这条Channel所有的处理逻辑都在一个叫做ChannelPipeline的对象里面，ChannelPipeline是一个双向链表结构，他和Channel之间是一对一的关系。
ChannelPipeline里面每个节点都是一个ChannelHandlerContext对象，这个对象能够拿到和Channel相关的所有的上下文信息，然后这个对象包着一个重要的对象，那就是逻辑处理器ChannelHandler。
channelHandler分类：
image.png
这两个子接口分别有对应的默认实现，ChannelInboundHandlerAdapter和ChanneloutBoundHandlerAdapter，它们分别实现了两大接口的所有功能，默认情况下会把读写事件传播到下一个handler。
image.png
inBoundHandler的事件通常只会传播到下一个inBoundHandler，outBoundHandler的事件通常只会传播到下一个outBoundHandler，两者相互不受干扰。
ByteToMessageDecoder：解码
public class PacketDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) {
out.add(PacketCodeC.INSTANCE.decode(in));
}
}
SimpleChannelInboundHandler：类型判断和对象传递自动实现，专注于处理对应指令即可。
public class LoginRequestHandler extends SimpleChannelInboundHandler<LoginRequestPacket> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, LoginRequestPacket loginRequestPacket) {
// 登录逻辑
}
}
MessageToByteEncoder：编码
public class PacketEncoder extends MessageToByteEncoder<Packet> {
@Override
protected void encode(ChannelHandlerContext ctx, Packet packet, ByteBuf out) {
PacketCodeC.INSTANCE.encode(out, packet);
}
}
image.png
七、拆包粘包理论与解决方案
对于操作系统来说，只认TCP协议。应用层是按照ByteBuf为单位来发送数据，但是到了底层操作系统仍然是按照字节流发送数据，因此，数据到了服务端，也是按照字节流的方式读入，然后到了Netty应用层面，重新拼装成ByteBuf，而这里的ByteBuf与客户端按顺序发送的ByteBuf可能是不对等的。因此，我们需要在客户端根据自定义协议来组装我们应用层的数据包，然后在服务端根据我们的应用层的协议来组装数据包，这个过程通常在服务端称为拆包，而在客户端称为粘包。
拆包原理：不断从TCP缓冲区中读取数据，每次读取完都需要判断是否是一个完整的数据包。
- 固定长度的拆包器FixedLengthFrameDecoder
- 行拆包器LineBasedFrameDecoder
- 分隔符拆包器DelimiterBasedFrameDecoder
- 基于长度域拆包器LengthFieldBasedFrameDecoder（常用），自定义协议中包含长度域字段，均可以使用这个拆包器来实现应用层拆包
new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 7, 4);
image.png
- 第一个参数指的是数据包的最大长度
- 第二个参数指的是长度域的偏移量
- 第三个参数指的是长度域的长度
拒绝非本协议连接：
public class Spliter extends LengthFieldBasedFrameDecoder {
private static final int LENGTH_FIELD_OFFSET = 7;
private static final int LENGTH_FIELD_LENGTH = 4;
public Spliter() {
super(Integer.MAX_VALUE, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH);
}
@Override
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
// 屏蔽非本协议的客户端
if (in.getInt(in.readerIndex()) != PacketCodeC.MAGIC_NUMBER) {
ctx.channel().close();
return null;
}
return super.decode(ctx, in);
}
}
image.png
八、channelHandler生命周期
ChannelHandler生命周期.png
- ChannelInitializer的实现原理：利用Netty的handler生命周期中channelRegistered()与handlerAdded()两个特性往pipeline添加handler。
- handlerAdded()与handlerRemoved()：用在资源的申请和释放。
- channelActive()与channelInActive()：TCP连接的建立与释放，统计单机的连接数；对客户端连接ip黑白名单的过滤。
- channelRead()：服务端拆包。
- channelReadComplete()：先调用write()方法，然后该方面里面调用ctx.channel().flush()方法，相当于批量刷新。
九、其他
- 通过ChannelHandler的热插拔机制来实现动态删除逻辑，应用程序性能处理更为高效（身份验证）。
- 共享handler：如果一个handler要被多个channel进行共享，必须要加上@ChannelHandler.Sharable，构造单例。
- 压缩handler-合并编解码器：MessageToMessageCodec，使用它可以让我们的编解码操作放到一个类里面去实现。
- 压缩handler-合并平行handler：定义一个map，存放指令到各个指令处理器的映射，调用指令handler的channelRead。
- 减少阻塞主线程的操作：耗时的操作丢到业务线程池中去处理。
- 准确统计处理时长：在业务线程中需要使用监听器回调的方式来统计耗时，如果在NIO线程中调用，就不需要这么干。
更改事件传播源：
- ctx.writeAndFlush()是从pipeline链中的当前节点开始往前找到第一个outBound类型的handler把对象往前进行传播，如果这个对象确认不需要经过其他outBound类型的handler处理，就使用这个方法。
image.png
- ctx.channel().writeAndFlush()是从pipeline链中的最后一个outBound类型的handler开始，把对象往前进行传播，如果你确认当前创建的对象需要经过后面的outBound类型的handler，那么就调用此方法。
image.png
心跳与空闲检测：IdleStateHandler
- 构造函数，有四个参数，其中第一个表示读空闲时间，指的是在这段时间内如果没有数据读到，就表示连接假死；
- 第二个是写空闲时间，指的是在这段时间如果没有写数据，就表示连接假死；
- 第三个参数是读写空闲时间，表示在这段时间内如果没有产生数据读或者写，就表示连接假死。写空闲和读写空闲为0，表示我们不关心者两类条件；
- 最后一个参数表示时间单位。
- 连接假死之后会回调channelIdle() 方法，可手动关闭连接。
- 通常空闲检测时间要比发送心跳的时间的两倍要长一些，这也是为了排除偶发的公网抖动，防止误判。
