Netty一些基本组件

Channel

介绍

这里先理解网络IO相关的Channel，理解Netty网络相关的Channel，可以先了解一下Java NIO的Channel，下面是NIO常用的两种用于处理网络IO的Channel

  1. ServerSocketChannel：用于监听新连接请求，只存在于服务端，在启动端口时创建，对应一个端口
  2. SocketChannel：用于服务端与客户端交互，同时存在服务端与客户端，在建立连接时创建，对应一个连接

Netty基于以上两个Channel实现了两个对应的Channel，NioServerSocketChannel和NioSocketChannel，这两个channel也是在使用Netty过程中最常用的，它们底层分别基于上面两个NIO的Channel，只是另外封装了一些Netty自己的东西，现在一般都是基于Bootstrap构建服务端或者客户端，所以一般都不需要自己定义Channel了，首先需要明确下面介绍一下在代码中channel的一些用法

NioServerSocketChannel

与Java NIO的ServerSocketChannel对应，绑定端口时生效，对应一个端口

使用Bootstrap绑定端口的时候，会得到一个FutureChannel，可以通过它获取NioServerSocketChannel，如下

1
2
ChannelFuture future = bootstrap.bind(8080).sync();
Channel channel = future.channel(); //这里获得的channel就是NioServerSocketChannel

获取NioServerSocketChannel，就可以对它进行添加处理器等操作，如下

1
2
3
4
5
6
7
8
9
10
//不能使用addLast，因为NioServerSocketChannel默认的处理器没有执行ctx.fireChannelRead(msg)，也就不会往下执行
channel.pipeline().addFirst(new ChannelInboundHandlerAdapter(){
    @Override
    //这里的msg就是NioSocketChannel，代表有新的连接
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Channel channel1 = ctx.channel();//这里的channel1和channel是一个对象
        System.out.println(StrUtil.format("线程{}接收到新连接",Thread.currentThread().getName()));
        ctx.fireChannelRead(msg);//这句话是必要的，否则NioServerSocketChannel一些默认的处理器就不会执行
    }
});

NioSocketChannel

与Java NIO的SocketChannel对应，有新连接时创建，对应一个连接

获取NioSocketChannel的方法很多，比如直接使用上面的handler中的msg就是方法之一，一般常见的做法是通过bootstrap.childHandler设置NioSocketChannel的handler，可以在handler中获取，然后进行一些操作，如下

1
2
3
4
5
6
7
8
9
bootstrap.childHandler(
    new ChannelInboundHandlerAdapter(){
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();//这个channel就是NioSocketChannel
        }
    }
);
//有一个特殊的handler：io.netty.channel.ChannelInitializer，基于上面代码的原理，提供了一种更简洁的代码

bootstrap.childHandler：该方法是为新的NioSocketChannel设置默认的处理器，并且可多次执行，如果仅需要对NioSocketChannel设置处理器，该方法就完全够用了

其他

除了Java的NIO（NIO底层基于select模型），Netty还实现了基于epoll的EpollServerSocketChannel和EpollSocketChannel（底层基于epoll模型，但仅适用于linux系统），还有一些其他网络IO相关channel，具体可查看代码

除了网络IO相关的Channel，还有文件IO相关的（比如实现了零拷贝），但是基本上大部分也是基于Java NIO

EventLoop

介绍

EventLoop是一个执行线程，负责处理Channel上的I/O事件和任务调度，确保所有I/O操作异步且非阻塞地执行，这里拿NioEventLoop举例

NioEventLoop有以下特点

  1. 内部包含一个selector
  2. 可以绑定一个或多个Channel（这一点基于selector，NioEventLoop与Channel绑定关系底层是selector与Java NIO的channel绑定（一对多）），Channel通常为NioServerSocketChannel和NioSocketChannel，需要注意，Channel只能绑定一个NioEventLoop
  3. 一个线程，它负责的IO事件和任务都在一个线程执行，它有一个名为run的核心方法，方法内部有一个for无限循环，所有工作都在for循环中同步执行
  4. 实现了Executor接口，包含了execute、submit、schedule等方法，可以用于执行特定的任务，有两个比较重要的队列，taskQueue和scheduledTaskQueue，用于存储普通任务和延时任务，实现了多生产者-单消费者模式，但是生产者、消费这和IO事件处理代码是同一线程，IO事件处理代码中添加的任务需要等到IO事件处理完成后才会执行

