netty组件详解-上

netty服务端示例:

private void doStart() throws InterruptedException {System.out.println("netty服务已启动");// 线程组EventLoopGroup group = new NioEventLoopGroup();try {// 创建服务器端引导类ServerBootstrap server = new ServerBootstrap();// 初始化服务器配置server.group(group) // 配置处理客户端的连接线程组.channel(NioServerSocketChannel.class) // 指定channel为 NioServerSocketChannel.localAddress(port) // 配置服务端口号.childHandler(new ChannelInitializer<SocketChannel>() { // 指定客户端通信的处理类，添加到pipline中，进行初始化@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {socketChannel.pipeline().addLast(new EchoServerHandler());}});// 绑定端口，sync()会阻塞到完成ChannelFuture sync = server.bind().sync();// 阻塞当前线程，直到服务器的ServerChannel被关闭sync.channel().closeFuture().sync();}finally {// 关闭资源group.shutdownGracefully().sync();}}

netty各组件解析；

EventLoop 与 EventGroup
EventLoop : 单线程+任务队列
EventGroup: 多个EventLoop

思考问题: netty底层每个channel中的事件都是由同一个EventLoop来处理的，而EventLoop是单线程的，这样无需考虑为了并发冲突而加锁的问题，提升了性能。并发高效的本质，不是关注如何科学安全的加锁，而是想尽办法避免加锁，来提升性能。
channel接口
每个channel都会被注册到一个EventLoop上，以下是Channel抽象出的方法

每个channel都有自己的生命周期,channel在生命周期的不同节点会回调不同的处理函数:
（1）当channel被注册到EventLoop中时，会调用isRegistered()方法，来确认是否注册成功
（2）每个channel在处理事件时，都有一个对应的pipeline，这个pipeline中以责任链模式来处理事件
channelPipeline &&channelHandler
channelPipeline 的实现是一个双向链表，链表的每个节点对应一个channelHandler,每个channel在处理事件时会调用channelPipeline中的各个channelHandler
channelHandler也有自己的生命周期，在添加到channelPipeline 或被移除出channelPipeline 时，会调用相应的生命周期方法。
channelPipeline的入站事件和出站事件
netty如何在同一个channelPipeline中区分出出站事件链路和入站事件链路?
ChanelHandlerContext 上下文
表示ChannelPipeline和ChannelHandler关联
ChannelPipeline 是双向链表
ChanelHandlerContext 维护了双向链表的pre和 next 指针
具体实现:

ChanelHandlerContext 的作用不仅仅只是维护了指针信息，而且还需要控制channelPipeline中每个ChannelHandler处理的方向和数据流动,比如像下面这些:

ChanelHandlerContext 中的写方法区别:

pipline中有一系列链式的处理逻辑:

ctx.write(in) / ctx.writeAndFlush(in):
在某个入站事件中的handler中直接找到pipline中最近的出站事件节点，在出站事件中输出数据.
ctx.pipeline().write(in) /ctx.channel().write(in)
在当前入站事件handler结束后，继续按照pipline中handler的顺序依次处理后，在输出数据。
对比上面两种做法：
可根据业务需求进行优化，不经过pipeline直接返回的效率更快。
channelHandler的适配器
channelHandler根据功能，设计了几种适配器，其中包括ChannelOutboundHandlerAdapter出站事件适配器
问题:为什么ChannelOutboundHandlerAdapter中包含一个read()事件方法
netty将read()动作打包成一个读事件放到了pipeLine中
channelHandler的并发共享机制：
根据netty的设计，每个socketchannel都是由一个EventLoop线程处理的，每个channel中包含一个pipeline,而pipeline中的每个handler在使用的时候都是重新new 的一个实例,由于每个socket都是独立的线程隔离的，因此每个socket是线程安全的。
但有些业务场景需要各个socket之间共享通信，比如要统计服务器接收到的报文总数。此时就要维护一个共享变量 total，每来一个新的
socket都要去进行 total+1的操作，此时就会产生并发安全问题。
netty如何解决这个问题呢，我们可以定义一个共享的hander，这个hander的定义成一个全局共享的，每个socketchannel的pipeline都添加这个handler，通过这个并发共享的handler来实现socket进程间的通信，代码如下:
（1）首先定义一个共享的handler，其内部实现统计报文的业务逻辑：

// 这个注解的含义就是声明这个handler是共享的
// 如果不声明这个注解的话，在添加到pipline中时会报错
@ChannelHandler.Sharable
public class MessageCountHandler extends ChannelDuplexHandler {private static final Logger LOG = LoggerFactory.getLogger(MessageCountHandler.class);private AtomicLong inCount = new AtomicLong(0);private AtomicLong outCount = new AtomicLong(0);@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {LOG.info("收到报文总数："+inCount.incrementAndGet());super.channelRead(ctx, msg);}@Overridepublic void flush(ChannelHandlerContext ctx) throws Exception {LOG.info("发出报文总数："+outCount.incrementAndGet());super.flush(ctx);}
}

（2）服务器端实现

public void start() throws InterruptedException {// 将统计报文的handler定义成共享变量final MessageCountHandler messageCountHandler = new MessageCountHandler();/*线程组*/EventLoopGroup boss  = new NioEventLoopGroup();EventLoopGroup work  = new NioEventLoopGroup();try {/*服务端启动类*/ServerBootstrap b = new ServerBootstrap();b.group(boss,work).channel(NioServerSocketChannel.class)/*指定使用NIO的通信模式*/.localAddress(new InetSocketAddress(port))/*指定监听端口*/ .childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LineBasedFrameDecoder(1024));ch.pipeline().addLast(messageCountHandler); // 添加一个共享的hander到pipeline中ch.pipeline().addLast(new EchoServerMCHandler());}});ChannelFuture f = b.bind().sync();/*异步绑定到服务器，sync()会阻塞到完成*/LOG.info("服务器启动完成");f.channel().closeFuture().sync();/*阻塞当前线程，直到服务器的ServerChannel被关闭*/} finally {boss.shutdownGracefully().sync();work.shutdownGracefully().sync();}}

netty内存泄露与资源释放注意事项
netty底层的实现机制是java的nio，因此其通信机制是面向缓冲区的，在nio中，当channel中有事件发生时，比如读事件时，会将数据读取到一块直接内存中，当我们处理完这部分数据的时候，应该将这块内存资源释放掉，以防止内存泄露。那么这部分netty是怎么做的呢?
（1）netty中数据的处理是在pipeline中，由每个handler进行处理，那么netty在定义pipeline时，默认在链表的头尾帮我们各自实现了一个handler用于资源分配与释放的。
源码:

从这个ch.pipeline()方法点进去后,找到对应实现

在这里插入图片描述

我们可以发现，pipeline的创建是基于DefaultChannelPipeline.class这个类

我们找到DefaultChannelPipeline.class这个类,可以看到这head和tail在pipeline初始化的时候就被添加进去了

我们找到这个方法看看这个headContext，可以看到他继承自AbstractChannelHandlerContext,并实现了入站和出站事件处理方法
在这里插入图片描述
可以看到这是一个内部类，可以看到，他是实现了资源释放及异常处理的方法

正常情况下，如果事件在pipeline中正常传递的情况下，我们无需手动去管理资源，但是有一种情况，需要手动释放资源

上面的情况就是，当事件读取发生异常，或因为某些业务需求，不能将该事件向pipeline中传递时，需要自己实现资源释放逻辑
在这里插入图片描述
此外，大部分的业务逻辑是在入站事件中资源在某个handler中读取异常时终止传递，否则就正常传递，针对这种业务，netty还单独实现了一个handler来实现异常时自动释放资源,即 SimpleChannelInboundHandler:
实现SimpleChannelInboundHandler后，在发生异常时我们无需手动去释放资源，看源码:

在这里插入图片描述
因此，到这里，关于资源释放的问题，我们可以有三种做法:
（1）无论何时都保证让业务在pipeline中正常传递，依靠DefaultPipeLine中的head和tail来保证资源的释放
（2）在代码逻辑中手动释放资源如: ctx.fireChannelRead(msg);
（3）继承SimpleChannelInboundHandler这个类，重写channelRead0(ChannelHandlerContext ctx, ByteBuf msg)方法

同时处理入站和出站事件
netty中根据业务模型为我们提供了 ChannelInboundHandlerAdapter 和 ChannelOutboundHandler分别处理入站和出站事件，但是有时，我们需要同时处理入站和出站事件，这里netty为我们提供了 ChannelDuplexHandler 这个实现，我们看源码: