程序员最近都爱上了这个网站  程序员们快来瞅瞅吧!  it98k网:it98k.com

本站消息

站长简介/公众号

  出租广告位,需要合作请联系站长


+关注
已关注

分类  

暂无分类

标签  

暂无标签

日期归档  

Java - I/O 多路复用(NIO)

发布于2021-06-12 14:18     阅读(713)     评论(0)     点赞(2)     收藏(3)


NIO

NIO,有人解释为new I/O,有人解释为Non-block I/O(我更倾向后者)。

正是由于Java传统BIO的拙劣表现,才使得Java支持非阻塞I/O的呼声日渐高涨,最终,JDK1.4版本提供了新的NIO类库,Java终于也可以支持非阻塞I/O 了。NIO主要的类和接口如下:

  • 进行异步I/O操作的缓冲区ByteBuffer等;
  • 进行异步I/O操作的管道Pipe;
  • 进行各种I/O操作(异步或者同步)的Channel,包括ServerSocketChannel和SocketChannel。
  • 多种字符集的编码能力和解码能力;
  • 实现非阻塞I/O操作的多路复用器selector:
  • 基千流行的Perl实现的正则表达式类库;
  • 文件通道FileChannelo。

新的NIO类库的提供,极大地促进了基于Java的异步非阻塞编程的发展和应用,但是,它依然有不完善的地方,特别是对文件系统的处理能力仍显不足,主要问题如下。

  • 没有统一的文件属性(例如读写权限);
  • API能力比较弱,例如目录的级联创建和递归遍历,往往需要自己实现:
  • 底层存储系统的一些高级API无法使用:
  • 所有的文件操作都是同步阻塞调用,不支持异步文件读写操作。

2011年7月28日,JDKI.7正式发布。她将原来的NIO类库进行了升级,被称为NIO2.0。它主要提供了如下三个方 面的改进。

  • 能够提供能够批量获取文件属性的API,这些API具有平台无关性,不与特定的文件系统相耦合,另外它还提供了标准文件系统的SPI,供各个服务提供商扩展实现;
  • 提供AIO功能,支持基于文件的异步I/O操作和针对网络套接字的异步操作;
  • 完成JSR-5I定义的通道功能,包括对配置和多播数据报的支持等;

NIO类库概念和功能介绍

缓冲区Buffer

Buffer是一个对象,它包含一些要写入或者要读出的数据。在NIO库中,所有数据都是用缓冲区处理的。在读取数据时,它是直接读到缓冲区中的;在写入数据时,写入到缓冲区中。任何吋候访问NIO中的数据,都是通过缓冲区进行操作。

缓冲区实质上是一个数组。通常它是一个字节数组(ByteBuffer),也可以使用其他种类的数组。但是一个缓冲区不仅仅是一个数组,缓冲区提供了对数据的结构化访问以及维护读写位置(limit)等信息。

最常用的缓冲区是ByteBuffer, 一个ByteBuffer提供了一组功能用于操作byte数组。除了 ByteBuffer,还有其他的一些缓冲区,事实上,每一种Java基本类型(除了 Boolean类型)都对应有一种缓冲区,具体如下。

  • ByteBuffer:字节缓冲区
  • CharBuffer:字符缓冲区
  • ShortBuffer:短整型缓冲K
  • IntBuffer:整形缓冲区
  • LongBuffer:长整形缓冲区
  • FloatBuffer:浮点型缓冲区
  • DoubleBuffer: 双精度浮点型缓冲区

每—个Buffer类都是Buffer接口的一个T实例。除了 ByteBuffer,每一个Buffe类都有完全一样的操作,只是它们所处理的数据类型不一样。因为大多数标准I/O操作都使用ByteBuffer,所以它除了具有一般缓冲区的操作之外还提供一些特有的操作,方便网络读写。

管道channel

Channel是一个通道,可以通过它读取和写入数据,它就像自来水管一样,网络数据通过Channel读取和写入。通道与流的不同之处在于通道是双向的,流只是在一个方向上移动(一个流必须是InputStream或者OutputStream的子类),而且通道可以用于读、写或者同时用于读写。

