完全搞懂 netty 线程模型

时间 2019-11-17

标签完全 netty 线程模型栏目 Netty 繁體版

原文原文链接

编者注：Netty是Java领域有名的开源网络库，特色是高性能和高扩展性，所以不少流行的框架都是基于它来构建的，好比咱们熟知的Dubbo、Rocketmq、Hadoop等。本文就netty线程模型展开分析讨论下 : )java

IO模型react

BIO：同步阻塞IO模型；
NIO：基于IO多路复用技术的“非阻塞同步”IO模型。简单来讲，内核将可读可写事件通知应用，由应用主动发起读写操做；
AIO：非阻塞异步IO模型。简单来讲，内核将读完成事件通知应用，读操做由内核完成，应用只需操做数据便可；应用作异步写操做时当即返回，内核会进行写操做排队并执行写操做。

NIO和AIO不一样之处在于应用是否进行真正的读写操做。linux

reactor和proactor模型nginx

reactor：基于NIO技术，可读可写时通知应用；
proactor：基于AIO技术，读完成时通知应用，写操做应用通知内核。

netty线程模型

netty的线程模型是基于Reactor模型的。程序员

netty单线程模型

Reactor 单线程模型，是指全部的 I/O 操做都在同一个 NIO 线程上面完成的，此时NIO线程职责包括：接收新建链接请求、读写操做等。
面试

在一些小容量应用场景下，可使用单线程模型（注意，Redis的请求处理也是单线程模型，为何Redis的性能会如此之高呢？由于Redis的读写操做基本都是内存操做，而且Redis协议比较简洁，序列化/反序列化耗费性能更低）。可是对于高负载、大并发的应用场景却不合适，主要缘由以下：数据库

一个NIO线程同时处理成百上千的链接，性能上没法支撑，即使NIO线程的CPU负荷达到100%，也没法知足海量消息的编码、解码、读取和发送。
当NIO线程负载太重以后，处理速度将变慢，这会致使大量客户端链接超时，超时以后每每会进行重发，这更加剧了NIO线程的负载，最终会致使大量消息积压和处理超时，成为系统的性能瓶颈。
可靠性问题：一旦NIO线程意外跑飞，或者进入死循环，会致使整个系统通讯模块不可用，不能接收和处理外部消息，形成节点故障。

Reactor多线程模型

Rector 多线程模型与单线程模型最大的区别就是有一组 NIO 线程来处理链接读写操做，一个NIO线程处理Accept。一个NIO线程能够处理多个链接事件，一个链接的事件只能属于一个NIO线程。后端

在绝大多数场景下，Reactor 多线程模型能够知足性能需求。可是，在个别特殊场景中，一个 NIO 线程负责监听和处理全部的客户端链接可能会存在性能问题。例如并发百万客户端链接，或者服务端须要对客户端握手进行安全认证，可是认证自己很是损耗性能。在这类场景下，单独一个 Acceptor 线程可能会存在性能不足的问题，为了解决性能问题，产生了第三种 Reactor 线程模型——主从Reactor 多线程模型。安全

Reactor主从多线程模型

主从 Reactor 线程模型的特色是：服务端用于接收客户端链接的再也不是一个单独的 NIO 线程，而是一个独立的 NIO 线程池。Acceptor 接收到客户端 TCP链接请求并处理完成后（可能包含接入认证等），将新建立的 SocketChannel注册到 I/O 线程池（sub reactor 线程池）的某个I/O线程上，由它负责SocketChannel 的读写和编解码工做。Acceptor 线程池仅仅用于客户端的登陆、握手和安全认证，一旦链路创建成功，就将链路注册到后端 subReactor 线程池的 I/O 线程上，由 I/O 线程负责后续的 I/O 操做。网络

netty线程模型思考

netty 的线程模型并非一成不变的，它实际取决于用户的启动参数配置。经过设置不一样的启动参数，Netty 能够同时支持 Reactor 单线程模型、多线程模型。

