Netty入门初识Netty helloword netty组件EventLoop源码分析

1.概述

Netty是一个异步的、基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端。

Netty: Home

消息驱动：鼠标自己点击不需要和系统有过多的交互，消息由系统（第三方）循环检测，来捕获并放入消息队列。消息对于点击事件来说是被动产生的，高内聚。

事件驱动：鼠标点击产生点击事件后要向系统发送消息 “我点击了” 的消息，消息是主动产生的。再发送到消息队列中。事件往往会将事件源包装起来。

事件驱动往往和轮询机制相关，它们通常被统称为 event loop。重点在于并不会给每一个事件分配一个轮询来探知其变化，而是设置一个中央轮询中心，用这个轮询中心去轮询每个注册的对象。轮询中心一旦检测到了注册其中的对象有事件发生，那么就通知对此事件感兴趣的对象。而对此事件感兴趣的对象此时会调用的方法被称为回调函数。

参考：事件驱动和消息驱动_wjjiang2333的博客-CSDN博客_消息驱动和事件驱动

事件队列（event queue）：接收事件的入口，存储待处理事件
分发器（event mediator）：将不同的事件分发到不同的业务逻辑单元
事件通道（event channel）：分发器与处理器之间的联系渠道
事件处理器（event processor）：实现业务逻辑，处理完成后会发出事件，触发下一步操作

Netty在java网络应用框架中的地位就好比：spring在javaee开发中的地位

以下的框架都使用到了Netty因为他们都有网络通信需求：

Cassandra - nosql数据库
Spark - 大数据分布式计算框架
Hadoop - 大数据分布式存储框架
RocketMQ - ali开源的消息队列
ElasticSearch - 搜素引擎
gRPC - rpc框架
Dubbo - rpc框架
spring5.x - flux api完全抛弃了tomcat，使用Netty作为服务器端
Zookeeper - 分布式协调框架

Netty的优势：

Netty vs NIO

NIO工作量大，bug多
需要自己构建协议
解决TCP传输问题，如粘包、半包
epoll空轮训导致的cpu100%
对API进行了增强，是指更易使用

Netty采用的模型：CSDN

2.HelloWorld

首先从HelloWorld入手，开发一个简单的服务器和客户端，客户端向服务器发送HelloWorld，服务器接收不返回。

加入Netty的依赖包(注意5.之后的版本已经被弃用)

    <dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.69.Final</version></dependency>

