yyds网络编程Netty

1、Netty的简介

1.1、Netty 是基于 Java NIO 的client-server(客户端服务器)框架，使用 Netty 可以快速开发网络应用，Netty 提供了高层次的抽象来简化 TCP 和 UDP 服务器的编程，但是你仍然可以使用底层的 API。

1.2、Netty 的内部实现是很复杂的，但是 Netty 提供了简单易用的API从网络处理代码中解耦业务逻辑。Netty 是完全基于 NIO 实现的，所以整个 Netty 都是异步的。

1.3、Netty 是最流行的 NIO 框架，它已经得到成百上千的商业、商用项目验证，许多框架和开源组件的底层 rpc 都是使用的 Netty，比如我们常用的 Dubbo、RocketMQ、Elasticsearch、gRPC等都用到了Netty。
1.4、支持多种协议 如 FTP，SMTP，HTTP 以及各种二进制和基于文本的传统协议。

2、为什么要用Netty?

2.1 统一的 API，支持多种传输类型，阻塞和非阻塞的。
2.2 简单而强大的线程模型。
2.3 功能强大，预置了编解码器解决 TCP 粘包/拆包功能，支持多种主流协议；
2.4 真正的无连接数据包套接字支持。
2.5 比直接使用 Java 核心 API 有更高的吞吐量、更低的延迟、更低的资源消耗和更少的内存复制。
2.6 安全性不错，有完整的 SSL/TLS 以及 StartTLS 支持。
2.7 社区活跃，版本迭代周期短，发现的BUG可以被及时修复，同时，更多的新功能会加入。
2.8 成熟稳定，经历了大型项目的使用和考验，而且很多开源项目都使用到了 Netty， 比如我们经常接触的 Dubbo、RocketMQ 等等。
2.9 成熟、稳定，Netty修复了已经发现的所有JDK NIO BUG，业务开发人员不需要再为NIO的BUG而烦恼。

3、Netty的应用场景

3.1 作为 RPC 框架的网络通信工具
比如我们的微服务框架中，不同的服务之间经常需要相互调用，这个时候就需要RPC框架了。

3.2 实现一个实时通讯系统
使用 Netty 我们可以实现一个可以聊天功能，比如微信的实时通讯系统。

Netty是基于Java NIO实现的，不可少的要讲一下NIO的基本概念：

阻塞（Block）与非阻塞（Non-Block）

阻塞（Block）：需要等待缓冲区中的数据准备好过后才处理其他的事情，否则一直等待在那里。

非阻塞（Non-Block）：进程访问数据缓冲区的时候，如果数据没有准备好则直接返回，不会等待。如果数据已经准备好，也直接返回。

同步（Synchronous）与异步（Asynchronous）

同步（Synchronous）：客户端发出请求到服务端，无论服务端是否阻塞以及其他情况，只要服务端同步返回结果，就认定是同步。

异步（Asynchronous）：客户端发出请求到服务端，同步返回值中没有结果，Client只需要等待通知，收到通知后才能拉取结果，就是异步。

Java BIO与NIO的不同之处 BIO（传统IO）：

传统的 java.io 包，它是基于流模型实现的，交互的方式是同步、阻塞方式，也就是说在读入输入流或者输出流时，在读写动作完成之前，线程会一直阻塞在那里，它们之间的调用时可靠的线性顺序。

NIO（Non-blocking/New I/O）：

NIO 是一种同步非阻塞的 I/O 模型，于 Java 1.4 中引入，对应 java.nio 包，提供了 Channel , Selector，Buffer 等抽象。NIO 中的 N 可以理解为 Non-blocking，不单纯是 New。它支持面向缓冲的，基于通道的 I/O 操作方法。NIO 提供了与传统 BIO 模型中的 Socket 和 ServerSocket 相对应的 SocketChannel 和 ServerSocketChannel 两种不同的套接字通道实现,两种通道都支持阻塞和非阻塞两种模式。对于高负载、高并发的（网络）应用，应使用 NIO 的非阻塞模式来开发。

Java NIO 和 BIO 之间第一个最大的区别是，BIO 是面向流的，NIO 是面向缓冲区的。

NIO的特点：