为了尽量地提高性能，Netty 在不少地方进行了无锁化的设计，例如在 I/O 线程内部进行串行操做，避免多线程竞争致使的性能降低问题。表面上看，串行化设计彷佛 CPU 利用率不高，并发程度不够。可是，经过调整 NIO 线程池的线程参数，能够同时启动多个串行化的线程并行运行，这种局部无锁化的串行线程设计相比一个队列多个工做线程的模型性能更优。（小伙伴们后续多线程并发流程可参考该类实现方案）

Netty 的 NioEventLoop 读取到消息以后，直接调用 ChannelPipeline 的fireChannelRead (Object msg)。只要用户不主动切换线程，一直都是由NioEventLoop 调用用户的 ChannelHandler，期间不进行线程切换。这种串行化处理方式避免了多线程操做致使的锁的竞争，从性能角度看是最优的。

Netty拥有两个NIO线程池，分别是bossGroup和workerGroup，前者处理新建链接请求，而后将新创建的链接轮询交给workerGroup中的其中一个NioEventLoop来处理，后续该链接上的读写操做都是由同一个NioEventLoop来处理。注意，虽然bossGroup也能指定多个NioEventLoop（一个NioEventLoop对应一个线程），可是默认状况下只会有一个线程，由于通常状况下应用程序只会使用一个对外监听端口。

这里试想一下，难道不能使用多线程来监听同一个对外端口么，即多线程epoll_wait到同一个epoll实例上？

epoll相关的主要两个方法是epoll_wait和epoll_ctl，多线程同时操做同一个epoll实例，那么首先须要确认epoll相关方法是否线程安全：简单来讲，epoll是经过锁来保证线程安全的, epoll中粒度最小的自旋锁ep->lock(spinlock)用来保护就绪的队列, 互斥锁ep->mtx用来保护epoll的重要数据结构红黑树。

看到这里，可能有的小伙伴想到了Nginx多进程针对监听端口的处理策略，Nginx是经过accept_mutex机制来保证的。accept_mutex是nginx的(新建链接)负载均衡锁，让多个worker进程轮流处理与client的新链接。当某个worker进程的链接数达到worker_connections配置（单个worker进程的最大处理链接数）的最大链接数的7/8时，会大大减少获取该worker获取accept锁的几率，以此实现各worker进程间的链接数的负载均衡。accept锁默认打开，关闭它时nginx处理新建链接耗时会更短，可是worker进程之间可能链接不均衡，而且存在“惊群”问题。只有在使能accept_mutex而且当前系统不支持原子锁时，才会用文件实现accept锁。注意，accept_mutex加锁失败时不会阻塞当前线程，相似tryLock。

现代linux中，多个socker同时监听同一个端口也是可行的，nginx 1.9.1也支持这一行为。linux 3.9以上内核支持SO_REUSEPORT选项，容许多个socker bind/listen在同一端口上。这样，多个进程能够各自申请socker监听同一端口，当链接事件来临时，内核作负载均衡，唤醒监听的其中一个进程来处理，reuseport机制有效的解决了epoll惊群问题。

再回到刚才提出的问题，java中多线程来监听同一个对外端口，epoll方法是线程安全的，这样就可使用使用多线程监听epoll_wait了么，固然是不建议这样干的，除了epoll的惊群问题以外，还有一个就是，通常开发中咱们使用epoll设置的是LT模式（水平触发方式，与之相对的是ET默认，前者只要链接事件未被处理就会在epoll_wait时始终触发，后者只会在真正有事件来时在epoll_wait触发一次），这样的话，多线程epoll_wait时就会致使第一个线程epoll_wait以后还未处理完毕已发生的事件时，第二个线程也会epoll_wait返回，显然这不是咱们想要的，关于java nio的测试demo以下：

