Java 网络 IO（BIO、NIO、AIO）

简介

Java 支持三种网络 IO 模子：BIO、NIO、AIO。

• Java BIO 是同步阻塞模子，一个连接对应一个线程，客户端有连接请求时服务端就启动一个线程，纵然这个连接不做任何事变也会占用线程资源。
  
• Java NIO 是同步非阻塞模子，一个线程可以处置惩罚多个连接，客户端连接请求会注册到多路复用器(Selector)，多路复用器检测到连接有 IO 时间就会处置惩罚。
  
• Java AIO 是异步非阻塞模子，AIO 引入了异步通道的概念，读写异步通道会立刻返回，读写的数据由 Future 或 CompletionHandler 进一步处置惩罚。
  

BIO 实用于连接数少的场景，程序编写比较简单，对服务器的资源要求比较高，JDK1.4之前的唯一选择。NIO 实用于连接数多的场景，例如谈天服务器、服务器间通讯等，程序编写比较复杂，JDK1.4开始支持。AIO 也实用于连接数多的场景，但更加偏向于异步操作多的场景。
Java BIO

模子示例

客户端代码示例

1. import java.io.*;
2. import java.net.InetSocketAddress;
3. import java.net.Socket;
4. import java.net.SocketAddress;
6. public class BIOClient {
7.     public static void main(String[] args) {
8.         new BIOClient().start("localhost", 6666);
9.     }
11.     public void start(String host, int port) {
12.         // 初始化 socket
13.         Socket socket = new Socket();
15.         try {
16.             // 设置 socket 连接
17.             SocketAddress remote = new InetSocketAddress(host, port);
18.             socket.setSoTimeout(5000);
19.             socket.connect(remote);
21.             // 发送数据
22.             PrintWriter writer = getWriter(socket);
23.             writer.write("hello server");
24.             writer.flush();
26. //            // 发起请求
27. //            PrintWriter writer = getWriter(socket);
28. //            writer.write(compositeRequest(host));
29. //            writer.flush();
30. //
31. //            // 读取响应
32. //            String msg;
33. //            BufferedReader reader = getReader(socket);
34. //            while ((msg = reader.readLine()) != null) {
35. //                System.out.println(msg);
36. //            }
37.         } catch (IOException e) {
38.             e.printStackTrace();
39.         } finally {
40.             try {
41.                 socket.close();
42.             } catch (IOException e) {
43.                 e.printStackTrace();
44.             }
45.         }
46.     }
48.     private BufferedReader getReader(Socket socket) throws IOException {
49.         InputStream in = socket.getInputStream();
50.         return new BufferedReader(new InputStreamReader(in));
51.     }
53.     private PrintWriter getWriter(Socket socket) throws IOException {
54.         OutputStream out = socket.getOutputStream();
55.         return new PrintWriter(new OutputStreamWriter(out));
56.     }
58.     private String compositeRequest(String host) {
59.         return "GET / HTTP/1.1\r\n" +
60.                 "Host: " + host + "\r\n" +
61.                 "User-Agent: curl/7.43.0\r\n" +
62.                 "Accept: */*\r\n\r\n";
63.     }
64. }

复制代码

服务端代码示例

1. import java.io.InputStream;
2. import java.net.ServerSocket;
3. import java.net.Socket;
4. import java.util.concurrent.ExecutorService;
5. import java.util.concurrent.Executors;
7. public class BIOServer {
8.     public static void main(String[] args) throws Exception {
9.         // 创建一个线程池
10.         ExecutorService pool = Executors.newCachedThreadPool();
11.         // 创建 ServerSocket
12.         ServerSocket serverSocket = new ServerSocket(6666);
13.         while (true) {
14.             // 等待客户端连接
15.             final Socket socket = serverSocket.accept();
16.             // 接收到一个客户端连接 放入线程池进行处理
17.             pool.execute(() -> process(socket));
18.         }
19.     }
21.     static void process(Socket socket) {
22.         try {
23.             byte[] bytes = new byte[1024];
24.             // 通过 socket 获取输入流
25.             InputStream inputStream = socket.getInputStream();
26.             // 循环读取客户端发送的数据
27.             while (true) {
28.                 // 没有数据的时候这里会阻塞等待
29.                 int read = inputStream.read(bytes);
30.                 if (read == -1) break;
31.                 // 输出客户端发送的数据
32.                 System.out.println(new String(bytes, 0, read));
33.             }
34.         } catch (Exception e) {
35.             e.printStackTrace();
36.         } finally {
37.             try {
38.                 socket.close();
39.             } catch (Exception e) {
40.                 e.printStackTrace();
41.             }
42.         }
43.     }
44. }

