Java NIO 概述

netty和java nio其实有些千丝万缕的联系,在正式学习netty之前,对java NIO有一定的理解对后续的学习是很有帮助的。故在此花一定篇幅来了解Java NIO。Linux有5中I/O模型,java NIO属于其中的多路复用I/O模型。本文主要介绍下Java NIO的一些基础概念以及其使用的基本流程。主要内容分为三个部分:Channel、Buffer、Selector。

一、Channel

Channel,通道,所有NIO的操作都是从Channel开始,Channel,就类似于传统IO中的stream(流),主要区别在于Channel是双向的,既能读又能写,而stream则是单向的,只能进行读或写;Channel 可以异步地读写, 而 Stream 是阻塞的同步读写;Channel 总是从 Buffer 中读取数据, 或将数据写入到 Buffer 中。传统的IO例子如下

    public static void main(String[] args) throws IOException  {
        File file = new File("data.txt");
        InputStream inputStream = new FileInputStream(file);
        byte[] bytes = new byte[1024];
        inputStream.read(bytes);
        inputStream.close();
    }  

以下是常用的几种通道:

  • FileChannel:不能设置为非阻塞模式.
  • SocketChanel
  • ServerSocketChannel
  • DatagramChannel

通过使用FileChannel可以从文件读或者向文件写入数据;通过SocketChannel,以TCP来向网络连接的两端读写数据;通过ServerSocketChanel能够监听客户端发起的TCP连接,并为每个TCP连接创建一个新的SocketChannel来进行数据读写;通过DatagramChannel,以UDP协议来向网络连接的两端读写数据。

public class Test {
    public static void main(String[] args) throws IOException  {
        File file = new File("data.txt");
        FileOutputStream outputStream = new FileOutputStream(file);
        FileChannel channel = outputStream.getChannel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        String string = "java nio";
        buffer.put(string.getBytes());
        buffer.flip(); //将 NIO Buffer 转换为读模式.
        channel.write(buffer);
        channel.close();
        outputStream.close();
    }  
}

在这里稍微详细说明下SocketChanel和ServerSocketChannel的操作:
SocketChannel 是一个客户端用来进行 TCP 连接的 Channel.

SocketChannel

创建一个 SocketChannel 的方法有两种:

  • 打开一个 SocketChannel, 然后将其连接到某个服务器中
  • 当一个 ServerSocketChannel 接受到连接请求时, 会返回一个 SocketChannel 对象.
1、打开 SocketChannel
SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("http://example.com", 80));
2、关闭
socketChannel.close();
3、读取数据
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = socketChannel.read(buf);

如果 read()返回 -1, 那么表示连接中断了.

4、写入数据
String newData = "New String to write to file..." + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()) {
    channel.write(buf);
}
非阻塞模式

我们可以设置 SocketChannel 为异步模式, 这样我们的 connect, read, write 都是异步的了.

1、连接
socketChannel.configureBlocking(false);
socketChannel.connect(new InetSocketAddress("http://example.com", 80));
while(! socketChannel.finishConnect() ){
    //wait, or do something else...    
}

上述代码中有while循环来判断是否已经连接了主机是因为在异步模式下,connect会立即返回,此时连接或许还没有建立。

2、读写

在异步模式下, 读写的方式是一样的。在读取时, 因为是异步的, 因此我们必须检查 read 的返回值, 来判断当前是否读取到了数据。

ServerSocketChannel

ServerSocketChannel用于服务端监听TCP链接:

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(9999));
while(true){
   SocketChannel socketChannel = serverSocketChannel.accept();
   //...
}
1、打开
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
2、关闭
serverSocketChannel.close();
3、监听连接

我们可以使用ServerSocketChannel.accept()方法来监听客户端的 TCP 连接请求, accept()方法会阻塞, 直到有连接到来, 当有连接时, 这个方法会返回一个 SocketChannel 对象:

while(true){
    SocketChannel socketChannel = serverSocketChannel.accept();
    //...
}
4、非阻塞模式

