本文介绍Java的网络编程, 如何从BIO走到NIO再走到今天的Netty

1. BIO

就是最基本的Socket操作, 可以随便找一本大学Java教材, 一般都会有

  1. public static void main(String[] args) throws IOException {
  2.     //...
  3.     ServerSocket server = new ServerSocket(8080);
  4.     while (true) {
  5.         Socket client = server.accept(); //主线程: 阻塞操作1
  6.         //一般情况下, 在accept之后会把这个client交给其他线程处理
  7.         InputStream inputStream = client.getInputStream(); //其他线程: 阻塞操作2
  8.     }
  9.     //... 省略若干代码
  10. }

2. 第一代 Java NIO

NIO的出现主要是为了解决BIO的问题
优点:

  1. 解决了BIO线程数量限制, 个位数的线程就可以处理大量的连接
  2. 将建立连接 和 接收数据进行了区分
  1. public static void main(String[] args) throws IOException {
  2.     LinkedList<SocketChannel> clientList = new LinkedList<>();
  3.     ServerSocketChannel ss = ServerSocketChannel.open();
  4.     ss.bind(new InetSocketAddress(8080));
  5.     ss.configureBlocking(false);//设置为非阻塞
  6.     while (true) {
  7.         //--------------------主线程----------------------------
  8.         //这里的代码块, 实际上就是可以用一个线程轮询, 是否有新的连接进入
  9.         SocketChannel client = ss.accept();//非阻塞, 可能返回null
  10.         if (client != null) {
  11.             //这里非空,就表示有新的连接进入
  12.             client.configureBlocking(false);//设置为非阻塞
  13.             clientList.add(client);
  14.         }
  15.         //--------------------主线程----------------------------
  17.         //--------------------工作线程(池)----------------------------
  18.         ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
  19.         //这里可以在其他线程池处理, 用一个或者几个线程疯狂遍历已经连接的客户端, 检查有没有新的数据进来
  20.         //当然这里的代码就是在主线程中遍历,并读取数据,也是可以的, 就是效率肯定没有另外拉个线程读取数据来得高
  21.         for (SocketChannel c : clientList) {//!!!!!!!!!!!!!!注意这里的轮询非常浪费CPU, 需要每个连接都去检查
  22.             int num = c.read(buffer);//!!!!!!!!!!! 从客户端读取数据, 虽然这一步不是阻塞的, 但是这一步会发生用户态到内核态的转换
  23.             if (num > 0) {//读得到就处理, 读不到就拉倒
  24.                 buffer.flip();
  25.                 byte[] aaa = new byte[buffer.limit()];
  26.                 buffer.get(aaa);
  27.                 String b = new String(aaa);
  28.                 System.out.println(b);
  29.                 buffer.clear();
  30.             }
  31.         }
  32.         //--------------------工作线程(池)----------------------------
  33.     }
  34. }

缺点:

  1. 建立连接的线程,和读取数据的线程, 都需要不停的轮询, 造成了大量的无效的操作, 浪费了大量CPU资源
  2. accept和read函数实际上会导致一次用户态到内核态的切换,也会浪费CPU

3. 第二代 Java NIO

使用selector的NIO:
用一条线程绑定多路复用器来处理所有连接(包括accept事件,read事件,write事件等)