public class NioDemo {
    private static AtomicBoolean flag = new AtomicBoolean(true);
    public static void main(String[] args) throws Exception {
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.socket().bind(new InetSocketAddress(8080));
        // non-block io
        serverChannel.configureBlocking(false);
        Selector selector = Selector.open();
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        // 多线程执行
        Runnable task = () -> {
            try {
                while (true) {
                    if (selector.select(0) == 0) {
                        System.out.println("selector.select loop... " + Thread.currentThread().getName());
                        Thread.sleep(1);
                        continue;
                    }

                    if (flag.compareAndSet(true, false)) {
                        System.out.println(Thread.currentThread().getName() + " over");
                        return;
                    }

                    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();

                        // accept event
                        if (key.isAcceptable()) {
                            handlerAccept(selector, key);
                        }

                        // socket event
                        if (key.isReadable()) {
                            handlerRead(key);
                        }

                        /**
                         * Selector不会本身从已选择键集中移除SelectionKey实例，必须在处理完通道时手动移除。
                         * 下次该通道变成就绪时，Selector会再次将其放入已选择键集中。
                         */
                        iter.remove();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        };

        List<Thread> threadList = new ArrayList<>();
        for (int i = 0; i < 2; i++) {
            Thread thread = new Thread(task);
            threadList.add(thread);
            thread.start();
        }
        for (Thread thread : threadList) {
            thread.join();
        }
        System.out.println("main end");
    }

    static void handlerAccept(Selector selector, SelectionKey key) throws Exception {
        System.out.println("coming a new client... " + Thread.currentThread().getName());
        Thread.sleep(10000);
        SocketChannel channel = ((ServerSocketChannel) key.channel()).accept();
        channel.configureBlocking(false);
        channel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
    }

    static void handlerRead(SelectionKey key) throws Exception {
        SocketChannel channel = (SocketChannel) key.channel();
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.clear();

        int num = channel.read(buffer);
        if (num <= 0) {
            // error or fin
            System.out.println("close " + channel.getRemoteAddress());
            channel.close();
        } else {
            buffer.flip();
            String recv = Charset.forName("UTF-8").newDecoder().decode(buffer).toString();
            System.out.println("recv: " + recv);

            buffer = ByteBuffer.wrap(("server: " + recv).getBytes());
            channel.write(buffer);
        }
    }
}

netty线程模型实践

(1) 时间可控的简单业务直接在 I/O 线程上处理

时间可控的简单业务直接在 I/O 线程上处理，若是业务很是简单，执行时间很是短，不须要与外部网络交互、访问数据库和磁盘，不须要等待其它资源，则建议直接在业务 ChannelHandler 中执行，不须要再启业务的线程或者线程池。避免线程上下文切换，也不存在线程并发问题。

(2) 复杂和时间不可控业务建议投递到后端业务线程池统一处理

复杂度较高或者时间不可控业务建议投递到后端业务线程池统一处理，对于此类业务，不建议直接在业务 ChannelHandler 中启动线程或者线程池处理，建议将不一样的业务统一封装成 Task，统一投递到后端的业务线程池中进行处理。过多的业务ChannelHandler 会带来开发效率和可维护性问题，不要把 Netty 看成业务容器，对于大多数复杂的业务产品，仍然须要集成或者开发本身的业务容器，作好和Netty 的架构分层。

(3) 业务线程避免直接操做 ChannelHandler

业务线程避免直接操做 ChannelHandler，对于 ChannelHandler，IO 线程和业务线程均可能会操做，由于业务一般是多线程模型，这样就会存在多线程操做ChannelHandler。为了尽可能避免多线程并发问题，建议按照 Netty 自身的作法，经过将操做封装成独立的 Task 由 NioEventLoop 统一执行，而不是业务线程直接操做，相关代码以下所示：

若是你确认并发访问的数据或者并发操做是安全的，则无需画蛇添足，这个须要根据具体的业务场景进行判断，灵活处理。

推荐阅读

欢迎小伙伴关注【TopCoder】阅读更多精彩好文。