复制代码

Java NIO

NIO 接纳 Reactor 模式，属于 IO 多路复用模子，可以用一个线程处置惩罚多个请求。NIO 有三大核心模块，通道（Channel）、缓冲区（Buffer）、选择器（Selector）。NIO 的非阻塞模式，使主线程在未发生数据读写事件时无需阻塞，可以继承做其他事变，这就大大增强了服务器的并发处置惩罚能力。
模子示例

Selector 对应一个线程，一个 Selector 可以对应多个 Channel，一个 Channel 对应一个 Buffer。程序切换到哪个 Channel 是由事件决定的，Selector 会根据不同的事件切换不同的 Channel。下图描述了 Channel、Buffer 和 Selector 的关系。

MappedByteBuffer 简介

NIO 提供的 MappedByteBuffer 支持支持在内存（堆外内存）中修改文件，可以淘汰一次数据拷贝。文件同步的部分，由 NIO 本身完成。
代码示例

1. import java.io.RandomAccessFile;
2. import java.nio.MappedByteBuffer;
3. import java.nio.channels.FileChannel;
5. /**
6. * 说明 1.MappedByteBuffer 可让文件直接在内存（堆外内存）修改,操作系统不需要拷贝一次
7. */
8. public class MappedByteBufferTest {
10.     public static void main(String[] args) throws Exception {
12.         RandomAccessFile randomAccessFile = new RandomAccessFile("1.txt", "rw");
13.         //获取对应的通道
14.         FileChannel channel = randomAccessFile.getChannel();
16.         /**
17.          * 参数 1:FileChannel.MapMode.READ_WRITE 使用的读写模式
18.          * 参数 2：0：可以直接修改的起始位置
19.          * 参数 3:5: 是映射到内存的大小（不是索引位置），即将 1.txt 的多少个字节映射到内存
20.          * 可以直接修改的范围就是 0-5
21.          * 实际类型 DirectByteBuffer
22.          */
23.         MappedByteBuffer mappedByteBuffer = channel.map(FileChannel.MapMode.READ_WRITE, 0, 5);
25.         mappedByteBuffer.put(0, (byte) 'H');
26.         mappedByteBuffer.put(3, (byte) '9');
27.         mappedByteBuffer.put(5, (byte) 'Y');//IndexOutOfBoundsException
29.         randomAccessFile.close();
30.         System.out.println("修改成功~~");
31.     }
32. }

复制代码

 NIO 编程代码原理分析图

关于 NIO 非阻塞网络编程相关的（Selector、SelectionKey、ServerScoketChannel 和 SocketChannel）关系梳理图

服务端代码示例

可以结合上面的原理图观察代码实现细节