Channel的类继承图如下。

因为Channel是全双工的,所以它可以比流更好地映射底层操作系统的API。特别是在UNIX网络编程模型中,底层操作系统的通道都是全双工的,同时支持读写操作。

多路复用器Selector

多路复用器提供选择已经就绪的任务的能力。 简单来讲, Selector会不断地轮询注册在其上的Channel,如果某个Channel上面有新的TCP 连接接入、 读和写事件, 这个Channel就处于就绪状态, 会被Selector轮询出来, 然后通过SelectionKey可以获取就绪Channel的集合, 进行后续的I/O操作。

使用NIO实现服务端

  1. public class TimeServerNio {
  2. public static void main(String[] args) {
  3. //设置监听端口
  4. int port = 8080;
  5. if (args != null && args.length > 0) {
  6. try {
  7. port = Integer.valueOf(args[0]);
  8. } catch (NumberFormatException e) {
  9. //采用默认值
  10. }
  11. }
  12. //创建了一个被称为MultiplexerTimeServer的多路复用类, 它是个一个独立
  13. //的线程, 负责轮洵多路复用器Selctor,可以处理多个客户端的并发接入
  14. MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
  15. new Thread(timeServer, "NIO-MultiplexerTirneServer-001H").start();
  16. }
  17. }
  18. public class MultiplexerTimeServer implements Runnable {
  19. private Selector selector;
  20. private ServerSocketChannel servChannel;
  21. private volatile boolean stop;
  22. /**
  23. * 在构造方法中进行资源初始化, 创建多路复用器Selector、
  24. * ServerSocketChannel,对 Channel 和 TCP 参数进行配置
  25. *
  26. * @param port
  27. */
  28. public MultiplexerTimeServer(int port) {
  29. try {
  30. selector = Selector.open();
  31. servChannel = ServerSocketChannel.open();
  32. //设置为异步非阻塞模式
  33. servChannel.configureBlocking(false);
  34. servChannel.socket().bind(new InetSocketAddress(port), 1024);
  35. //注册selector
  36. servChannel.register(selector, SelectionKey.OP_ACCEPT);
  37. System.out.println("The time server is start in port : " + port);
  38. } catch (IOException e) {
  39. //资源初始化失败,退出系统
  40. e.printStackTrace();
  41. System.exit(1);
  42. }
  43. }
  44. public void stop() {
  45. this.stop = true;
  46. }
  47. @Override
  48. public void run() {
  49. while (!stop) {
  50. try {
  51. //循环遍历selector,它的休眠时间为1s
  52. selector.select(1000);
  53. //返回就绪状态的Channel的SelectionKey集合
  54. Set<SelectionKey> selectedKeys = selector.selectedKeys();
  55. Iterator<SelectionKey> it = selectedKeys.iterator();
  56. SelectionKey key = null;
  57. //通过对就绪状态的Channel集合进行迭代, 可以进行网络的异步读写操作
  58. while (it.hasNext()) {
  59. key = it.next();
  60. it.remove();
  61. try {
  62. handleInput(key);
  63. } catch (Exception e) {
  64. if (key != null) {
  65. key.cancel();
  66. if (key.channel() != null) {
  67. key.channel().close();
  68. }
  69. }
  70. }
  71. }
  72. } catch (Throwable t) {
  73. t.printStackTrace();
  74. }
  75. }
  76. // 多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源
  77. if (selector != null) {
  78. try {
  79. selector.close();
  80. } catch (IOException e) {
  81. e.printStackTrace();
  82. }
  83. }
  84. }
  85. private void handleInput(SelectionKey key) throws IOException {
  86. if (key.isValid()) {
  87. // 处理新接入的请求消息
  88. if (key.isAcceptable()) {
  89. // Accept the new connection
  90. ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
  91. SocketChannel sc = ssc.accept();
  92. sc.configureBlocking(false);
  93. // Add the new connection to the selector
  94. sc.register(selector, SelectionKey.OP_READ);
  95. }
  96. if (key.isReadable()) {
  97. // Read the data
  98. SocketChannel sc = (SocketChannel) key.channel();
  99. ByteBuffer readBuffer = ByteBuffer.allocate(1024);
  100. int readBytes = sc.read(readBuffer);
  101. if (readBytes > 0) {
  102. readBuffer.flip();
  103. byte[] bytes = new byte[readBuffer.remaining()];
  104. readBuffer.get(bytes);
  105. String body = new String(bytes, "UTF-8");
  106. System.out.println("The time server receive order : "
  107. + body);
  108. String currentTime = "QUERY TIME ORDER"
  109. .equalsIgnoreCase(body) ? new java.util.Date(
  110. System.currentTimeMillis()).toString()
  111. : "BAD ORDER";
  112. doWrite(sc, currentTime);
  113. } else if (readBytes < 0) {
  114. // 对端链路关闭
  115. key.cancel();
  116. sc.close();
  117. } else {
  118. ; // 读到0字节,忽略
  119. }
  120. }
  121. }
  122. }
  123. /**
  124. * 将应答消息异步发送给客户端
  125. *
  126. * @param channel channel
  127. * @param response response
  128. * @throws IOException exception
  129. */
  130. private void doWrite(SocketChannel channel, String response)
  131. throws IOException {
  132. if (response != null && response.trim().length() > 0) {
  133. //将字符串编码成字节数组, 根据字节数组的容量创建ByteBuff
  134. byte[] bytes = response.getBytes();
  135. ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
  136. //将字节数据复制到缓冲区
  137. writeBuffer.put(bytes);
  138. writeBuffer.flip();
  139. //将缓冲区中的字节数组发送出去
  140. channel.write(writeBuffer);
  141. }
  142. }
  143. }