服务器代码：

 public static void main(String[] args) {//服务端启动器，负责组装netty组件，协调他们的工作new ServerBootstrap()//BossEventLoop、WorkerEventLoop,  包涵线程和选择器.group(new NioEventLoopGroup())//选择基于NIO的ServerSocketChannel.channel(NioServerSocketChannel.class)//决定了worker(child)将来能执行哪些操作(handler).childHandler(//channel代表和客户端进行数据读写的通道 Initializer初始化器，负责添加别的handlernew ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel sc) throws Exception {//添加具体的handlersc.pipeline().addLast(new StringDecoder());//将ByteBuf转换为字符串//添加自定义handlersc.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Override//读事件                    ，msg为上一步被转换为字符串的对象public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println(msg);}});}})//绑定监听端口     .bind(8080);}

客户端代码：

 public static void main(String[] args) throws InterruptedException, IOException {//创建启动器类，对应于服务器端的ServerBootstrapChannel channel = new Bootstrap()//添加EventLoop.group(new NioEventLoopGroup())//选择客户端channel实现.channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Override//在连接建立后调用protected void initChannel(NioSocketChannel ch) throws Exception {//编码器，把字符串变为字节数组，发送到服务器ch.pipeline().addLast(new StringEncoder());}})//连接到服务器.connect(new InetSocketAddress("127.0.0.1", 8080))//sync是为了让客户端先同步的方式连上然后再执行后面的信息发送逻辑.sync()//代表连接对象，也就是SocketChannel.channel();channel.writeAndFlush("hello world");}

服务端成功打印了HelloWorld。

理解：

把channel理解为数据的通道

把msg理解为流动的数据，最开始输入的是ByteBuf，但经过pipeline的加工，会变成其它类型的对象，最后输出又变为ByteBuf

把handler理解为数据的处理工序
工序有多道，合并在一起就是pipeline，pipeline负责发布事件(读、读取完成……)传播给每个handler，handler对自己感兴趣的事件进行处理(重写了相应事件处理方法)

handler分为Inbound和Outbound两类

把eventLoop理解为处理数据的工人(线程)
工人可以管理多个channel的 io操作，并且一旦工人负责了某个channel，就要负责到底(绑定)

工人既可以执行io操作，也可以进行任务处理，每位工人有任务队列，队列里可以堆放多个channel的待处理任务，任务分为普通任务、定时任务

工人按照pipeline顺序，依次按照 handler的规划（代码）处理数据，可以为每道工序指定不同的工人

3.组件

3.1EventLoop*

EventLoop本质是一个单线程执行器(同时维护了一个Selector)，里面有run方法处理Channel源源不断的IO事件。

继承关系：

一条线是继承自j.u.c.ScheduledExecutorService因此包含了线程池中所有的方法
另一条线是继承自netty自己的OrderedEventExecutor
- 提供了boolean inEventLoop(Thread thread)方法判断一个线程是否属于此EventLoop
- 提供了parent方法来看看自己属于哪个EventLoopGroup
shutdownGracefully(); 优雅的关闭事件循环组

EventLoopGroup是一组EventLoop，Channel一般会调用EventLoopGroup的 register方法来绑定其中一个EventLoop，后续这个Channel 上的 io事件都由此EventLoop来处理(保证了io事件处理时的线程安全)

继承自netty自己的 EventExecutorGroup实现了lterable接口提供遍历EventLoop 的能力
另有next方法获取集合中下一个 EventLoop
创建时在构造方法可以指定一共多少个核心线程数，默认为下面的数量，根据Netty参数和物理机的线程数决定。建议IO密集型2*n的核心数，计算密集型n+1的核心数

private static final int DEFAULT_EVENT_LOOP_THREADS =
Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));

演示EventLoopGroup执行普通任务和定时任务：

public static void main(String[] args) {//创建EventLoopGroup,构造方法可以指定核心线程数，默认只有1个//NioEventLoopGroup实现类，功能最全，io事件、普通任务、定时任务都可以处理EventLoopGroup group=new NioEventLoopGroup(2);//DefaultEventLoop主要处理普通任务和定时任务EventLoopGroup group1=new DefaultEventLoop();//采用轮询的方式去获得事件循环组里的事件循环EventLoop next = group.next();//执行普通任务group.next().execute(()->{System.out.println(Thread.currentThread().getName());//nioEventLoopGroup-2-1});//执行定时任务group.next().scheduleAtFixedRate(()->{System.out.println(Thread.currentThread().getName());//nioEventLoopGroup-2-2},1, 1,TimeUnit.SECONDS);
}

演示EventLoopGroup执行IO事件：

服务器：

public static void main(String[] args) {new ServerBootstrap()//细分任务。第一个表示boss处理accept事件，第二个是worker处理读写事件.group(new NioEventLoopGroup(),new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class)//将来建立连接后给SocketChannel添加一些处理器.childHandler(new ChannelInitializer<NioSocketChannel>() {@Override//连接建立后执行protected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Override//关心channel的读事件，重写该方法,此时的msg是ByteBuf类型public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;//可以直接转为StringSystem.out.println(Thread.currentThread().getName()+buf.toString(StandardCharsets.UTF_8));}});}}).bind(8080);
}

客户端：

//创建启动器类，对应于服务器端的ServerBootstrap
Channel channel = new Bootstrap()//添加EventLoop.group(new NioEventLoopGroup())//选择客户端channel实现.channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Override//在连接建立后调用protected void initChannel(NioSocketChannel ch) throws Exception {//编码器，把字符串变为自己数组，发送到服务器ch.pipeline().addLast(new StringEncoder());}})//连接到服务器.connect(new InetSocketAddress("127.0.0.1", 8080)).sync().channel();
System.in.read();

调试：

可以看到初始时指定两个EventLoop线程，当有三个客户端时，共用这两个EventLoop。图解如下

其中的head和相当于哨兵

多reactor多线程:

任务再细分：上面代码是多Reactor单线程，IO读写，业务操作都是在Reactor线程中完成的。而多Reactor多线程是要将业务操作从(从Reactor)中分离，当一个客户端的工作量比较大需要花费较长的运行时间时，会影响到它对应的worker，我们此时需要创建一个独立的组处理。

public class HelloNetty {public static void main(String[] args) throws InterruptedException {//创建一个新的EventLoopGroup去专门处理耗时较长的操作，而不是让NioEventLoopGroup去执行耗时较长的任务EventLoopGroup d = new DefaultEventLoopGroup(3);NioEventLoopGroup boss = new NioEventLoopGroup();NioEventLoopGroup worker = new NioEventLoopGroup(2);ChannelFuture cf = new ServerBootstrap().group(boss, worker).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 128)//设置线程队列等待连接的个数.childOption(ChannelOption.SO_KEEPALIVE, true)//设置保持活动连接状态.childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast("handler1",new ChannelInboundHandlerAdapter(){@Override//这个方法就是上面的Nio事件循环的handler，public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf= (ByteBuf) msg;System.out.println("Nio"+buf.toString(StandardCharsets.UTF_8));//把消息传给下一个Handlerctx.fireChannelRead(msg);}}).addLast(d,"def",new ChannelInboundHandlerAdapter(){@Override//这个方法是默认事件循环需要做的操作public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf= (ByteBuf) msg;System.out.println("default"+buf.toString(StandardCharsets.UTF_8));}});}}).bind(8080).sync();cf.channel().closeFuture().sync();}
}

总结一波：也就是我创建了个Boss和两个Worker(即有一个WorkerGroup里面有两个Worker)；Boss专门监听客户端连接，Worker专管客户端的读写事件；现在有个客户端发消息来了，我不仅要取到消息，还要处理对应消息的业务逻辑，业务逻辑处理的事件很长，如果按以前的只添加一个Handler，取消息和业务逻辑全部甩在这里面，那么这个绑定的worker就会去处理这个业务逻辑处理很久，后面还有客户也发了消息到这个worker，也就得不到及时的处理。此时我们用到的多reactor多线程就是我这个worker只取到消息，取到数据后脏话累活又甩出去了留给default那个事件循环组内的线程执行，这样worker处理可读channel的效率就提高了。如下图

多个Handle如何切换的？

如果有多个handler，其中一个必须切换到下一个handler否则这个调用链就会断掉，需采用特定的方法去切换。

那么上面的多个不同的handler(即NioEventLoop->DefaultEventLoop)是如何切换的呢？

首先查看抽象类，ChannelInboundHandlerAdapter的channelRead方法(我们已经重写了该方法)，看上面的代码我们也写上了该fireChannelRead(msg)方法。

往ChannelHandlerContext ctx继续走来到ChannelHandlerContext接口的fireChannelRead方法，实际，该接口的继承类为AbstractChannelHandlerContext，所以主要的代码在该类中，如下

//fireChannelRead方法的根源地，也就是在AbstractChannelHandlerContext抽象类中
public ChannelHandlerContext fireChannelRead(Object msg) {invokeChannelRead(this.findContextInbound(32), msg);//调用下面的方法return this;}static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);//返回下一个 handler的事件循环(EventLoop)，因为EventLoop继承了EventExecutor EventExecutor executor = next.executor();//判断下一个EventLoop是否与当前的EventLoop是同一个线程//如果是的话，直接就可以调用了if (executor.inEventLoop()) {//这个方法里面会执行我们override的channelRead()方法next.invokeChannelRead(m);} //否则，将要执行的代码作为任务提交给下一个EventLoop处理(换人)else {//下一个handler提交任务到它的线程池executor.execute(new Runnable() {public void run() {next.invokeChannelRead(m);}});}}

除了fireChannelRead方法，super.channelRead(ctx, msg);方法也可以切换，因为该方法里面调用的就是fireChannelRead。

小伙伴们可以仔细品味这其中的源码，有一个更好的理解。

关闭：

group.shutdownGracefully();

用于关闭事件循环组，不是立即关闭，而是等到里面的数据全部处理完再关闭，期间不会接收新的任务。

受篇幅影响，还有几个组件为了方便我会写在另一篇博客：

Netty 组件 Channel 、Future 、Promise_清风拂来水波不兴的博客-CSDN博客