1. import java.io.IOException;
2. import java.net.InetSocketAddress;
3. import java.nio.ByteBuffer;
4. import java.nio.channels.*;
5. import java.util.Iterator;
7. public class GroupChatServer {
9.     // 定义属性
10.     private Selector selector;
11.     private ServerSocketChannel listenChannel;
13.     private static final int PORT = 6667;
15.     // 构造器执行初始化工作
16.     public GroupChatServer() {
17.         try {
18.             // 得到选择器
19.             selector = Selector.open();
20.             // 监听端口的主线程
21.             listenChannel = ServerSocketChannel.open();
22.             // 绑定端口
23.             listenChannel.socket().bind(new InetSocketAddress(PORT));
24.             // 设置非阻塞模式
25.             listenChannel.configureBlocking(false);
26.             // 将该 listenChannel 注册到 selector
27.             listenChannel.register(selector, SelectionKey.OP_ACCEPT);
28.         } catch (IOException e) {
29.             e.printStackTrace();
30.         }
31.     }
33.     public void listen() {
34.         try {
35.             // 循环处理
36.             while (true) {
37.                 int count = selector.select();
38.                 // 有事件处理
39.                 if (count > 0) {
40.                     // 遍历得到 selectionKey 集合
41.                     Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
42.                     while (iterator.hasNext()) {
43.                         // 取出 selectionKey
44.                         SelectionKey key = iterator.next();
45.                         // 监听到 accept
46.                         if (key.isAcceptable()) {
47.                             SocketChannel sc = listenChannel.accept();
48.                             sc.configureBlocking(false);
49.                             // 将该 sc 注册到 selector
50.                             sc.register(selector, SelectionKey.OP_READ);
51.                             // 提示
52.                             System.out.println(sc.getRemoteAddress() + " 上线 ");
53.                         }
54.                         if (key.isReadable()) {// 通道发送read事件，即通道是可读的状态
55.                             // 处理读(专门写方法..)
56.                             readData(key);
57.                         }
58.                         // 当前的 key 删除，防止重复处理
59.                         iterator.remove();
60.                     }
61.                 } else {
62.                     System.out.println("等待....");
63.                 }
64.             }
65.         } catch (Exception e) {
66.             e.printStackTrace();
67.         } finally {
68.             // 发生异常处理....
69.         }
70.     }
72.     // 读取客户端消息
73.     public void readData(SelectionKey key) {
74.         SocketChannel channel = null;
75.         try {
76.             // 得到 channel
77.             channel = (SocketChannel) key.channel();
78.             // 创建 buffer
79.             ByteBuffer buffer = ByteBuffer.allocate(1024);
80.             int count = channel.read(buffer);// NIO这里不会阻塞 因为事件触发时必然已经有数据了 所以叫非阻塞IO
81.             // 根据 count 的值做处理
82.             if (count > 0) {
83.                 // 把缓存区的数据转成字符串
84.                 String msg = new String(buffer.array());
85.                 // 输出该消息
86.                 System.out.println("form客户端:" + msg);
87.                 // 向其它的客户端转发消息(去掉自己),专门写一个方法来处理
88.                 sendInfoToOtherClients(msg, channel);
89.             }
90.         } catch (IOException e) {
91.             try {
92.                 System.out.println(channel.getRemoteAddress() + "离线了..");
93.                 // 取消注册
94.                 key.cancel();
95.                 // 关闭通道
96.                 channel.close();
97.             } catch (IOException e2) {
98.                 e2.printStackTrace();
99.             }
100.         }
101.     }
103.     // 转发消息给其它客户(通道)
104.     private void sendInfoToOtherClients(String msg, SocketChannel self) throws IOException {
105.         System.out.println("服务器转发消息中...");
106.         // 遍历所有注册到 selector 上的 SocketChannel,并排除 self
107.         for (SelectionKey key : selector.keys()) {
108.             // 通过 key 取出对应的 SocketChannel
109.             Channel targetChannel = key.channel();
110.             // 排除自己
111.             if (targetChannel instanceof SocketChannel && targetChannel != self) {
112.                 // 转型
113.                 SocketChannel dest = (SocketChannel) targetChannel;
114.                 // 将 msg 存储到 buffer
115.                 ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
116.                 // 将 buffer 的数据写入通道
117.                 dest.write(buffer);
118.             }
119.         }
120.     }
122.     public static void main(String[] args) {
123.         // 创建服务器对象
124.         GroupChatServer groupChatServer = new GroupChatServer();
125.         groupChatServer.listen();
126.     }
127. }

复制代码

Java AIO

AIO 是异步非阻塞的，引入了异步通道的概念，接纳 Proactor 模式，操作系统完成数据拷贝操作后才会关照服务端线程。AIO 本质上还是 IO 多路复用模子，与 NIO 比起来，AIO 只是在非阻塞的前提下增加了异步功能，详细则表现在代码编写以及数据传输两个层面。

• 从代码编写角度来说，原来的同步方法会阻塞等候接口返回，而现在可以异步等候返回结果。
  
• 从数据传输角度来说，每个请求都须要传输数据，NIO 固然好坏阻塞的，但是事件到达后，NIO 须要本身把数据从内核空间复制到用户空间。AIO 引入异步逻辑后，事件到达后系统不会立刻关照服务端线程，而是会本身把数据从内核空间复制到用户空间，完成这个操作后，才会关照服务端线程行止理。
  

AIO 的使用场景还是比较少，现在大部分开源框架中应该还是以使用 NIO 为主，AIO 在性能方面的提升还是比较有限，紧张的变化还是增加了异步功能。
如何明白 Reactor 和 Proactor 的区别？

Reactor 可以明白为「来了事件操作系统关照应用进程，让应用进程来处置惩罚」，而Proactor 可以明白为「来了事件操作系统来处置惩罚，处置惩罚完再关照应用进程」。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。