优点:

  • 不在需要不停的轮询了, 当发生连接或者读取事件的时候, 是由多路复用器来通知我们, 节约了CPU资源; (这里可以选择 1.阻塞等待多路复用器 或者 2.轮询查看多路复用器)
  1. public class SelectorNIO_1 {
  3.     public static void main(String[] args) throws IOException {
  4.         SocketMultiplexSingleThreadV1 service = new SocketMultiplexSingleThreadV1();
  5.         service.start();
  6.     }
  9.     /**
  10.      * 单线程处理多路复用器的demo
  11.      */
  12.     static class SocketMultiplexSingleThreadV1 {
  14.         private ServerSocketChannel server = null;
  16.         private Selector selector = null;
  18.         private int port = 8080;
  20.         public void initServer() throws IOException {
  21.             server = ServerSocketChannel.open();
  22.             server.configureBlocking(false);//非阻塞
  23.             server.bind(new InetSocketAddress(port));
  24.             selector = Selector.open();//创建一个多路复用器
  25.             server.register(selector, SelectionKey.OP_ACCEPT);//向多路复用器 注册server, 并且设置监听ACCEPT事件
  26.         }
  28.         public void start() throws IOException {
  29.             //1. 启动服务器,同时注册号多路复用器
  30.             initServer();
  31.             while (true) {
  32.                 //2. 从多路复用器中拿需要处理的文件描述符
  33.                 int readyNum = selector.select();//会阻塞,直到有响应事件, 这个阻塞是针对多路复用器的,要注意!!
  34.                 //select()：阻塞到至少有一个通道在你注册的事件上就绪了。
  35.                 //select(long timeout)：和select()一样，但最长阻塞时间为timeout毫秒。
  36.                 //selectNow()：非阻塞，只要有通道就绪就立刻返回。
  37.                 if (readyNum > 0) {
  38.                     //3. 遍历需要处理的key
  39.                     Set<SelectionKey> selectionKeys = selector.selectedKeys();//获取有事件的key, 这个实际上拿到的是多路复用器中的一个成员变量, 搞完只要要把key remove出去, 非常麻烦
  40.                     Iterator<SelectionKey> iterator = selectionKeys.iterator();
  41.                     while (iterator.hasNext()) {
  42.                         SelectionKey key = iterator.next();
  43.                         iterator.remove();//这里需要remove掉, 如果不remove掉,下次循环还会遍历到这个key, 设计非常傻逼
  44.                         //4. 处理这个key
  45.                         if (key.isAcceptable()) {
  46.                             acceptHandler(key);//建立连接操作
  47.                         } else if (key.isReadable()) {
  48.                             readHandler(key);//读取数据曹邹
  49.                         }
  50.                     }
  51.                 }
  52.             }
  53.         }
  55.         //监听到客户端的连接请求事件之后
  56.         private void acceptHandler(SelectionKey key) throws IOException {
  57.             ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
  58.             //1. 创建连接
  59.             SocketChannel client = ssc.accept();//建立连接
  60.             client.configureBlocking(false);
  61.             ByteBuffer buffer = ByteBuffer.allocateDirect(8192);
  62.             //2. 将这个连接和一个buffer绑定到一起, buffer当做key的attachment存进去
  63.             //3. 向多路复用器注册监听 该连接的读取事件
  64.             SelectionKey selectionKey = client.register(selector, SelectionKey.OP_READ, buffer);//向多路复用器绑定,读取事件
  65.             //这个selectionKey和到时候发生读取事件读取的key是一个对象, 这里暂时没用它
  66.             System.out.println("新客户端: " + client.getRemoteAddress());
  67.         }
  69.         private void readHandler(SelectionKey key) throws IOException {
  70.             SocketChannel client = (SocketChannel) key.channel();
  71.             ByteBuffer buffer = (ByteBuffer) key.attachment();//这个就是在建立连接的时候,当做attachment存进去的buffer
  72.             buffer.clear();
  73.             int read;
  74.             while (true) {
  75.                 read = client.read(buffer);
  76.                 if (read > 0) {
  77.                     buffer.flip();
  78.                     while (buffer.hasRemaining()) {
  79.                         //这里是把数据写回客户端
  80.                         //这里可以写其他的自己的业务逻辑
  81.                         client.write(buffer);
  82.                     }
  83.                     buffer.clear();
  84.                 } else if (read == 0) {
  85.                     break;
  86.                 } else { //-1 可能是客户端关闭了, close_wait
  87.                     client.close();
  88.                     break;
  89.                 }
  90.             }
  91.         }
  92.     }
  93. }

缺点:

  1. 由于只有一个线程在处理连接, 当我们线程处理读写事件的时候, 没法立刻对连接事件作出响应
  2. 多个读写事件都在等待一个线程处理, 也就无法将网卡的IO能力完全发挥出来

4. 第三代Java NIO

使用Selector的NIO
�1个boss线程专门处理accept事件
n个worker线程专门处理读写事件