在非阻塞模式下, accept()是非阻塞的, 因此如果此时没有连接到来, 那么 accept()方法会返回null:

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(9999));
serverSocketChannel.configureBlocking(false);
while(true){
    SocketChannel socketChannel = serverSocketChannel.accept();
    if(socketChannel != null){
        //do something with socketChannel...
        }
}

二、Buffer

一个 Buffer 可以理解成一块内存区域, 我们其实是在这个内存中对数据进行读写。Buffer实质上就是是这样的内存块的一个封装, 并提供了一些操作方法让我们能够方便地进行数据的读写。如果我们需要和Channel交互时就需要用到Buffer,数据从buffer读取到Channel,并从Channel写入Buffer。


客户度向服务器写数据

Channel提供从文件、网络读取数据的渠道,但是读取或写入的数据都必须经由Buffer。客户端发送数据时,必须先将数据存入Buffer中,然后将Buffer中的内容写入通道。服务端这边接收数据必须通过Channel将数据读入到Buffer中,然后再从Buffer中取出数据来处理。

基本使用

常用Buffer 类型有:
ByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer
网络读写最常用的是ByteBuffer。
使用 NIO Buffer 的步骤如下:

  • 将数据写入到 Buffer 中.
  • 调用 Buffer.flip()方法, 将 NIO Buffer 转换为读模式.
  • 从 Buffer 中读取数据
  • 调用 Buffer.clear() 或 Buffer.compact()方法(清洗Buffer的操作), 将 Buffer 转换为写模式。case如下:
public class Test {
    public static void main(String[] args) {
        IntBuffer intBuffer = IntBuffer.allocate(2);
        intBuffer.put(12345678);
        intBuffer.put(2);
        intBuffer.flip();
        System.err.println(intBuffer.get());
        System.err.println(intBuffer.get());
    }
}
Buffer属性
  • capacity
  • position
  • limit
1、Capacity

一个内存块会有一个固定的大小, 即容量(capacity), 我们最多写入capacity 个单位的数据到 Buffer 中, 例如一个 DoubleBuffer, 其 Capacity 是100, 那么我们最多可以写入100个 double 数据.

2、Position

当从一个 Buffer 中写入数据时, 我们是从 Buffer 的一个确定的位置(position)开始写入的. 在最初的状态时, position 的值是0. 每当我们写入了一个单位的数据后, position 就会递增一.
当我们从 Buffer 中读取数据时, 我们也是从某个特定的位置开始读取的. 当我们调用了 filp()方法将 Buffer 从写模式转换到读模式时, position 的值会自动被设置为0, 每当我们读取一个单位的数据, position 的值递增1.
position 表示了读写操作的位置指针.

3、limit

limit - position 表示此时还可以写入/读取多少单位的数据.
例如在写模式, 如果此时 limit 是10, position 是2, 则表示已经写入了2个单位的数据, 还可以写入 10 - 2 = 8 个单位的数据.

分配Buffer
ByteBuffer buf = ByteBuffer.allocate(48);

上述代买分配了48 * sizeof(Byte)字节的内存空间。
在这里稍微提一下关于 Direct Buffer 和 Non-Direct Buffer 的区别:
Direct Buffer:

所分配的内存不在 JVM 堆上, 不受 GC 的管理.(但是 Direct Buffer 的 Java 对象是由 GC 管理的, 因此当发生 GC, 对象被回收时, Direct Buffer也会被释放,因为对象关联了堆外内存,详情可见:http://www.importnew.com/26334.html)
因为 Direct Buffer 不在 JVM 堆上分配, 因此 Direct Buffer 对应用程序的内存占用的影响就不那么明显(实际上还是占用了这么多内存, 但是 JVM 不好统计到非 JVM 管理的内存.)
申请和释放 Direct Buffer 的开销比较大. 因此正确的使用 Direct Buffer 的方式是在初始化时申请一个 Buffer, 然后不断复用此 buffer, 在程序结束后才释放此 buffer.
使用 Direct Buffer 时, 当进行一些底层的系统 IO 操作时, 效率会比较高, 因为此时 JVM 不需要拷贝 buffer 中的内存到中间临时缓冲区中.

Non-Direct Buffer:

直接在 JVM 堆上进行内存的分配, 本质上是 byte[] 数组的封装.
因为Non-Direct Buffer在JVM堆中, 因此当进行操作系统底层IO操作中时, 会将此 buffer 的内存复制到中间临时缓冲区中. 因此Non-Direct Buffer的效率就较低.

写入数据到 Buffer
int bytesRead = inChannel.read(buf); 
buf.put(127);
从 Buffer 中读取数据
int bytesWritten = inChannel.write(buf);
byte aByte = buf.get();
重置 position

Buffer.rewind()方法可以重置 position 的值为0, 因此我们可以重新读取/写入 Buffer 了.rewind() 主要针对于读模式. 在读模式时, 读取到 limit 后, 可以调用 rewind() 方法, 将读 position 置为0.
如果是读模式, 则重置的是读模式的 position, 如果是写模式, 则重置的是写模式的 position.

remark()和reset()
public class Test {
    public static void main(String[] args) {
        IntBuffer intBuffer = IntBuffer.allocate(2);
        intBuffer.put(1);
        intBuffer.put(2);
        intBuffer.flip();
        System.out.println(intBuffer.get());
        System.out.println("position: " + intBuffer.position());

        intBuffer.mark();
        System.out.println(intBuffer.get());

        System.out.println("position: " + intBuffer.position());
        intBuffer.reset();
        System.out.println("position: " + intBuffer.position());
        System.out.println(intBuffer.get());
    }
}

result is:

1
position: 1
2
position: 2
position: 1
2

Buffer.mark()将当前的 position 的值保存起来, 随后可以通过调用 Buffer.reset()方法将 position 的值回复回来。

flip、rewind和clear区别

flip 方法源码

public final Buffer flip() {
    limit = position;
    position = 0;
    mark = -1;
    return this;
}

Buffer 的读/写模式共用一个 position 和 limit 变量.
当从写模式变为读模式时, 原先的 写 position 就变成了读模式的 limit.
rewind 方法源码

public final Buffer rewind() {
    position = 0;
    mark = -1;
    return this;
}

rewind, 即倒带, 这个方法仅仅是将 position 置为0.
clear 方法源码:

public final Buffer clear() {
    position = 0;
    limit = capacity;
    mark = -1;
    return this;
}

根据源码我们可以知道, clear 将 positin 设置为0, 将 limit 设置为 capacity.
clear 方法使用场景:
在一个已经写满数据的 buffer 中, 调用 clear, 可以从头读取 buffer 的数据.
为了将一个 buffer 填充满数据, 可以调用 clear, 然后一直写入, 直到达到 limit.

selector

Selector 允许一个单一的线程来操作多个 Channel. 其实就是Linux五种IO模型中的多路复用模型,异步非阻塞。如果我们的应用程序中使用了多个 Channel,那么使用 Selector 很方便的实现这样的目的, 但是因为在一个线程中使用了多个 Channel, 因此也会造成了每个 Channel 传输效率的降低.
使用 Selector 的图解如下:



为了使用 Selector, 我们首先需要将 Channel 注册到 Selector 中, 随后调用 Selector 的 select()方法, 这个方法会阻塞, 直到注册在 Selector 中的 Channel 发送可读写事件. 当这个方法返回后, 当前的这个线程就可以处理 Channel 的事件了。

1、创建选择器selector

Selector selector = Selector.open();

2、把channel注册到selector

channel.configureBlocking(false);
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

如果一个 Channel 要注册到 Selector 中, 那么这个 Channel 必须是非阻塞的, 即channel.configureBlocking(false);
因为 Channel 必须要是非阻塞的, 因此 FileChannel 是不能够使用选择器的, 因为 FileChannel 都是阻塞的.

Channel.register()的第二个参数是指定了我们对 Channel 的什么类型的事件感兴趣, 后续会根据感兴趣的时间进行逻辑处理,这些事件有:

  • Connect, 即连接事件(TCP 连接), 表示客户与服务器的连接已经建立成功,对应于SelectionKey.OP_CONNECT
  • Accept, 即确认事件,表示服务器监听到了客户连接,服务器可以接收这个连接了, 对应于SelectionKey.OP_ACCEPT
  • Read, 即读事件, 对应于SelectionKey.OP_READ, 表示 buffer 可读.
  • Write, 即写事件, 对应于SelectionKey.OP_WRITE, 表示 buffer 可写.

一个 Channel发出一个事件也可以称为 对于某个事件, Channel 准备好了. 因此一个 Channel 成功连接到了另一个服务器也可以被称为 connect ready.

SelectionKey

channel注册到selector时会返回SelectionKey,该对象包含的内容有:

  • interest set, 即我们感兴趣的事件集, 即在调用 register 注册 channel 时所设置- 的 interest set.
  • ready set
  • channel
  • selector
  • attached object, 可选的附加对象
interest set

我们可以通过如下方式获取 interest set:

int interestSet = selectionKey.interestOps();
boolean isInterestedInAccept  = interestSet & SelectionKey.OP_ACCEPT;
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;   
ready set

代表了 Channel 所准备好了的操作。可以使用如下方法进行判断:

int readySet = selectionKey.readyOps();
selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();
Attaching Object

我们可以在selectionKey中附加一个对象:

selectionKey.attach(theObject);
Object attachedObj = selectionKey.attachment();

或者在注册时直接附加:

SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

3、获取可以操作channel

select()方法返回的值表示有多少个 Channel 可操作。如果 select()方法返回值表示有多个 Channel 准备好了, 那么我们可以通过 Selected key set 访问这个 Channel:

Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
    SelectionKey key = keyIterator.next();
    if(key.isAcceptable()) {
        // 。。。
    } else if (key.isConnectable()) {
        // 。。。
    } else if (key.isReadable()) {
        // 。。。
    } else if (key.isWritable()) {
        // 。
    }
    keyIterator.remove();
}

