这是我参与11月更文挑战的第18天，活动详情查看：2021最后一次更文挑战

Java 网络IO API

Java 网络IO相关的API有三类：NIO、BIO、AIO、IO Multiplexing。
那么对应的IO如何使用到呢？

BIO

阻塞IO，在调用read和write时均会阻塞。read通常比较容易理解，当socket接受到数据read就会返回。那么write呢？write会阻塞我一开始也是迷迷糊糊，知道后来学到了 socket buffer这个数据结构才略懂一点，write API会将传入的byte 写到socket文件对应socket buffer中，这个数据结构类似一个队列或者链表，并且是有限的，当socket buffer容量不足时write API就会阻塞，知道有空闲空间。

在Java中，可以使用JDK提供的API使用BIO。下面是一个bio便携的client、server交互的例子。

java复制代码public class Server {
    public static void main(String[] args) {
        // 服务端占用端口
        int port = 8000;
        // 创建 serversocker
        ServerSocket serverSocket = null;
        try {
            serverSocket = new ServerSocket(port);
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (serverSocket != null) {
            while (true) {
                InputStream is = null;
                OutputStream os = null;
                Socket client = null;
                try {
                    // accept()方法会阻塞，直到有client连接后才会执行后面的代码
                    client = serverSocket.accept();
                    is = client.getInputStream();
                    os = client.getOutputStream();

                    // 3.server收到消息在控制台的打印，并回复"Hi client,I am Server."
                    byte[] buffer = new byte[5];
                    int len = 0;
                    // 使用ByteArrayOutputStream，避免缓冲区过小导致中文乱码
                    ByteArrayOutputStream baos = new ByteArrayOutputStream();
                    while ((len = is.read(buffer)) != -1) {
                        baos.write(buffer,0,len);
                    }
                    System.out.println(baos.toString());
                    // 服务端回复客户端消息
                    os.write("Hi client,I am Server.".getBytes());
                    os.flush(); // 刷新缓存，避免消息没有发送出去
                    client.shutdownOutput();
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    // 程序异常或者执行完成，关闭流，防止占用资源
                    if (client != null) {
                        try {
                            client.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    if (is != null) {
                        try {
                            is.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    if (os != null) {
                        try {
                            os.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }

            }
        }


    }
}


public class Client {
    public static void main(String[] args) {
        int port = 8000;
        Socket client = null;
        InputStream is = null;
        OutputStream os = null;
        try {
            // 1.client连接server
            client = new Socket("localhost", port);
            is = client.getInputStream();
            os = client.getOutputStream();

            // 2.client发送"Hi Server,I am client."
            os.write("Hi Server,I am client.".getBytes());
            os.flush();
            // 调用shutdownOutput()方法表示客户端传输完了数据，否则服务端的
            // read()方法会一直阻塞
            // (你可能会问我这不是写了 read()!=-1, -1表示的文本文件的结尾字符串，而对于字节流数据，
            // 是没有 -1 标识的，这就会使服务端无法判断客户端是否发送完成，导致read()方法一直阻塞)
            client.shutdownOutput();

            // 4.client收到消息在控制台打印。
            int len = 0;
            byte[] buffer = new byte[5];
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            while ((len = is.read(buffer)) != -1) {
                baos.write(buffer,0,len);
            }
            System.out.println(baos.toString());
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 程序异常或者执行完成，关闭流，防止占用资源
            try {
                if (is != null) {
                    is.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                if (os != null) {
                    os.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                if (client != null) {
                    client.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }


    }
}

NIO

BIO满足了应用层进行通信的基本需求，但是如果一直在accept、read、write这种地方干等着，CPU一直不被使用，岂不是很浪费。所以NIO出现了，NIO称作NO Block IO，其特点在于提供了非阻塞的accept，read，write。相当于Java中Lock 的tryLoack。当tryAccept返回失败，则表示没有socket建立连接，此时可以去干点别的。

但其实这个效果非常的鸡肋，假若你去干其它事情的时候，有client建立连接，你就无法及时响应，java中通过NIO java.nio.channels.spi.AbstractSelectableChannel#configureBlocking可以设置非阻塞模式，但是这个API只支持NIO的socket设置。

IO Multiplexing

BIO满足了应用层传输数据的基本需求，但在实际使用中，client一定是多个，并且还可能同时与一个server进行数据传输，所以在server端需要使用多线程的方式来接收客户端的请求。所以请求的并发数量会与机器的线程数量成正比，对外服务的性能也会受到操作系统分配给进程的最大线程数的影响。因为每个线程都需要阻塞到read、write操作，完全不能干点别的了。

操作系统的网络协议栈的开发者也觉得这个操作非常拉胯，如同LOL王者辅助带一个青铜ADC，任由你机器配置在牛，性能还是不行，熟话说好马配好鞍，经过操作系统开发者的反复努力，开发出了一个API，我们大众程序员调用这个API可以直接获取到哪些socket可以read，哪些socket可以write，哪些socket需要建立连接，这个直接获取的机制，称之为IO多路复用，其最大的意义在于单线程可以处理多个socket，IO多路复用的具体实现依赖于底层的操作系统，不同实现有：

select
poll
epoll

java中可以如下使用select实现的IO多路复用：

java复制代码public class NIOClient {

    public static void main(String[] args) {
        SocketAddress socketAddress = new InetSocketAddress(8000);
        SocketChannel socketChannel = null;
        try {
            socketChannel = SocketChannel.open(socketAddress);
            socketChannel.configureBlocking(false);
            if (socketChannel.finishConnect()) {

                ByteBuffer buffer = ByteBuffer.allocate(1024);
                // 客户端发送数据 "Hi Server,I am client."
                buffer.clear();
                buffer.put("Hi Server,I am client.".getBytes());
                buffer.flip();
                while (buffer.hasRemaining()) {
                    socketChannel.write(buffer);
                }
                // 客户端接收服务端数据打印在控制台

                buffer.clear();
                int len = socketChannel.read(buffer);
                while (len > 0) {
                    buffer.flip();
                    while (buffer.hasRemaining()) {
                        System.out.print((char) buffer.get());
                    }
                    System.out.println();
                    buffer.clear();
                    len = socketChannel.read(buffer);
                }
                if (len == -1) {
                    socketChannel.close();
                }
            }


        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (socketChannel != null) {
                    socketChannel.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}


public class NIOServer {


    public static void main(String[] args) {
        ServerSocketChannel serverSocketChannel = null;
        Selector selector = null;
        try {
            // 初始化一个 serverSocketChannel
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(8000));

            // 设置serverSocketChannel为非阻塞模式
            // 即 select()会立即得到返回
            serverSocketChannel.configureBlocking(false);

            // 初始化一个 selector
            selector = Selector.open();

            // 将 serverSocketChannel 与 selector绑定
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

            while (true) {
                // 通过操作系统监听变化的socket个数
                // 在windows平台通过selector监听(轮询所有的socket进行判断，效率低)
                // 在Linux2.6之后通过epool监听(事件驱动方式，效率高)
                int count = selector.select(3000);
                if (count > 0) {
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();

                        if (key.isAcceptable()) {
                            handleAccept(key);
                        }
                        if (key.isReadable()) {
                            handleRead(key);
                        }
                        if (key.isWritable() && key.isValid()) {
                            handleWrite(key);
                        }
                        if (key.isConnectable()) {
                            System.out.println("isConnectable = true");
                        }
                        iterator.remove();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (serverSocketChannel != null) {
                    serverSocketChannel.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                if (selector != null) {
                    selector.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private static void handleWrite(SelectionKey key) {
        // 获取 client 的 socket
        SocketChannel clientChannel = (SocketChannel) key.channel();
        // 获取缓冲区
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.clear();
        buffer.put("Hi client,I am Server.".getBytes());
        buffer.flip();
        try {
            while (buffer.hasRemaining()) {
                clientChannel.write(buffer);
            }
            buffer.compact();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void handleRead(SelectionKey key) {
        // 获取 readable 的客户端 socketChannel
        SocketChannel clientChannel = (SocketChannel) key.channel();
        // 读取客户端发送的消息信息,我们已经在 acceptable 中设置了缓冲区
        // 所以直接冲缓冲区读取信息
        ByteBuffer buffer = (ByteBuffer) key.attachment();

        // 获取 client 发送的消息
        try {
            int len = clientChannel.read(buffer);
            while (len > 0) {
                // 设置 limit 位置
                buffer.flip();
                // 开始读取数据
                while (buffer.hasRemaining()) {
                    byte b = buffer.get();
                    System.out.print((char) b);
                }
                System.out.println();
                // 清除 position 位置
                buffer.clear();
                // 从新读取 len
                len = clientChannel.read(buffer);
            }
            if (len == -1) {
                clientChannel.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }



    }

    private static void handleAccept(SelectionKey key) {
        // 获得 serverSocketChannel
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
        try {
            // 获得 socketChannel,就是client的socket
            SocketChannel clientChannel = serverSocketChannel.accept();
            if (clientChannel == null) return;
            // 设置 socketChannel 为无阻塞模式
            clientChannel.configureBlocking(false);
            // 将其注册到 selector 中，设置监听其是否可读，并分配缓冲区
            clientChannel.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocateDirect(512));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

JAVA NIO API相关资料：tutorials.jenkov.com/java-nio/in…

AIO

IO Multiplexing 以及非常牛皮了，但在实际使用中，感觉还是不够方便也不够快捷。因为作为开发者的我还需要主动查询哪些socket可以read、可以write、可以accept。

能不能更方便点，我事先给你个回调函数，操作系统你接收到cliet的数据久直接调用我这个函数，这样不是更方便，我什么也不用等，岂不是非常舒服呀呀呀呀！

贴心的操作系统开发者看到这个需要给你竖起了大拇指，不愧是996的程序员真有想法，给你扣波“666”。

以下为在Java中使用AIO，可以看到使用起来十分复杂，并且性能与IO Multiplesing差不并不是特别大，所以通常开发很少用AIO。

server.java

java复制代码package com.github.jiangxch.jdk.demo.netio.aio;


import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;

public class AioServer {

    public AsynchronousServerSocketChannel serverSocketChannel;

    public void listen() throws Exception {
        //打开一个服务端通道
        serverSocketChannel = AsynchronousServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(9988));//监听9988端口
        //监听
        serverSocketChannel.accept(this, new CompletionHandler<AsynchronousSocketChannel, AioServer>() {
            @Override
            public void completed(AsynchronousSocketChannel client, AioServer attachment) {
                try {
                    if (client.isOpen()) {
                        System.out.println("接收到新的客户端的连接，地址：" + client.getRemoteAddress());
                        final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        //读取客户端发送的数据
                        client.read(byteBuffer, client, new CompletionHandler<Integer, AsynchronousSocketChannel>() {
                            @Override
                            public void completed(Integer result, AsynchronousSocketChannel attachment) {
                                try {
                                    //读取请求，处理客户端发送的数据
                                    byteBuffer.flip();
                                    String content = Charset.defaultCharset().newDecoder().decode(byteBuffer).toString();
                                    System.out.println("服务端接受到客户端发来的数据：" + content);
                                    //向客户端发送数据
                                    ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
                                    writeBuffer.put("Server send".getBytes());
                                    writeBuffer.flip();
                                    attachment.write(writeBuffer).get();
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }

                            @Override
                            public void failed(Throwable exc, AsynchronousSocketChannel attachment) {
                                try {
                                    exc.printStackTrace();
                                    attachment.close();
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            }
                        });
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    //当有新的客户端接入的时候，直接调用accept的方法，递归执行下去，这样可以保证多个客户端都可以阻塞
                    attachment.serverSocketChannel.accept(attachment, this);
                }
            }

            @Override
            public void failed(Throwable exc, AioServer attachment) {
                exc.printStackTrace();
            }
        });
    }


    public static void main(String[] args) throws Exception {
        new AioServer().listen();
        Thread.sleep(Integer.MAX_VALUE);
    }

}

client.java

java复制代码package com.github.jiangxch.jdk.demo.netio.aio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutionException;


public class AioClient {
    public static void main(String[] args) throws IOException, InterruptedException {
        //打开一个客户端通道
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
        //与服务端建立连接
        channel.connect(new InetSocketAddress("127.0.0.1", 9988));
        //睡眠一秒，等待与服务端的连接
        Thread.sleep(1000);

        try {
            //向服务端发送数据
            channel.write(ByteBuffer.wrap("Hello,我是客户端".getBytes())).get();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        try {
            //从服务端读取返回的数据
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            channel.read(byteBuffer).get();//将通道中的数据写入缓冲Buffer
            byteBuffer.flip();
            String result = Charset.defaultCharset().newDecoder().decode(byteBuffer).toString();
            System.out.println("客户端收到服务端返回的内容：" + result);//服务端返回的数据
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

}

AIO与NIO相比，除了上面使用不同之外，异步回调的方式也不会因为系统调用而产生更多的上下文切换和数据拷贝。

操作系统网络 IO

如果仅仅只是开发功能，上面的APi基本足够了。但是对于面试、优化来说，还需要更深入一层理解操作系统如何实现IO多路复用。Linux中提供到了以下API接口。

select

Linux中select函数定义：

1 2	c复制代码#include <sys/select.h> int select(int maxfdpl,fd_set readset,fd_set writeset,fd_set excepset,const struct timeval timeout);

timeout参数

timeout是一个timeval的结构体，其数据结构定义：

c复制代码struct {
    long tv_sec;// 等待秒
    long tv_usec; // microseconds
}

fd_set数据结构定义：

c复制代码typedef struct fd_set {

__int32_t fds_bits[__DARWIN_howmany(__DARWIN_FD_SETSIZE, __DARWIN_NFDBITS)];

} fd_set;

fdset是一个数组，类型是32位的int类型。一个int类型可以表示32个文件描述符。可以通过判断int对应的位是0？or1？来判断是否可读可写。

该参数表示等待多久如果还无事件就绪，则直接返回。
传null，表示无超时设置，会永远等待直到事件就绪。
传0表示不等待，会立即返回。

excepset、writeset、readset
这三个参数代表要监听的条件，分别是异常事件就绪，写事件就绪，读事件就绪。
maxfdpl
该参数表示最大文件描述符的数量。

那么seelct是怎么工作的呢？

比如我们编写了一个NIO通信的java进程，就会在操作系统的内存地址中创建一个Process的数据结构，该数据结构会存储进程的pid，以及进程打开的文件对应的文件描述符，进程运行后，当进程接收到连接建立会创建一个套接字，Linux中一切皆文件，套接字也会被存放到Process，当调用select函数对套接字进行监听时，内核就会遍历这些套接字文件描述符对应的文件，在使用文件相关的api判断socket对应的文件是否可读可写，如果对应事件就绪，就会修改传入的event参数的对应bit位最后返回就绪的套接字给调用方。

那么对于一个socket套接字来说，什么情况下称作可读，什么情况称为可写呢？我们知道对于一个套接字，有读写两个缓冲区。在select函数中，定义了读的低水位，当读缓冲区的数据大于低水位，才会出发套接字的读就绪事件。写就绪也相同，当写缓冲区空闲大小大于写缓冲区的低水位才会触发。

所以并不是网卡家接收1byte就触发，接收1byte就触发一次。

poll

poll的功能与select类似。他俩的区别在于select出现的早，文件描述的最大限制是在代码中定义了一个常量表示=1024，poll则没有这个限制，poll能使用的最大文件描述仅收操作系统的限制。

epoll

epoll是对select的一个优化，我们来看下seelct的操作步骤：

c复制代码while(1) {
    select(fdlist);// 监听文件描述符
    for(fd : fdlist) {
        if(fd.can_read()) {
               
        }
        if(fd.can_write()) {
        
        }
    }

}

根据以上为代码，我们捋一下整个操作的性能开销：

select是系统调用，系统调用一定伴随着上下文切换和内存拷贝，频繁select会使得CPU过高。
select函数是通过遍历传入的所有描述符对应的文件来判断是否有事件就绪，因此如果传入的文件描述符过多，会使得扫描耗时更久。
select调用时参数和返回值混在了一起，需要我们调用方再次遍历进行判断。

epoll主要是对上面三个情况进行了优化。epoll的API定义如下：

1	c复制代码int epoll_create(int size);

1	csharp复制代码int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

1	arduino复制代码int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

epoll使用分为了三个函数，epoll_create只需调用一次，用于创建epoll的文件描述符，size参数只需大于0即可，最开始size表示初始化文件描述符的数量，若不够就会进行扩容。e poll_ctl函数则是向内核中的epoll实例添加、修改、删除对某个文件描述符上的事件监听。

op表示对于的操作，可以传入EPOLL_CTM_ADD,EPOLL_CTL_DEL,EPOLL_CTL_MOD。fd参数表示要监听的文件描述符。event参数结构体定义：

c复制代码struct epoll_event {
    __unit32_t events;// 表示监听什么操作，读？写？
    epoll_data_t data;// 用户数据变量

}

结构体中 events可以是以下几个宏的集合：

EPOLLIN: 监听读
EPOLLOUT：监听写
EPOLLERR：监听操作
EPOLLPRI：监听外来数据
EPOLLHUP：描述符被挂断
EPOLLET：设置触发方法，边缘触发，水平触发
EPOLLONESHOT：仅监听一次，设置该参数如果需要继续监听需要重新加入到epoll的监听事件队列中

epoll_wait方法的events参数用于接收返回值，会返回就绪的事件。

那么epoll相对selec，它快在哪里呢？

仅返回就绪的文件描述符

select在返回返回值时，会将所有描述符（不论就绪还是未就绪的）进行返回，而epoll只会返回就绪的文件描述符，以及拷贝就绪socket的到用户内存。

更高效的查找就绪事件的方式

select需要遍历传入的所有的文件描述符，找到描述符在操作系统中对应的文件，逐个判断文件是否满足设置的条件，进而触发对应的事件。所以select会随着文件描述符数量的增长性能降低，而epoll则不会。

操作系统会注册一个中断函数，当网卡接收到数据会出发这个中断函数，如果接收到的数据是需要监听的文件描述符，就会直接将该描述符放入就绪列表。当调用epoll_wait时，直接返回该列表。

除此之外，epoll还有比较高级特性就是工作方式。

LT：水平触发，当如果设置事件为水平触发，则调用epoll_wait时会判断事件是否是水平触发，如果是水平触发并且该事件未经过用户程序处理，则会重新放入就绪列表中。
ET：边缘触发，当epoll_wait返回后，必须立即处理就绪事件。

那epoll为什么要设计LT与ET呢？为了面试，此处说一下自己的理解吧：

最开始之初，epoll只有LT的触发模式，水平触发对应到代码中，如果你监听某个socket读数据，当读队列一直有数据时都会触发事件，这种方式会导致一直有事件触发，从而到导致频繁的系统调用。而如果设置了ET，只有当读缓冲区接收到数据时才会触发（从空-》有数据），会降低系统调用。

参考文章

cloud.tencent.com/developer/a…

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有