发布于2021-06-12 14:18 阅读(713) 评论(0) 点赞(2) 收藏(3)
NIO,有人解释为new I/O,有人解释为Non-block I/O(我更倾向后者)。
正是由于Java传统BIO的拙劣表现,才使得Java支持非阻塞I/O的呼声日渐高涨,最终,JDK1.4版本提供了新的NIO类库,Java终于也可以支持非阻塞I/O 了。NIO主要的类和接口如下:
新的NIO类库的提供,极大地促进了基于Java的异步非阻塞编程的发展和应用,但是,它依然有不完善的地方,特别是对文件系统的处理能力仍显不足,主要问题如下。
2011年7月28日,JDKI.7正式发布。她将原来的NIO类库进行了升级,被称为NIO2.0。它主要提供了如下三个方 面的改进。
Buffer是一个对象,它包含一些要写入或者要读出的数据。在NIO库中,所有数据都是用缓冲区处理的。在读取数据时,它是直接读到缓冲区中的;在写入数据时,写入到缓冲区中。任何吋候访问NIO中的数据,都是通过缓冲区进行操作。
缓冲区实质上是一个数组。通常它是一个字节数组(ByteBuffer),也可以使用其他种类的数组。但是一个缓冲区不仅仅是一个数组,缓冲区提供了对数据的结构化访问以及维护读写位置(limit)等信息。
最常用的缓冲区是ByteBuffer, 一个ByteBuffer提供了一组功能用于操作byte数组。除了 ByteBuffer,还有其他的一些缓冲区,事实上,每一种Java基本类型(除了 Boolean类型)都对应有一种缓冲区,具体如下。
每—个Buffer类都是Buffer接口的一个T实例。除了 ByteBuffer,每一个Buffe类都有完全一样的操作,只是它们所处理的数据类型不一样。因为大多数标准I/O操作都使用ByteBuffer,所以它除了具有一般缓冲区的操作之外还提供一些特有的操作,方便网络读写。
Channel是一个通道,可以通过它读取和写入数据,它就像自来水管一样,网络数据通过Channel读取和写入。通道与流的不同之处在于通道是双向的,流只是在一个方向上移动(一个流必须是InputStream或者OutputStream的子类),而且通道可以用于读、写或者同时用于读写。
Channel的类继承图如下。
因为Channel是全双工的,所以它可以比流更好地映射底层操作系统的API。特别是在UNIX网络编程模型中,底层操作系统的通道都是全双工的,同时支持读写操作。
多路复用器提供选择已经就绪的任务的能力。 简单来讲, Selector会不断地轮询注册在其上的Channel,如果某个Channel上面有新的TCP 连接接入、 读和写事件, 这个Channel就处于就绪状态, 会被Selector轮询出来, 然后通过SelectionKey可以获取就绪Channel的集合, 进行后续的I/O操作。
- public class TimeServerNio {
-
- public static void main(String[] args) {
- //设置监听端口
- int port = 8080;
- if (args != null && args.length > 0) {
- try {
- port = Integer.valueOf(args[0]);
- } catch (NumberFormatException e) {
- //采用默认值
- }
- }
-
- //创建了一个被称为MultiplexerTimeServer的多路复用类, 它是个一个独立
- //的线程, 负责轮洵多路复用器Selctor,可以处理多个客户端的并发接入
- MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
-
- new Thread(timeServer, "NIO-MultiplexerTirneServer-001H").start();
- }
-
- }
- public class MultiplexerTimeServer implements Runnable {
-
- private Selector selector;
-
- private ServerSocketChannel servChannel;
-
- private volatile boolean stop;
-
- /**
- * 在构造方法中进行资源初始化, 创建多路复用器Selector、
- * ServerSocketChannel,对 Channel 和 TCP 参数进行配置
- *
- * @param port
- */
- public MultiplexerTimeServer(int port) {
- try {
- selector = Selector.open();
- servChannel = ServerSocketChannel.open();
- //设置为异步非阻塞模式
- servChannel.configureBlocking(false);
- servChannel.socket().bind(new InetSocketAddress(port), 1024);
- //注册selector
- servChannel.register(selector, SelectionKey.OP_ACCEPT);
- System.out.println("The time server is start in port : " + port);
- } catch (IOException e) {
- //资源初始化失败,退出系统
- e.printStackTrace();
- System.exit(1);
- }
- }
-
- public void stop() {
- this.stop = true;
- }
-
-
- @Override
- public void run() {
- while (!stop) {
- try {
- //循环遍历selector,它的休眠时间为1s
- selector.select(1000);
- //返回就绪状态的Channel的SelectionKey集合
- Set<SelectionKey> selectedKeys = selector.selectedKeys();
- Iterator<SelectionKey> it = selectedKeys.iterator();
- SelectionKey key = null;
-
- //通过对就绪状态的Channel集合进行迭代, 可以进行网络的异步读写操作
- while (it.hasNext()) {
- key = it.next();
- it.remove();
- try {
- handleInput(key);
- } catch (Exception e) {
- if (key != null) {
- key.cancel();
- if (key.channel() != null) {
- key.channel().close();
- }
- }
- }
- }
- } catch (Throwable t) {
- t.printStackTrace();
- }
- }
-
- // 多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源
- if (selector != null) {
- try {
- selector.close();
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
- }
-
- private void handleInput(SelectionKey key) throws IOException {
-
- if (key.isValid()) {
- // 处理新接入的请求消息
- if (key.isAcceptable()) {
- // Accept the new connection
- ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
- SocketChannel sc = ssc.accept();
- sc.configureBlocking(false);
- // Add the new connection to the selector
- sc.register(selector, SelectionKey.OP_READ);
- }
- if (key.isReadable()) {
- // Read the data
- SocketChannel sc = (SocketChannel) key.channel();
- ByteBuffer readBuffer = ByteBuffer.allocate(1024);
- int readBytes = sc.read(readBuffer);
- if (readBytes > 0) {
- readBuffer.flip();
- byte[] bytes = new byte[readBuffer.remaining()];
- readBuffer.get(bytes);
- String body = new String(bytes, "UTF-8");
- System.out.println("The time server receive order : "
- + body);
- String currentTime = "QUERY TIME ORDER"
- .equalsIgnoreCase(body) ? new java.util.Date(
- System.currentTimeMillis()).toString()
- : "BAD ORDER";
- doWrite(sc, currentTime);
- } else if (readBytes < 0) {
- // 对端链路关闭
- key.cancel();
- sc.close();
- } else {
- ; // 读到0字节,忽略
- }
- }
- }
- }
-
- /**
- * 将应答消息异步发送给客户端
- *
- * @param channel channel
- * @param response response
- * @throws IOException exception
- */
- private void doWrite(SocketChannel channel, String response)
- throws IOException {
- if (response != null && response.trim().length() > 0) {
- //将字符串编码成字节数组, 根据字节数组的容量创建ByteBuff
- byte[] bytes = response.getBytes();
- ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
- //将字节数据复制到缓冲区
- writeBuffer.put(bytes);
- writeBuffer.flip();
- //将缓冲区中的字节数组发送出去
- channel.write(writeBuffer);
- }
- }
- }
- public class TimeClientNio {
-
- public static void main(String[] args) {
-
- int port = 8080;
- if (args != null && args.length > 0) {
- try {
- port = Integer.valueOf(args[0]);
- } catch (NumberFormatException e) {
- // 采用默认值
- }
- }
- new Thread(new TimeClientHandle("127.0.0.1", port), "TimeClient-001")
- .start();
- }
- }
- public class TimeClientHandle implements Runnable{
- private String host;
- private int port;
-
- private Selector selector;
- private SocketChannel socketChannel;
-
- private volatile boolean stop;
-
- /**
- * implements Runnable
- *
- * @param host host
- * @param port port
- */
- public TimeClientHandle(String host, int port) {
- this.host = host == null ? "127.0.0.1" : host;
- this.port = port;
- try {
- selector = Selector.open();
- socketChannel = SocketChannel.open();
- //设置为异步非阻塞模式
- socketChannel.configureBlocking(false);
- } catch (IOException e) {
- e.printStackTrace();
- System.exit(1);
- }
- }
-
- @Override
- public void run() {
- try {
- //发送连接请求
- doConnect();
- } catch (IOException e) {
- e.printStackTrace();
- System.exit(1);
- }
- while (!stop) {
- try {
- selector.select(1000);
- Set<SelectionKey> selectedKeys = selector.selectedKeys();
- Iterator<SelectionKey> it = selectedKeys.iterator();
- SelectionKey key = null;
- while (it.hasNext()) {
- key = it.next();
- it.remove();
- try {
- handleInput(key);
- } catch (Exception e) {
- if (key != null) {
- key.cancel();
- if (key.channel() != null)
- key.channel().close();
- }
- }
- }
- } catch (Exception e) {
- e.printStackTrace();
- System.exit(1);
- }
- }
-
- // 多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源
- if (selector != null){
- try {
- selector.close();
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-
- }
-
- private void handleInput(SelectionKey key) throws IOException {
-
- if (key.isValid()) {
- // 判断是否连接成功
- SocketChannel sc = (SocketChannel) key.channel();
- //如果返回值为true,说明客户端连接成功
- if (key.isConnectable()) {
- if (sc.finishConnect()) {
- sc.register(selector, SelectionKey.OP_READ);
- doWrite(sc);
- } else {
- System.exit(1);// 连接失败,进程退出
- }
- }
- if (key.isReadable()) {
- //预分配1M的接收缓冲K用于读取应答消息, 调川Socketchannel的read()方法进行异步读取操作
- ByteBuffer readBuffer = ByteBuffer.allocate(1024);
- int readBytes = sc.read(readBuffer);
-
- if (readBytes > 0) {
- readBuffer.flip();
- byte[] bytes = new byte[readBuffer.remaining()];
- readBuffer.get(bytes);
- String body = new String(bytes, "UTF-8");
- System.out.println("Now is : " + body);
- this.stop = true;
- } else if (readBytes < 0) {
- // 对端链路关闭
- key.cancel();
- sc.close();
- } else
- ; // 读到0字节,忽略
- }
- }
-
- }
-
- private void doConnect() throws IOException {
- // 如果直接连接成功,则注册到多路复用器上,发送请求消息,读应答
- if (socketChannel.connect(new InetSocketAddress(host, port))) {
- socketChannel.register(selector, SelectionKey.OP_READ);
- doWrite(socketChannel);
- } else {
- //如果没有直接连接成功, 则说明服务端没有返回TCP握手应答消息, 但这并不代表连接失败, 我们需要将SocketChannel
- //注册到多路复用器Selector上, 注册SelectionKey.OP CONNECT»当服务端返回TCP
- //syn-ack消息后, Selector就能够轮询到这个SocketChannel处于连接就绪状态
- socketChannel.register(selector, SelectionKey.OP_CONNECT);
- }
- }
-
- private void doWrite(SocketChannel sc) throws IOException {
- byte[] req = "QUERY TIME ORDER".getBytes();
- ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
- writeBuffer.put(req);
- writeBuffer.flip();
- sc.write(writeBuffer);
- //hasRemaining()方法对发送结果进行判断, 是否全部发送完成
- if (!writeBuffer.hasRemaining()) {
- System.out.println("Send order 2 server succeed.");
- }
- }
- }
NIO编程难度确实比同步阻塞BIO大很多, 上面的NIO 例子并没有考虑“ 半包读” 和“ 半包写” , 如果加上这些, 代码将会更加复杂。
作者:狗蛋来了
链接:http://www.javaheidong.com/blog/article/222114/be934dda2668d82095d8/
来源:java黑洞网
任何形式的转载都请注明出处,如有侵权 一经发现 必将追究其法律责任
昵称:
评论内容:(最多支持255个字符)
---无人问津也好,技不如人也罢,你都要试着安静下来,去做自己该做的事,而不是让内心的烦躁、焦虑,坏掉你本来就不多的热情和定力
Copyright © 2018-2021 java黑洞网 All Rights Reserved 版权所有,并保留所有权利。京ICP备18063182号-2
投诉与举报,广告合作请联系vgs_info@163.com或QQ3083709327
免责声明:网站文章均由用户上传,仅供读者学习交流使用,禁止用做商业用途。若文章涉及色情,反动,侵权等违法信息,请向我们举报,一经核实我们会立即删除!