注意, 在每次迭代时, 我们都调用 "keyIterator.remove()" 将这个 key 从迭代器中删除, 因为 select() 方法仅仅是简单地将就绪的 IO 操作放到 selectedKeys 集合中, 因此如果我们从 selectedKeys 获取到一个 key, 但是没有将它删除, 那么下一次 select 时, 这个 key 所对应的 IO 事件还在 selectedKeys 中。

selector的使用流程(重点)

1、通过 Selector.open() 打开一个 Selector.
2、将 Channel 注册到 Selector 中, 并设置需要监听的事件(interest set)
3、循环做以下流程:
  1)、调用 select() 方法
  2)、调用 selector.selectedKeys() 获取 selected keys
  3)、迭代每个 selected key:
    1)、从 selected key 中获取 对应的 Channel 和附加信息(如果有的话)
    2)、判断是哪些 IO 事件已经就绪了, 然后处理它们. 如果是 OP_ACCEPT 事件, 则调用 "SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept()" 获取 SocketChannel, 并将它设置为 非阻塞的, 然后将这个 Channel 注册到 Selector 中.
    3)、根据需要更改 selected key 的监听事件.
    4)、将已经处理过的 key 从 selected keys 集合中删除.
以下是完整的例子:

public class NioEchoServer {
    private static final int BUF_SIZE = 256;
    private static final int TIMEOUT = 3000;
    public static void main(String args[]) throws Exception {
        // 打开服务端 Socket
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 打开 Selector
        Selector selector = Selector.open();
        // 服务端 Socket 监听8080端口, 并配置为非阻塞模式
        serverSocketChannel.socket().bind(new InetSocketAddress(8080));
        serverSocketChannel.configureBlocking(false);
        // 将 channel 注册到 selector 中.
        // 通常我们都是先注册一个 OP_ACCEPT 事件, 然后在 OP_ACCEPT 到来时, 再将这个 Channel 的 OP_READ
        // 注册到 Selector 中.
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        while (true) {
            // 通过调用 select 方法, 阻塞地等待 channel I/O 可操作
            if (selector.select(TIMEOUT) == 0) {
                System.out.print(".");
                continue;
            }
            // 获取 I/O 操作就绪的 SelectionKey, 通过 SelectionKey 可以知道哪些 Channel 的哪类 I/O 操作已经就绪.
            Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();
                // 当获取一个 SelectionKey 后, 就要将它删除, 表示我们已经对这个 IO 事件进行了处理.
                keyIterator.remove();
                if (key.isAcceptable()) {
                    // 当 OP_ACCEPT 事件到来时, 我们就有从 ServerSocketChannel 中获取一个 SocketChannel,
                    // 代表客户端的连接
                    // 注意, 在 OP_ACCEPT 事件中, 从 key.channel() 返回的 Channel 是 ServerSocketChannel.
                    // 而在 OP_WRITE 和 OP_READ 中, 从 key.channel() 返回的是 SocketChannel.
                    SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
                    clientChannel.configureBlocking(false);
                    //在 OP_ACCEPT 到来时, 再将这个 Channel 的 OP_READ 注册到 Selector 中.
                    // 注意, 这里我们如果没有设置 OP_READ 的话, 即 interest set 仍然是 OP_CONNECT 的话, 那么 select 方法会一直直接返回.
                    clientChannel.register(key.selector(), OP_READ, ByteBuffer.allocate(BUF_SIZE));
                }
                if (key.isReadable()) {
                    SocketChannel clientChannel = (SocketChannel) key.channel();
                    ByteBuffer buf = (ByteBuffer) key.attachment();
                    long bytesRead = clientChannel.read(buf);
                    if (bytesRead == -1) {
                        clientChannel.close();
                    } else if (bytesRead > 0) {
                        key.interestOps(OP_READ | SelectionKey.OP_WRITE);
                        System.out.println("Get data length: " + bytesRead);
                    }
                }
                if (key.isValid() && key.isWritable()) {
                    ByteBuffer buf = (ByteBuffer) key.attachment();
                    buf.flip();
                    SocketChannel clientChannel = (SocketChannel) key.channel();
                    clientChannel.write(buf);
                    if (!buf.hasRemaining()) {
                        key.interestOps(OP_READ);
                    }
                    buf.compact();
                }
            }
        }
    }
}
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 204,293评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,604评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,958评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,729评论 1 277
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,719评论 5 366
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,630评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,000评论 3 397
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,665评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,909评论 1 299
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,646评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,726评论 1 330
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,400评论 4 321
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,986评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,959评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,197评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 44,996评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,481评论 2 342

推荐阅读更多精彩内容

  • 作者: 一字马胡 转载标志 【2017-11-24】 更新日志 一、Java OIO Java OIO (Jav...
    一字马胡阅读 1,337评论 0 12
  • Java NIO(New IO)是从Java 1.4版本开始引入的一个新的IO API,可以替代标准的Java I...
    zhisheng_blog阅读 1,112评论 0 7
  • Java NIO(New IO)是从Java 1.4版本开始引入的一个新的IO API,可以替代标准的Java I...
    编码前线阅读 2,258评论 0 5
  • Java NIO(New IO)是从Java 1.4版本开始引入的一个新的IO API,可以替代标准的Java I...
    JackChen1024阅读 7,534评论 1 143
  • 该落的已经落下了 只留光秃秃的枝丫 在刺骨的寒风中 瑟瑟发抖 一地的落叶 在未找到根的归宿之时 已被扫除了无名的幻...
    冯圈圈阅读 319评论 0 1