一个线程可以处理多个通道，减少线程创建数量，并且读或写不阻塞，就算没有数据可读写也不会发生阻塞导致线程资源的浪费。

Netty的执行流程：

接下来贴出Client 和Server代码讲解Netty 核心组件负责的主要功能。 Server代码：

   public void start(InetSocketAddress address) {
       //配置服务端的NIO线程组
       EventLoopGroup bossGroup = new NioEventLoopGroup(1);
       EventLoopGroup workerGroup = new NioEventLoopGroup();
       try {
           ServerBootstrap bootstrap = new ServerBootstrap()
                   .group(bossGroup, workerGroup)
                   .channel(NioServerSocketChannel.class)
                   .localAddress(address)
                   .childHandler(new NettyServerChannelInitializer())
                   .option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(1024*1024))
                   .childOption(ChannelOption.SO_KEEPALIVE, true);
           // 绑定端口，开始接收进来的连接
           ChannelFuture future = bootstrap.bind(address).sync();
           log.info("netty服务器开始监听端口：" + address.getPort());
           //关闭channel和块，直到它被关闭
           future.channel().closeFuture().sync();
       } catch (Exception e) {
           e.printStackTrace();
       } finally {
           //关闭主线程组
           bossGroup.shutdownGracefully();
           //关闭工作线程组
           workerGroup.shutdownGracefully();
       }
   }

Client代码：

    public void Client(InetSocketAddress address) {
     NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
                            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
                            pipeline.addLast(new DemoSocketClientHandler());
                        }
                    });

            ChannelFuture future = bootstrap.connect("localhost", 8888).sync();
            future.channel().closeFuture().sync();
        } finally {
            if(eventLoopGroup != null) {
                eventLoopGroup.shutdownGracefully();
            }
        }
    }
    }

Channel：

Channel 接口是 Netty 对网络操作抽象类，它除了包括基本的 I/O 操作，如 bind()、connect()、read()、write() 等。

比较常用的Channel接口实现类是NioServerSocketChannel（服务端）和NioSocketChannel（客户端），这两个 Channel 可以和 BIO 编程模型中的ServerSocket以及Socket两个概念对应上。Netty 的 Channel 接口所提供的 API，大大地降低了直接使用 Socket 类的复杂性。

而且我们对数据的读取和写入要通过Channel管道，它就像水管一样，是一个通道。通道不同于流的地方就是通道是双向的，可以用于读、写和同时读写操作。

EventLoop：

主要作用实际就是负责监听网络事件并调用事件处理器进行相关 I/O 操作的处理。

EventloopGroup：

顾名思义就是EventLoop的组。

EventLoopGroup 包含多个 EventLoop，每一个 EventLoop 通常内部包含一个线程，上面我们已经说了 EventLoop 的主要作用实际就是负责监听网络事件并调用事件处理器进行相关 I/O 操作的处理，并且 EventLoop 处理的 I/O 事件都将在它专有的 Thread 上被处理，即 Thread 和 EventLoop 属于 1 : 1 的关系，从而保证线程安全。

从上图结合以下的代码图可以看出：

其中 bossGroup 用于接收连接，workerGroup 用于具体的处理（消息的读写以及其他逻辑处理）。
当客户端通过 connect 方法连接服务端时，bossGroup 处理客户端连接请求。当客户端处理完成后，会将这个连接提交给 workerGroup 来处理，然后 workerGroup 负责处理其 IO 相关操作。

NioEventLoopGroup 初始化分析

讲解之前先讲一下EventLoopGroup 和NioEventLoopGroup 到底是什么关系，如下图

从上图中可以看出，在Netty中，EventLoopGroup和NioEventLoopGroup都是一个线程池，Netty中的EventLoopGroup接口直接继承JDK中的EventExecutorGroup接口。Netty中的NioEventLoopGroup是继承了MultithreadEventLoopGroup抽象类。

而MultithreadEventLoopGroup继承MultithreadEventExecutorGroup抽象类。MultithreadEventExecutorGroup抽象类实现了Netty中的EventLoopGroup接口。如果还不明白的话可以看一下继承关系：

这样是否更好的理解呢