核心方法

这里重点讲解一下run方法，也是NioEventLoop的核心方法之一

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
protected void run() {
    int selectCnt = 0;
    for (;;) {
        try {
            int strategy;
            try {
                strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
                switch (strategy) {
                    case SelectStrategy.CONTINUE:
                        continue;

                    case SelectStrategy.BUSY_WAIT:
                        // fall-through to SELECT since the busy-wait is not supported with NIO

                    case SelectStrategy.SELECT:
                        long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
                        if (curDeadlineNanos == -1L) {
                            curDeadlineNanos = NONE; // nothing on the calendar
                        }
                        nextWakeupNanos.set(curDeadlineNanos);
                        try {
                            if (!hasTasks()) {
                                strategy = select(curDeadlineNanos);
                            }
                        } finally {
                            // This update is just to help block unnecessary selector wakeups
                            // so use of lazySet is ok (no race condition)
                            nextWakeupNanos.lazySet(AWAKE);
                        }
                        // fall through
                    default:
                }
            } catch (IOException e) {
                // If we receive an IOException here its because the Selector is messed up. Let's rebuild
                // the selector and retry. https://github.com/netty/netty/issues/8566
                rebuildSelector0();
                selectCnt = 0;
                handleLoopException(e);
                continue;
            }

            selectCnt++;
            cancelledKeys = 0;
            needsToSelectAgain = false;
            final int ioRatio = this.ioRatio;
            boolean ranTasks;
            if (ioRatio == 100) {
                try {
                    if (strategy > 0) {
                        processSelectedKeys();
                    }
                } finally {
                    // Ensure we always run tasks.
                    ranTasks = runAllTasks();
                }
            } else if (strategy > 0) {
                final long ioStartTime = System.nanoTime();
                try {
                    processSelectedKeys();
                } finally {
                    // Ensure we always run tasks.
                    final long ioTime = System.nanoTime() - ioStartTime;
                    ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            } else {
                ranTasks = runAllTasks(0); // This will run the minimum number of tasks
            }

            if (ranTasks || strategy > 0) {
                if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
                    logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
                                 selectCnt - 1, selector);
                }
                selectCnt = 0;
            } else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
                selectCnt = 0;
            }
        } catch (CancelledKeyException e) {
            // Harmless exception - log anyway
            if (logger.isDebugEnabled()) {
                logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
                             selector, e);
            }
        } catch (Error e) {
            throw e;
        } catch (Throwable t) {
            handleLoopException(t);
        } finally {
            // Always handle shutdown even if the loop processing threw an exception.
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Error e) {
                throw e;
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }
}

这里重点看一下调用的以下三个方法

  1. select：io.netty.channel.nio.NioEventLoop#select，作用是监听新的IO事件，会阻塞线程，底层调用的是selector.select，使用了一个参数curDeadlineNanos（该方法下一次返回的时间（有IO事件也会提前返回），为空则必须有IO事件才会返回），这个参数来源于NioEventLoop中的延时任务的下一次执行时间，目的是为了让延时任务正常执行
  2. processSelectedKeys：io.netty.channel.nio.NioEventLoop#processSelectedKeys，这个方法是处理监听到的IO事件（accept、read、write等），会循环处理，处理完一个再处理下一个，这里就有个问题，因为是单线程，如果processSelectedKeys方法执行过久，会导致很久都无法执行到runAllTasks方法，也就会导致定时任务不能按规定时间执行
  3. runAllTasks：io.netty.util.concurrent.SingleThreadEventExecutor#runAllTasks()，用于执行提交的任务，它会先调用fetchFromScheduledTaskQueue方法将scheduledTaskQueue中到期的任务放入taskQueue，再统一循环执行taskQueue中的所有任务，执行完runAllTasks方法后，并且channel没有关闭，则继续for循环，回到select方法继续阻塞，等待其下次返回

用法

在一般情况下可以在handler中通过ctx获取eventLoop，然后进行提交任务等操作，如下

1
2
3
4
5
6
7
8
9
10
bootstrap.childHandler(
    new ChannelInboundHandlerAdapter(){
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ctx.channel().eventLoop().execute(() -> System.out.println("execute"));
            ctx.channel().eventLoop().submit(() -> System.out.println("submit"));
            ctx.channel().eventLoop().schedule(() -> System.out.println("submit"), 10, TimeUnit.SECONDS);
        }
    }
);

