netty入门-3 EventLoop和EventLoopGroup，简单的服务器实现_业界新闻

发布时间:2024-07-28 15:35

阅读量:0

文章目录

EventLoop和EventLoopGroup
结语

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>() {        @Override        protected void initChannel(NioSocketChannel ch) {//这里的参数我猜测就是我们新连接的channel        //这里对新channel添加我们自定义的handler            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {                @Override                public 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>() {                 @Override                 protected 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>() {         @Override         protected void initChannel(NioSocketChannel ch)  {             ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));             ch.pipeline().addLast(normalWorkers,"myhandler",               new ChannelInboundHandlerAdapter() {                 @Override                 public 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() {             @Override             public 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);