下面具体分析，从 NioEventLoopGroup 的初始化进入源码分析

    
    public NioEventLoopGroup() {
        this(0);
    }
   
    public NioEventLoopGroup(int nThreads) {
     // 第二个参数是这个group所包含的executor
        this(nThreads, (Executor)null);
    }
   
    public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory) {
        this(nThreads, threadFactory, SelectorProvider.provider());
    }

    public NioEventLoopGroup(int nThreads, Executor executor) {
       // 第三个参数是provider，其用于提供selector及selectable的channel，
        // 这个provider是当前JVM中唯一的一个单例的provider
        this(nThreads, executor, SelectorProvider.provider());
    }
  public NioEventLoopGroup(
            int nThreads, Executor executor, final SelectorProvider selectorProvider) {
        // 第四个参数是一个选择策略工厂实例
        this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
    }
    }

构造函数中指定的每个参数含义：

nThreads ：DEFAULT_EVENT_LOOP_THREADS//默认当前cpu的核数的２
倍。

SelectorProvider.provider()：

    public static SelectorProvider provider() {
        synchronized (lock) {
            if (provider != null)
                return provider;
            return AccessController.doPrivileged(
                new PrivilegedAction() {
                    public SelectorProvider run() {
                           // 1. java.nio.channels.spi.SelectorProvider 属性指定实现类
                            if (loadProviderFromProperty())
                                return provider;
                                    // 2. SPI 指定实现类
                            if (loadProviderAsService())
                                return provider;
                            provider = sun.nio.ch.DefaultSelectorProvider.create();
                            return provider;
                        }
                    });
        }
    }

从代码中可以看出这是一个单例的provider。

SelectStrategyFactory:DefaultSelectStrategyFactory.INSTANCE//默认选择策略工厂实例。
chooserFactory:DefaultEventExecutorChooserFactory.INSTANCE//选择器工厂实例

这里主要的是重点讲一下MultithreadEventExecutorGroup构造函数

   protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {
        this.terminatedChildren = new AtomicInteger();
        this.terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
        if (nThreads <= 0) {
            throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
        } else {
          //如果没有指定线程工厂，那么构造一个默认的线程工厂
            if (executor == null) {
             // 这个executor是group所包含的executor，其将来会为其所包含的每个eventLoop创建一个线程
                executor = new ThreadPerTaskExecutor(this.newDefaultThreadFactory());
            }

            this.children = new EventExecutor[nThreads];

            int j;
            for(int i = 0; i < nThreads; ++i) {
                boolean success = false;
                boolean var18 = false;

                try {
                    var18 = true;
                     // 创建eventLoop
                    this.children[i] = this.newChild((Executor)executor, args);
                       // 在NioEventLoopGroup中，chlidren的每个元素其实都是一个NioEventLoop
                    success = true;
                    var18 = false;
                } catch (Exception var19) {
                    throw new IllegalStateException("failed to create a child event loop", var19);
                } finally {
                    if (var18) {
                      // 在创建这些eventLoop过程中，只要有一个创建失败，则关闭之前所有已经创建好的eventLoop
                        if (!success) {
                            int j;
                            for(j = 0; j < i; ++j) {
                                // 关闭之前所有已经创建好的eventLoop
                                this.children[j].shutdownGracefully();
                            }
      // 终止所有eventLoop上所执行的任务
                            for(j = 0; j < i; ++j) {
                                EventExecutor e = this.children[j];

                                try {
                                    while(!e.isTerminated()) {
                                        e.awaitTermination(2147483647L, TimeUnit.SECONDS);
                                    }
                                } catch (InterruptedException var20) {
                                    Thread.currentThread().interrupt();
                                    break;
                                }
                            }
                        }

                    }
                }

                if (!success) {
                    for(j = 0; j < i; ++j) {
                        this.children[j].shutdownGracefully();
                    }

                    for(j = 0; j < i; ++j) {
                        EventExecutor e = this.children[j];

                        try {
                            while(!e.isTerminated()) {
                                e.awaitTermination(2147483647L, TimeUnit.SECONDS);
                            }
                        } catch (InterruptedException var22) {
                            Thread.currentThread().interrupt();
                            break;
                        }
                    }
                }
            }
    // 创建一个选择器
            this.chooser = chooserFactory.newChooser(this.children);
            FutureListener