使用NIO实现客户端

  1. public class TimeClientNio {
  2. public static void main(String[] args) {
  3. int port = 8080;
  4. if (args != null && args.length > 0) {
  5. try {
  6. port = Integer.valueOf(args[0]);
  7. } catch (NumberFormatException e) {
  8. // 采用默认值
  9. }
  10. }
  11. new Thread(new TimeClientHandle("127.0.0.1", port), "TimeClient-001")
  12. .start();
  13. }
  14. }
  15. public class TimeClientHandle implements Runnable{
  16. private String host;
  17. private int port;
  18. private Selector selector;
  19. private SocketChannel socketChannel;
  20. private volatile boolean stop;
  21. /**
  22. * implements Runnable
  23. *
  24. * @param host host
  25. * @param port port
  26. */
  27. public TimeClientHandle(String host, int port) {
  28. this.host = host == null ? "127.0.0.1" : host;
  29. this.port = port;
  30. try {
  31. selector = Selector.open();
  32. socketChannel = SocketChannel.open();
  33. //设置为异步非阻塞模式
  34. socketChannel.configureBlocking(false);
  35. } catch (IOException e) {
  36. e.printStackTrace();
  37. System.exit(1);
  38. }
  39. }
  40. @Override
  41. public void run() {
  42. try {
  43. //发送连接请求
  44. doConnect();
  45. } catch (IOException e) {
  46. e.printStackTrace();
  47. System.exit(1);
  48. }
  49. while (!stop) {
  50. try {
  51. selector.select(1000);
  52. Set<SelectionKey> selectedKeys = selector.selectedKeys();
  53. Iterator<SelectionKey> it = selectedKeys.iterator();
  54. SelectionKey key = null;
  55. while (it.hasNext()) {
  56. key = it.next();
  57. it.remove();
  58. try {
  59. handleInput(key);
  60. } catch (Exception e) {
  61. if (key != null) {
  62. key.cancel();
  63. if (key.channel() != null)
  64. key.channel().close();
  65. }
  66. }
  67. }
  68. } catch (Exception e) {
  69. e.printStackTrace();
  70. System.exit(1);
  71. }
  72. }
  73. // 多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源
  74. if (selector != null){
  75. try {
  76. selector.close();
  77. } catch (IOException e) {
  78. e.printStackTrace();
  79. }
  80. }
  81. }
  82. private void handleInput(SelectionKey key) throws IOException {
  83. if (key.isValid()) {
  84. // 判断是否连接成功
  85. SocketChannel sc = (SocketChannel) key.channel();
  86. //如果返回值为true,说明客户端连接成功
  87. if (key.isConnectable()) {
  88. if (sc.finishConnect()) {
  89. sc.register(selector, SelectionKey.OP_READ);
  90. doWrite(sc);
  91. } else {
  92. System.exit(1);// 连接失败,进程退出
  93. }
  94. }
  95. if (key.isReadable()) {
  96. //预分配1M的接收缓冲K用于读取应答消息, 调川Socketchannel的read()方法进行异步读取操作
  97. ByteBuffer readBuffer = ByteBuffer.allocate(1024);
  98. int readBytes = sc.read(readBuffer);
  99. if (readBytes > 0) {
  100. readBuffer.flip();
  101. byte[] bytes = new byte[readBuffer.remaining()];
  102. readBuffer.get(bytes);
  103. String body = new String(bytes, "UTF-8");
  104. System.out.println("Now is : " + body);
  105. this.stop = true;
  106. } else if (readBytes < 0) {
  107. // 对端链路关闭
  108. key.cancel();
  109. sc.close();
  110. } else
  111. ; // 读到0字节,忽略
  112. }
  113. }
  114. }
  115. private void doConnect() throws IOException {
  116. // 如果直接连接成功,则注册到多路复用器上,发送请求消息,读应答
  117. if (socketChannel.connect(new InetSocketAddress(host, port))) {
  118. socketChannel.register(selector, SelectionKey.OP_READ);
  119. doWrite(socketChannel);
  120. } else {
  121. //如果没有直接连接成功, 则说明服务端没有返回TCP握手应答消息, 但这并不代表连接失败, 我们需要将SocketChannel
  122. //注册到多路复用器Selector上, 注册SelectionKey.OP CONNECT»当服务端返回TCP
  123. //syn-ack消息后, Selector就能够轮询到这个SocketChannel处于连接就绪状态
  124. socketChannel.register(selector, SelectionKey.OP_CONNECT);
  125. }
  126. }
  127. private void doWrite(SocketChannel sc) throws IOException {
  128. byte[] req = "QUERY TIME ORDER".getBytes();
  129. ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
  130. writeBuffer.put(req);
  131. writeBuffer.flip();
  132. sc.write(writeBuffer);
  133. //hasRemaining()方法对发送结果进行判断, 是否全部发送完成
  134. if (!writeBuffer.hasRemaining()) {
  135. System.out.println("Send order 2 server succeed.");
  136. }
  137. }
  138. }