�优点:

  1. 1个线程专门处理ACCEPT事件, 对新连接的响应非常快
  2. 专门有n个线程, 来处理读写事件, 效率非常的高
  3. 思想上, 这种设计是能够最大的压榨机器性能的
  1. public class SelectorNIO_2 {
  3.     public static void main(String[] args) throws IOException, InterruptedException {
  4.         //new的时候,就创建了boss线程和worker线程,只是没有start
  5.         SocketMultiplexingThreads socketMultiplexingThreads = new SocketMultiplexingThreads(9090, 4);
  6.         //start boss线程和worker线程, 此时可以接收请求了
  7.         socketMultiplexingThreads.start();
  8.     }
  10.     //包装了boss线程和worker线程的创建初始化过程
  11.     static class SocketMultiplexingThreads {
  13.         //服务
  14.         private ServerSocketChannel server;
  16.         //boss线程
  17.         private BossThread bossThread;
  19.         //worker线程
  20.         private List<WorkerThread> workerThreads;
  22.         //服务绑定的端口号
  23.         private int port;
  25.         //构造方法, 同时初始化server, bossThread, workerThread
  26.         public SocketMultiplexingThreads(int port, int workerNum) throws IOException {
  27.             this.port = port;
  28.             server = ServerSocketChannel.open();
  29.             server.configureBlocking(false);//非阻塞
  30.             server.bind(new InetSocketAddress(this.port));
  31.             this.bossThread = new BossThread(server, this);
  32.             //创建boss线程
  33.             workerThreads = new ArrayList<>(workerNum);
  34.             for (int i = 0; i < workerNum; i++) {
  35.                 //创建worker线程
  36.                 workerThreads.add(new WorkerThread(server));
  37.             }
  38.         }
  40.         //粗糙的启动方法
  41.         public void start() throws InterruptedException {
  42.             bossThread.start();
  43.             for (WorkerThread workerThread : workerThreads) {
  44.                 workerThread.start();
  45.             }
  46.             Thread.sleep(1000 * 60 * 60);
  47.         }
  49.         //随机获取一个worker线程
  50.         WorkerThread randomGetWorkerThread() {
  51.             Random random = new Random();
  52.             int index = random.nextInt(workerThreads.size());
  53.             return workerThreads.get(index);
  54.         }
  55.     }
  57.     //boss线程只处理accept事件
  58.     static class BossThread extends Thread {
  60.         private Selector bossSelector;
  61.         private SocketMultiplexingThreads socketMultiplexingThreads;
  63.         public BossThread(ServerSocketChannel server, SocketMultiplexingThreads socketMultiplexingThreads) throws IOException {
  64.             this.socketMultiplexingThreads = socketMultiplexingThreads;
  65.             //创建Boss多路复用器
  66.             this.bossSelector = Selector.open();
  67.             //向主线程注册ACCEPT事件
  68.             server.register(bossSelector, SelectionKey.OP_ACCEPT);//向多路复用器 注册server, 并且设置监听ACCEPT事件
  69.             System.out.println("Boss 线程创建完毕");
  70.         }
  72.         @lombok.SneakyThrows
  73.         @Override
  74.         public void run() {
  75.             //boss线程不停的轮询, 等待新的连接, 专门处理accept事件
  76.             while (true) {
  77.                 int readyNum = bossSelector.select(100);
  78.                 if (readyNum > 0) {
  79.                     Set<SelectionKey> selectionKeys = bossSelector.selectedKeys();//获取有事件的key, 这个实际上拿到的是多路复用器中的一个成员变量, 搞完只要要把key remove出去, 非常麻烦
  80.                     Iterator<SelectionKey> iterator = selectionKeys.iterator();
  81.                     while (iterator.hasNext()) {
  82.                         System.out.println("boss 线程处理事件");
  83.                         SelectionKey key = iterator.next();
  84.                         iterator.remove();//这里需要remove掉, 如果不remove掉,下次循环还会遍历到这个key, 设计非常傻逼
  85.                         //Boss线程只处理Accept事件
  86.                         if (key.isAcceptable()) {
  87.                             acceptHandler(key);//建立连接操作
  88.                         }
  89.                     }
  90.                 }
  91.             }
  92.         }
  94.         //处理accept事件,将读事件绑定到worker线程上去
  95.         private void acceptHandler(SelectionKey key) throws IOException {
  96.             ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
  97.             //1. 创建连接
  98.             SocketChannel client = ssc.accept();//建立连接
  99.             client.configureBlocking(false);
  100.             //2. 分配buffer给这个连接, 其实这里如果把分配内存放到worker线程中去做, boss能处理更多的连接, 这里就不演示了, 不然代码就不清晰了
  101.             ByteBuffer buffer = ByteBuffer.allocateDirect(8192);
  102.             //3. 获取到worker线程的多路复用器
  103.             WorkerThread workerThread = socketMultiplexingThreads.randomGetWorkerThread();//这里用的是随机获取一个, 算法可以自己实现
  104.             Selector workerSelector = workerThread.getWorkerSelector();
  105.             //4. 向多路复用器注册监听 该连接的读取事件
  106.             SelectionKey selectionKey = client.register(workerSelector, SelectionKey.OP_READ, buffer);//向多路复用器绑定,读取事件
  107.             //这个selectionKey和到时候发生读取事件读取的key是一个对象, 这里暂时没用它
  108.             System.out.println("向worker线程注册读取事件: " + client.getRemoteAddress());
  109.         }
  110.     }
  112.     //worker线程只处理读写事件
  113.     static class WorkerThread extends Thread {
  115.         private Selector workerSelector;
  116.         private static AtomicInteger accumulator = new AtomicInteger(0);
  117.         private int id = accumulator.getAndIncrement();//作为线程的标识
  119.         public WorkerThread(ServerSocketChannel server) throws IOException {
  120.             //创建Worker多路复用器
  121.             this.workerSelector = Selector.open();
  122.             System.out.println(this + " worker 线程创建完毕");
  123.         }
  125.         public Selector getWorkerSelector() {
  126.             return this.workerSelector;
  127.         }
  129.         @SneakyThrows
  130.         @Override
  131.         public void run() {
  132.             //worker线程不停的轮询, 等待新的连接, 专门处理读事件
  133.             while (true) {
  134.                 int readyNum = workerSelector.select(100);
  135.                 if (readyNum > 0) {
  136.                     Set<SelectionKey> selectionKeys = workerSelector.selectedKeys();//获取有事件的key, 这个实际上拿到的是多路复用器中的一个成员变量, 搞完只要要把key remove出去, 非常麻烦
  137.                     Iterator<SelectionKey> iterator = selectionKeys.iterator();
  138.                     while (iterator.hasNext()) {
  139.                         System.out.println(this + " worker 线程处理事件");
  140.                         SelectionKey key = iterator.next();
  141.                         iterator.remove();//这里需要remove掉, 如果不remove掉,下次循环还会遍历到这个key, 设计非常傻逼
  142.                         //Boss线程只处理Accept事件
  143.                         if (key.isReadable()) {
  144.                             readHandler(key);//建立连接操作
  145.                         }
  146.                     }
  147.                 }
  148.             }
  149.         }
  151.         //worker线程处理读写事件
  152.         private void readHandler(SelectionKey key) throws IOException {
  153.             SocketChannel client = (SocketChannel) key.channel();
  154.             ByteBuffer buffer = (ByteBuffer) key.attachment();//这个就是在建立连接的时候,当做attachment存进去的buffer
  155.             buffer.clear();
  156.             int read;
  157.             while (true) {
  158.                 read = client.read(buffer);
  159.                 if (read > 0) {
  160.                     buffer.flip();
  161.                     while (buffer.hasRemaining()) {
  162.                         //这里是把数据写回客户端
  163.                         //这里可以写其他的自己的业务逻辑
  164.                         client.write(buffer);
  165.                     }
  166.                     buffer.clear();
  167.                 } else if (read == 0) {
  168.                     break;
  169.                 } else { //-1 可能是客户端关闭了, close_wait
  170.                     client.close();
  171.                     break;
  172.                 }
  173.             }
  174.         }
  176.         @Override
  177.         public String toString() {
  178.             return "worker线程[" + id + "]";
  179.         }
  180.     }
  181. }

