netty入门-3 EventLoop和EventLoopGroup，简单的服务器实现

文章目录

EventLoop和EventLoopGroup
- 服务器与客户端基本使用
- 增加非NIO工人
- NioEventLoop 处理普通任务与定时任务
结语

EventLoop和EventLoopGroup

二者大概是什么这里不再赘述，前一篇已简述过。
不理解也没关系。
下面会简单使用，看了就能明白是什么
这篇文章只说NioEventLoopGroup
后续文章会有服务器创建类BootStrap的方法总结。

服务器与客户端基本使用

EventLoopGroup的常用实现类是NioEventLoopGroup
就以此为例，我们来看这两者，在使用netty创建一个服务器的过程中处于什么位置。

PS:我觉得既然是专注于这个东西怎么用，就不要太纠结于底层。
我当时看到childHandler，就想弄明白这个到底是怎么把Handler加到新Channel的Pipeline中的。这就要涉及源码。下面这个链式调用其他部分也是一样。想搞明白每个调用的底层还是需要看源码。会给初学者带来没必要的精力耗费。我觉得需要收住好奇心，先会简单使用，再说底层实现。

//服务端
//ServerBootstrap相当于提供了创建服务器的辅助类。允许通过链式调用更优雅的启动一个服务器。
//我们通过一系列链式调用完成了服务器的创建。
new ServerBootstrap()//group相当于我们配置EventLoopGroup的地方，它将用于后续的事件处理.group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))//channel指定通道类型，NioServerSocketChannel是netty的封装类，即NIO中的升级版.channel(NioServerSocketChannel.class)//这个名字child代表子通道，什么意思？即我们有一个NioServerSocketChannel来处理Accept请求。接受的每个连接都会创建自己的SocketChannel用于通信，可当作child，即子通道。//所以顾名思义子通道处理器。即为新创建的每一个子通道都会绑定这个ChannelInitializer，那么它又是做什么的？//ChannelInitializer会在每个新Channel被注册到EventLoopGroup时执行内部我们重写的initChannel方法，将我们在initChannel方法中自定义的处理器添加到我们这个channel的channelpipeline中。即为我们新连接的Channel添加一个handler。关于handler和pipeline，在上篇中简单讲述了是什么。//简单来说，它是对每个新建立的连接Channel，指定我们写好的处理逻辑，之后Channel的读写操作都会经过这些逻辑。.childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) {//这里的参数我猜测就是我们新连接的channel//这里对新channel添加我们自定义的handlerch.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;if (byteBuf != null) {byte[] buf = new byte[16];ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes());log.debug(new String(buf));}}});}//绑定端口，并sync阻塞住}).bind(8080).sync();

这里还有一些理解
channel()方法，将我们指定的NioServerSocketChannel 注册到group中的第一个group内的某个EventLoop，由该EventLoop监听。第一个group中的EventLoop被称为Boss，专门处理连接请求。
所有连接由NioServerSocketChannel处理，同时新连接产生的SocketChannel会注册到第二个group内的某个EventLoop，由该EventLoop监听。这些EventLoop被称为Worker。处理读写请求。
上面程序中相当于有两个Worker，因为创建EventLoopGroup时参数为2。

所以小结一下。每个Channel都会由一个EventLoop监听。并且在事件发生时调用对应的处理器进行处理。即完成对事件的监听和处理。

//客户端
//通过链式调用，拿到连接完毕的Channel
Channel channel = new Bootstrap().group(new NioEventLoopGroup(1)).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {System.out.println("init...");ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));}}).channel(NioSocketChannel.class).connect("localhost", 8080).sync().channel();// 发送消息
channel.writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("wangwu".getBytes()));Thread.sleep(2000);channel.writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("wangwu".getBytes()));

这里多创建几个客户端发消息，会发现服务器中的两个Worker会分别处理属于自己管理Channel的事件。

增加非NIO工人

下面程序有两个handler
LoggingHandler和"myhandler"
引入了一个我们自己创建的DefaultEventLoopGroup 去处理"myhandler"的任务，即非NIO工人

//额外的工程组
//非NIO工人2个
DefaultEventLoopGroup normalWorkers = new DefaultEventLoopGroup(2);
new ServerBootstrap().group(new NioEventLoopGroup(1), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch)  {ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));ch.pipeline().addLast(normalWorkers,"myhandler",new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;if (byteBuf != null) {byte[] buf = new byte[16];ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes());log.debug(new String(buf));}}});}}).bind(8080).sync();

看完这个代码，我们看添加handler时的参数，我们发现"myhandler"被交给了我们自己一开始创建的DefaultEventLoopGroup normalWorkers 来执行。
关系见下图
head和tail看做pipeline中handler链的头尾，可视为无意义。
h1代表LoggingHandler
h2代表"myhandler"
在这里插入图片描述

那么这里老师巧妙的引出了pipeline处理过程中的handler的切换，因为两个handler分别是两个不同的EventLoop去执行，就需要一个执行完后交给另一个EventLoop去执行。然后引出了源码，比较容易让人接受，理解。

//handler换人源码
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);// 下一个 handler 的事件循环是否与当前的事件循环是同一个线程EventExecutor executor = next.executor();// 是，直接调用if (executor.inEventLoop()) {next.invokeChannelRead(m);} // 不是，将要执行的代码作为任务提交给下一个事件循环处理（换人）else {executor.execute(new Runnable() {@Overridepublic void run() {next.invokeChannelRead(m);}});}
}

如果两个 handler 绑定的是同一个线程（EventLoop），那么就直接调用
否则，把要调用的代码封装为一个任务对象，由下一个 handler 的线程来调用

NioEventLoop 处理普通任务与定时任务

NioEventLoop还可以执行普通任务和定时任务

//普通任务
NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);
log.debug("server start...");
Thread.sleep(2000);
nioWorkers.execute(()->{log.debug("normal task...");
});//定时任务
NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);
log.debug("server start...");
Thread.sleep(2000);
nioWorkers.scheduleAtFixedRate(() -> {log.debug("running...");
}, 0, 1, TimeUnit.SECONDS);