NIO编程难度确实比同步阻塞BIO大很多, 上面的NIO 例子并没有考虑“ 半包读” 和“ 半包写” , 如果加上这些, 代码将会更加复杂。

NIO优点总结

  1. 客户端发起的连接操作是异步的, 可以通过在多路复用器注册OP_CONNECT等待后续结果, 不需要像之前的客户端那样被同步阻塞。
  2. Socketchannel的读写操作都是异步的, 如果没有可读写的数据它不会同步等待,直接返回, 这样I/O通信线程就可以处理其他的链路, 不需要同步等待这个链路可用。
  3. 线程模型的优化: 由于JDK的Selector在Linux等主流操作系统上通过epoll实现, 它没有连接句柄数的限制(只受限于操作系统的最大句柄数或者对单个进程的句柄限制), 这意味着一个Selector线程可以同时处理成千上万个客户端连接, 而且性能不会随着客户端的增加而线性下降, 因此, 它非常适合做高性能、 高负载的网络服务器。

 



所属网站分类: 技术文章 > 博客

作者:狗蛋来了

链接:http://www.javaheidong.com/blog/article/222114/be934dda2668d82095d8/

来源:java黑洞网

任何形式的转载都请注明出处,如有侵权 一经发现 必将追究其法律责任

2 0
收藏该文
已收藏

评论内容:(最多支持255个字符)