缺点:

  1. 需要自己直接使用NIO的接口, 那么意味着可能会遇到各种坑
  2. 已经有Netty这种成熟框架包装了NIO的接口, 没有必要自己再写一套基于NIO的框架(但是方便帮助自己理解netty的思想)

5. Netty的使用

  • Netty对JDK的NIO又再进行了一次包装, 底层还是调用JDK的NIO接口进行操作
  • Netty的设计哲学是将所有网络操作抽象成 事件, 再将事件交给 EventLoop 去处理; (包括建立连接,读,写 都是事件)

  • 让网络编程的程序员可以更聚焦于业务的处理

    5.1 单线程Netty

  • 用一个线程去处理所有 事件 ```java public class NettyIO_1 {

    public static void main(String[] args) throws InterruptedException {

    1.   //boss线程只设置1个
    2.   NioEventLoopGroup boss = new NioEventLoopGroup(1);//group可以设置多个线程,但是我们这只设置了1个
    3.   ServerBootstrap boot = new ServerBootstrap();
    4.   //注意,这里特地写成两个boss
    5.   //意味着: boss线程不但处理accept事件,也处理读写事件
    6.   boot.group(boss, boss)
    7.           .channel(NioServerSocketChannel.class)
    8.           .childHandler(new ChannelInitializer<NioSocketChannel>() {
    9.               @Override
    10.               protected void initChannel(NioSocketChannel channel) throws Exception {
    11.                   ChannelPipeline pipeline = channel.pipeline();
    12.                   pipeline.addLast(new MyInbound());
    13.               }
    14.           });
    15.   ChannelFuture future = boot.bind(8080).sync();
    16.   future.channel().closeFuture().sync();

    }

    static class MyInbound extends ChannelInboundHandlerAdapter {

    1.   @Override
    2.   public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    3.       //todo
    4.   }
    6.   @Override
    7.   public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    8.       //todo
    9.   }

    }

}

  2. <a name="anM7H"></a>
  3. ## 5.2 多线程Netty
  4. ```java
  5. public class NettyIO_2 {
  7.     public static void main(String[] args) throws InterruptedException {
  8.         //boss线程设置4个
  9.         //线程既是boss也是worker
  10.         NioEventLoopGroup boss = new NioEventLoopGroup(4);
  11.         ServerBootstrap boot = new ServerBootstrap();
  12.         boot.group(boss, boss)
  13.                 .channel(NioServerSocketChannel.class)
  14.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  15.                     @Override
  16.                     protected void initChannel(NioSocketChannel channel) throws Exception {
  17.                         ChannelPipeline pipeline = channel.pipeline();
  18.                         pipeline.addLast(new MyInbound());
  19.                     }
  20.                 });
  21.         ChannelFuture future = boot.bind(8080).sync();
  22.         future.channel().closeFuture().sync();
  23.     }
  25.     static class MyInbound extends ChannelInboundHandlerAdapter {
  26.         @Override
  27.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  28.             ctx.write(msg);//echo的作用, 将读到的东西,写回客户端
  29.         }
  31.         @Override
  32.         public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  33.             ctx.flush();
  34.         }
  35.     }
  37. }