上面的代码中提交了三个任务，这三个任务都不会立即执行，需要等到当前IO事件处理代码执行完之后才会在runAllTasks方法中执行，第三个任务也许更久，第一个和第二个的区别在于submit方法会返回Future

EventLoopGroup

EventLoopGroup 是 Netty 中用于管理多个 EventLoop 的组件，通常包含多个 EventLoop 以实现高效的并发处理

一般使用bootstrap.group方法设置EventLoopGroup，如下

1
2
3
4
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)；

这里的bossGroup负责处理ServerSocketChannel上的连接事件（这里bossGroup即使设置了多线程，但是一个ServerSocketChannel，即一个端口上也只会分配一个EventLoop），workerGroup则负责处理SocketChannel上的读写事件

EventLoopGroup的使用也决定了使用何种Reactor线程模型

单reactor、单线程模型如下

1
2
3
EventLoopGroup group = new NioEventLoopGroup(1);
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(group)；

上面的情况为ServerSocketChannel和SocketChannel共用一个EventLoopGroup，并且是单线程的

单reactor、多线程模型如下

1
2
3
EventLoopGroup group = new NioEventLoopGroup(8);
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(group)；

上面的情况为ServerSocketChannel和SocketChannel共用一个EventLoopGrop，但是是多线程的

多reactor、多线程模型如下

1
2
3
4
EventLoopGroup bossGroup = new NioEventLoopGroup(8);
EventLoopGroup workerGroup = new NioEventLoopGroup(8);
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)；

上面的情况为ServerSocketChannel和SocketChannel使用不同的EventLoopGroup，并且是多线程的

ChannelHandler

ChannelHandler是Netty处理IO事件的核心组件，负责数据的编解码、业务逻辑处理等，比如上面介绍的

NioServerSocketChannel和NioSocketChannel添加的事件处理器，并且可以添加多个，ChannelHandler可以被多个Channel添加，和Channel是多对多的关系

可以理解为消息处理器，有两种：ChannelInboundHandler、ChannelOutboundHandler，前者是入站消息处理，当有读事件时，会生效，后者是出战消息处理，当由写消息时，会生效，此外还有一种特殊的双通道处理：ChannelDuplexHandler，读写事件它都会生效，因为他同时实现了上面两种Handler

ChannelPipeline

ChannelPipeline则是ChannelHandler的容器，它维护了一个ChannelHandler的有序列表，负责将事件传递给正确的ChannelHandler，并确保它们按照既定的顺序执行，它也是直接和channel和handler之间的桥梁

由多个ChannelHandler组成，当有IO读写事件时，如果是读事件，消息会先到达HeadContext（ChannelPipeline的头部），然后一直往下走，经过各种消息处理器，如果是写事件，当前消息处理器创建消息后，会往上走，到达HeadContext，然后发送出去，如果不是消息处理器创建的消息，而是channel直接创建的消息，就从管道尾部出发（TailContext），即入站消息从上往下，出战消息从下往上

ChannelHandlerContext

ChannelHandlerContext是ChannelHandler的上下文环境，提供了在ChannelHandler的业务代码中访问Channel、EventLoop、ChannelPipeline等组件的功能，它本身并不算组件

Codec

编解码器，一种特殊的ChannelHandler，因为netty本身只支持ByteBuf，所以需要编解码器将ByteBuf转为我们需要的类型，或者将我们生成的类型数据转为ByteBuf交给netty处理，它有两个主要的方法

  1. decode：负责解码，将ByteBuf转为我们需要的类型
  2. encode：负责编码，将我们生成的类型数据转为ByteBuf

codec常见的类有io.netty.handler.codec.http.HttpServerCodec

除了codec同时负责编码和解码以外，也有只负责其中一种的handler

  1. decoder(解码器)：具有decode方法，实现ChannelInboundHandler，比如io.netty.handler.codec.http.HttpRequestDecoder
  2. encoder(编码器)：具有encode方法，实现ChannelOutboundHandler，比如io.netty.handler.codec.http.HttpRequestEncoder

图解

画板

本站总访问量：次本站总访问数：人
本站由 Peng Tong 使用 Stellar 1.29.1 主题创建。
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议，转载请注明出处。