5.3 Netty的经典使用

  • 在多线程的基础上,将线程分为boss和worker
  • boss 专门用来处理连接事件, 只有一个线程

  • worker 专门用来处理读写事件, 可以有多个线程 ```java public class NettyIO_3 {

    public static void main(String[] args) throws InterruptedException {

    1.      //bossGroup, 专门处理连接事件
    2.   NioEventLoopGroup boss = new NioEventLoopGroup(99);//即使这里传入了99, 会创建99个EventLoop对象, 但是线程被使用的只有一个, 也就是说只有一个线程处理ACCEPT事件
    3.   //workerGroup 专门处理读写事件
    4.   NioEventLoopGroup worker = new NioEventLoopGroup(20);//20个线程用来处理读写事件
    5.   ServerBootstrap boot = new ServerBootstrap();
    6.   boot.group(boss, worker)
    7.           .channel(NioServerSocketChannel.class)
    8.           .childHandler(new ChannelInitializer<NioSocketChannel>() {
    9.               @Override
    10.               protected void initChannel(NioSocketChannel channel) throws Exception {
    11.                   ChannelPipeline pipeline = channel.pipeline();
    12.                   pipeline.addLast(new MyInbound());
    13.               }
    14.           });
    15.   ChannelFuture future = boot.bind(8080).sync();
    16.   future.channel().closeFuture().sync();

    }

    static class MyInbound extends ChannelInboundHandlerAdapter {

    1.   @Override
    2.   public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    3.       ctx.write(msg);//echo的作用, 将读到的东西,写回客户端
    4.   }
    6.   @Override
    7.   public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    8.       ctx.flush();
    9.   }

    }

} ```