第3章
Java NIO通信基础详解
高性能的Java通信,绝对离不开Java NIO技术,现在主流的技术框架或中间件服务器,都使用了Java NIO技术,譬如Tomcat、Jetty、Netty。学习和掌握NIO技术,已经不是一项加分技能,而是一项必备技能。不管是面试,还是实际开发,作为Java的“攻城狮”(工程师的谐音),都必须掌握NIO的原理和开发实践技能。
3.1 Java NIO简介
在1.4版本之前,Java IO类库是阻塞IO;从1.4版本开始,引进了新的异步IO库,被称为Java New IO类库,简称为JAVA NIO。New IO类库的目标,就是要让Java支持非阻塞IO,基于这个原因,更多的人喜欢称Java NIO为非阻塞IO(Non-Block IO),称“老的”阻塞式Java IO为OIO(Old IO)。总体上说,NIO弥补了原来面向流的OIO同步阻塞的不足,它为标准Java代码提供了高速的、面向缓冲区的IO。
Java NIO由以下三个核心组件组成:
- Channel(通道)
- Buffer(缓冲区)
- Selector(选择器)
如果理解了第1章的四种IO模型,大家一眼就能识别出来,Java NIO,属于第三种模型—— IO 多路复用模型。当然,Java NIO组件,提供了统一的API,为大家屏蔽了底层的不同操作系统的差异。
后面的章节,我们会对以上的三个Java NIO的核心组件,展开详细介绍。先来看看Java的NIO和OIO的简单对比。
3.1.1 NIO和OIO的对比
在Java中,NIO和OIO的区别,主要体现在三个方面:
(1)OIO是面向流(Stream Oriented)的,NIO是面向缓冲区(Buffer Oriented)的。
何谓面向流,何谓面向缓冲区呢?
OIO是面向字节流或字符流的,在一般的OIO操作中,我们以流式的方式顺序地从一个流(Stream)中读取一个或多个字节,因此,我们不能随意地改变读取指针的位置。而在NIO操作中则不同,NIO中引入了Channel(通道)和Buffer(缓冲区)的概念。读取和写入,只需要从通道中读取数据到缓冲区中,或将数据从缓冲区中写入到通道中。NIO不像OIO那样是顺序操作,可以随意地读取Buffer中任意位置的数据。
(2)OIO的操作是阻塞的,而NIO的操作是非阻塞的。
NIO如何做到非阻塞的呢?大家都知道,OIO操作都是阻塞的,例如,我们调用一个read方法读取一个文件的内容,那么调用read的线程会被阻塞住,直到read操作完成。
而在NIO的非阻塞模式中,当我们调用read方法时,如果此时有数据,则read读取数据并返回;如果此时没有数据,则read直接返回,而不会阻塞当前线程。NIO的非阻塞,是如何做到的呢?其实在上一章,答案已经揭晓了,NIO使用了通道和通道的多路复用技术。
(3)OIO没有选择器(Selector)概念,而NIO有选择器的概念。
NIO的实现,是基于底层的选择器的系统调用。NIO的选择器,需要底层操作系统提供支持。而OIO不需要用到选择器。
3.1.2 通道(Channel)
在OIO中,同一个网络连接会关联到两个流:一个输入流(Input Stream),另一个输出流(Output Stream)。通过这两个流,不断地进行输入和输出的操作。
在NIO中,同一个网络连接使用一个通道表示,所有的NIO的IO操作都是从通道开始的。一个通道类似于OIO中的两个流的结合体,既可以从通道读取,也可以向通道写入。
3.1.3 Selector 选择器
首先,回顾一个基础的问题,什么是IO多路复用?指的是一个进程/线程可以同时监视多个文件描述符(一个网络连接,操作系统底层使用一个文件描述符来表示),一旦其中的一个或者多个文件描述符可读或者可写,系统内核就通知该进程/线程。在Java应用层面,如何实现对多个文件描述符的监视呢?需要用到一个非常重要的Java NIO组件——Selector 选择器。
选择器的神奇功能是什么呢?它一个IO事件的查询器。通过选择器,一个线程可以查询多个通道的IO事件的就绪状态。
实现IO多路复用,从具体的开发层面来说,首先把通道注册到选择器中,然后通过选择器内部的机制,可以查询(select)这些注册的通道是否有已经就绪的IO事件(例如可读、可写、网络连接完成等)。
一个选择器只需要一个线程进行监控,换句话说,我们可以很简单地使用一个线程,通过选择器去管理多个通道。这是非常高效的,这种高效来自于Java的选择器组件Selector,以及其背后的操作系统底层的IO多路复用的支持。
与OIO相比,使用选择器的最大优势:系统开销小,系统不必为每一个网络连接(文件描述符)创建进程/线程,从而大大减小了系统的开销。
3.1.4 缓冲区(Buffer)
应用程序与通道(Channel)主要的交互操作,就是进行数据的read读取和write写入。为了完成如此大任,NIO为大家准备了第三个重要的组件——NIO Buffer(NIO缓冲区)。通道的读取,就是将数据从通道读取到缓冲区中;通道的写入,就是将数据从缓冲区中写入到通道中。
缓冲区的使用,是面向流的OIO所没有的,也是NIO非阻塞的重要前提和基础之一。
下面从缓冲区开始,详细介绍NIO的Buffer(缓冲区)、Channel(通道)、Selector(选择器)三大核心组件。
3.2 详解NIO Buffer类及其属性
NIO的Buffer(缓冲区)本质上是一个内存块,既可以写入数据,也可以从中读取数据。NIO的Buffer类,是一个抽象类,位于java.nio包中,其内部是一个内存块(数组)。
NIO的Buffer与普通的内存块(Java数组)不同的是:NIO Buffer对象,提供了一组更加有效的方法,用来进行写入和读取的交替访问。
需要强调的是:Buffer类是一个非线程安全类。
3.2.1 Buffer类
Buffer类是一个抽象类,对应于Java的主要数据类型,在NIO中有8种缓冲区类,分别如下:ByteBuffer、CharBuffer、DoubleBuffer、FloatBuffer、IntBuffer、LongBuffer、ShortBuffer、MappedByteBuffer。
前7种Buffer类型,覆盖了能在IO中传输的所有的Java基本数据类型。第8种类型MappedByteBuffer是专门用于内存映射的一种ByteBuffer类型。
实际上,使用最多的还是ByteBuffer二进制字节缓冲区类型,后面会看到。
3.2.2 Buffer类的重要属性
Buffer类在其内部,有一个byte[]数组内存块,作为内存缓冲区。为了记录读写的状态和位置,Buffer类提供了一些重要的属性。其中,有三个重要的成员属性:capacity(容量)、position(读写位置)、limit(读写的限制)。
除此之外,还有一个标记属性:mark(标记),可以将当前的position临时存入mark中;需要的时候,可以再从mark标记恢复到position位置。
1. capacity属性
Buffer类的capacity属性,表示内部容量的大小。一旦写入的对象数量超过了capacity容量,缓冲区就满了,不能再写入了。
Buffer类的capacity属性一旦初始化,就不能再改变。原因是什么呢?Buffer类的对象在初始化时,会按照capacity分配内部的内存。在内存分配好之后,它的大小当然就不能改变了。
再强调一下,capacity容量不是指内存块byte[]数组的字节的数量。capacity容量指的是写入的数据对象的数量。
前面讲到,Buffer类是一个抽象类,Java不能直接用来新建对象。使用的时候,必须使用Buffer的某个子类,例如使用DoubleBuffer,则写入的数据是double类型,如果其capacity是100,那么我们最多可以写入100个double数据。
2. position属性
Buffer类的position属性,表示当前的位置。position属性与缓冲区的读写模式有关。在不同的模式下,position属性的值是不同的。当缓冲区进行读写的模式改变时,position会进行调整。
在写入模式下,position的值变化规则如下:(1)在刚进入到写模式时,position值为0,表示当前的写入位置为从头开始。(2)每当一个数据写到缓冲区之后,position会向后移动到下一个可写的位置。(3)初始的position值为0,最大可写值position为limit– 1。当position值达到limit时,缓冲区就已经无空间可写了。
在读模式下,position的值变化规则如下:(1)当缓冲区刚开始进入到读模式时,position会被重置为0。(2)当从缓冲区读取时,也是从position位置开始读。读取数据后,position向前移动到下一个可读的位置。(3)position最大的值为最大可读上限limit,当position达到limit时,表明缓冲区已经无数据可读。
起点在哪里呢?当新建一个缓冲区时,缓冲区处于写入模式,这时是可以写数据的。数据写入后,如果要从缓冲区读取数据,这就要进行模式的切换,可以使用(即调用)flip翻转方法,将缓冲区变成读取模式。
在这个flip翻转过程中,position会进行非常巨大的调整,具体的规则是:position由原来的写入位置,变成新的可读位置,也就是0,表示可以从头开始读。flip翻转的另外一半工作,就是要调整limit属性。
3. limit属性
Buffer类的limit属性,表示读写的最大上限。limit属性,也与缓冲区的读写模式有关。在不同的模式下,limit的值的含义是不同的。
在写模式下,limit属性值的含义为可以写入的数据最大上限。在刚进入到写模式时,limit的值会被设置成缓冲区的capacity容量值,表示可以一直将缓冲区的容量写满。
在读模式下,limit的值含义为最多能从缓冲区中读取到多少数据。
一般来说,是先写入再读取。当缓冲区写入完成后,就可以开始从Buffer读取数据,可以使用flip翻转方法,这时,limit的值也会进行非常大的调整。
具体如何调整呢?将写模式下的position值,设置成读模式下的limit值,也就是说,将之前写入的最大数量,作为可以读取的上限值。
在flip翻转时,属性的调整,将涉及position、limit两个属性,这种调整比较微妙,不是太好理解,举一个简单例子:
首先,创建缓冲区。刚开始,缓冲区处于写模式。position为0,limit为最大容量。
然后,向缓冲区写数据。每写入一个数据,position向后面移动一个位置,也就是position的值加1。假定写入了5个数,当写入完成后,position的值为5。
这时,使用(即调用)flip方法,将缓冲区切换到读模式。limit的值,先会被设置成写模式时的position值。这里新的limit是5,表示可以读取的最大上限是5个数。同时,新的position会被重置为0,表示可以从0开始读。
3.2.3 4个属性的小结
除了前面的3个属性,第4个属性mark(标记)比较简单。就是相当一个暂存属性,暂时保存position的值,方便后面的重复使用position值。
下面用一个表格总结一下 Buffer类的4个重要属性,参见表3-1。
表3-1 Buffer四个重要属性的取值说明
3.3 详解NIO Buffer类的重要方法
本小节将详细介绍Buffer类使用中常用的几个方法,包含Buffer实例的获取、对Buffer实例的写入、读取、重复读、标记和重置等。
3.3.1 allocate()创建缓冲区
在使用Buffer(缓冲区)之前,我们首先需要获取Buffer子类的实例对象,并且分配内存空间。
为了获取一个Buffer实例对象,这里并不是使用子类的构造器new来创建一个实例对象,而是调用子类的allocate()方法。
下面的程序片段就是用来获取一个整型Buffer类的缓冲区实例对象,代码如下:
package com.crazymakercircle.bufferDemo;
//...
public class UseBuffer
{
static IntBufferintBuffer = null;
public static void allocatTest()
{
//调用allocate方法,而不是使用new
intBuffer = IntBuffer.allocate(20);
//输出buffer的主要属性值
Logger.info("------------after allocate------------------");
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
}
//...
}
例子中,IntBuffer是具体的Buffer子类,通过调用IntBuffer.allocate(20),创建了一个Intbuffer实例对象,并且分配了20 * 4个字节的内存空间。
通过程序的输出结果,我们可以查看一个新建缓冲区实例对象的主要属性值,如下所示:
allocatTest |> ------------after allocate------------------
allocatTest |> position=0
allocatTest |> limit=20
allocatTest |> capacity=20
从上面的运行结果,可以看出:
一个缓冲区在新建后,处于写入的模式,position写入位置为0,最大可写上限limit为的初始化值(这里是20),而缓冲区的容量capacity也是初始化值。
3.3.2 put()写入到缓冲区
在调用allocate方法分配内存、返回了实例对象后,缓冲区实例对象处于写模式,可以写入对象。要写入缓冲区,需要调用put方法。put方法很简单,只有一个参数,即为所需要写入的对象。不过,写入的数据类型要求与缓冲区的类型保持一致。
接着前面的例子,向刚刚创建的intBuffer缓存实例对象中,写入的5个整数,代码如下:
package com.crazymakercircle.bufferDemo;
//...
public class UseBuffer
{
static IntBufferintBuffer = null;
//省略了创建缓冲区的代码,具体看源代码工程
public static void putTest()
{
for (int i = 0; i< 5; i++)
{
//写入一个整数到缓冲区
intBuffer.put(i);
}
//输出缓冲区的主要属性值
Logger.info("------------after put------------------");
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
}
//...
}
写入5个元素后,同样输出缓冲区的主要属性值,输出的结果如下:
putTest |> ------------after putTest------------------
putTest |> position=5
putTest |> limit=20
putTest |> capacity=20
从结果可以看到,position变成了5,指向了第6个可以写入的元素位置。而limit最大写入元素的上限、capacity最大容量的值,并没有发生变化。
3.3.3 flip()翻转
向缓冲区写入数据之后,是否可以直接从缓冲区中读取数据呢?呵呵,不能。
这时缓冲区还处于写模式,如果需要读取数据,还需要将缓冲区转换成读模式。flip()翻转方法是Buffer类提供的一个模式转变的重要方法,它的作用就是将写入模式翻转成读取模式。
接着前面的例子,演示一下flip()方法的使用:
package com.crazymakercircle.bufferDemo;
//...
public class UseBuffer
{
static IntBufferintBuffer = null;
//省略了缓冲区的创建、写入的代码,具体看源代码工程
public static void flipTest()
{
//翻转缓冲区,从写模式翻转成读模式
intBuffer.flip();
//输出缓冲区的主要属性值
Logger.info("------------after flip ------------------");
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
}
//...
}
在调用flip进行模式翻转之后,缓冲区的属性有了奇妙的变化,输出如下:
flipTest |> ------------after flipTest ------------------
flipTest |> position=0
flipTest |> limit=5
flipTest |> capacity=20
调用flip方法后,之前写入模式下的position值5,变成了可读上限limit值5;而新的读取模式下的position值,简单粗暴地变成了0,表示从头开始读取。
对flip()方法的从写入到读取转换的规则,详细的介绍如下:
首先,设置可读的长度上限limit。将写模式下的缓冲区中内容的最后写入位置position值,作为读模式下的limit上限值。
其次,把读的起始位置position的值设为0,表示从头开始读。
最后,清除之前的mark标记,因为mark保存的是写模式下的临时位置。在读模式下,如果继续使用旧的mark标记,会造成位置混乱。
有关上面的三步,其实可以查看flip方法的源代码,Buffer.flip()方法的源代码如下:
public final Buffer flip() {
limit = position; //设置可读的长度上限limit,为写入的position
position = 0; //把读的起始位置position的值设为0,表示从头开始读
mark = UNSET_MARK; // 清除之前的mark标记
return this;
}
至此,大家都知道了,如何将缓冲区切换成读取模式。
新的问题来了,在读取完成后,如何再一次将缓冲区切换成写入模式呢?可以调用Buffer.clear() 清空或者Buffer.compact()压缩方法,它们可以将缓冲区转换为写模式。
Buffer的模式转换,大致如图3-1所示。
图3-1 缓冲区读写模式的转换
3.3.4 get()从缓冲区读取
调用flip方法,将缓冲区切换成读取模式。这时,可以开始从缓冲区中进行数据读取了。读数据很简单,调用get方法,每次从position的位置读取一个数据,并且进行相应的缓冲区属性的调整。
接着前面flip的使用实例,演示一下缓冲区的读取操作,代码如下:
package com.crazymakercircle.bufferDemo;
//...
public class UseBuffer
{
static IntBufferintBuffer = null;
//省略了缓冲区的创建、写入、翻转的代码,具体看源代码工程
public static void getTest()
{
//先读2个
for (int i = 0; i< 2; i++)
{
int j = intBuffer.get();
Logger.info("j = " + j);
}
//输出缓冲区的主要属性值
Logger.info("------------after get 2 int ------------------");
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
//再读3个
for (int i = 0; i< 3; i++)
{
int j = intBuffer.get();
Logger.info("j = " + j);
}
//输出缓冲区的主要属性值
Logger.info("------------after get 3 int ------------------");
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
}
//...
}
先读2个,再读3个,运行后,输出的结果如下:
getTest |> ------------after get 2 int ------------------
getTest |> position=2
getTest |> limit=5
getTest |> capacity=20
getTest |> ------------after get 3 int ------------------
getTest |> position=5
getTest |> limit=5
getTest |> capacity=20
从程序的输出结果,我们可以看到,读取操作会改变可读位置position的值,而limit值不会改变。如果position值和limit的值相等,表示所有数据读取完成,position指向了一个没有数据的元素位置,已经不能再读了。此时再读,会抛出BufferUnderflowException异常。
这里强调一下,在读完之后,是否可以立即进行写入模式呢?不能。现在还处于读取模式,我们必须调用Buffer.clear()或Buffer.compact(),即清空或者压缩缓冲区,才能变成写入模式,让其重新可写。
另外,还有一个问题:缓冲区是不是可以重复读呢?答案是可以的。
3.3.5 rewind()倒带
已经读完的数据,如果需要再读一遍,可以调用rewind()方法。rewind()也叫倒带,就像播放磁带一样倒回去,再重新播放。
接着前面的代码,继续rewind方法使用的演示,示例代码如下:
package com.crazymakercircle.bufferDemo;
//...
public class UseBuffer
{
static IntBufferintBuffer = null;
//省略了缓冲区的创建、写入、读取的代码,具体看源代码工程
public static void rewindTest() {
//倒带
intBuffer.rewind();
//输出缓冲区属性
Logger.info("------------after rewind ------------------");
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
}
//...
}
这个范例程序的执行结果如下:
rewindTest |> ------------after rewind ------------------
rewindTest |> position=0
rewindTest |> limit=5
rewindTest |> capacity=20
rewind ()方法,主要是调整了缓冲区的position属性,具体的调整规则如下:
(1)position重置为0,所以可以重读缓冲区中的所有数据。
(2)limit保持不变,数据量还是一样的,仍然表示能从缓冲区中读取多少个元素。
(3)mark标记被清理,表示之前的临时位置不能再用了。
Buffer.rewind()方法的源代码如下:
public final Buffer rewind() {
position = 0;//重置为0,所以可以重读缓冲区中的所有数据
mark = -1; // mark标记被清理,表示之前的临时位置不能再用了
return this;
}
通过源代码,我们可以看到rewind()方法与flip()很相似,区别在于:rewind()不会影响limit属性值;而flip()会重设limit属性值。
在rewind倒带之后,就可以再一次读取,重复读取的示例代码如下:
package com.crazymakercircle.bufferDemo;
//...
public class UseBuffer
{
static IntBufferintBuffer = null;
//省略了缓冲区的读取、倒带的代码,具体看源代码工程
public static void reRead() {
for (int i = 0; i< 5; i++) {
if (i == 2) {
//临时保存,标记一下第3个位置
intBuffer.mark();
}
//读取元素
int j = intBuffer.get();
Logger.info("j = " + j);
}
//输出缓冲区的属性值
Logger.info("------------after reRead------------------");
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
}
//...
}
这段代码,和前面的读取示例代码基本相同,只是增加了一个mark调用。
3.3.6 mark( )和reset( )
Buffer.mark()方法的作用是将当前position的值保存起来,放在mark属性中,让mark属性记住这个临时位置;之后,可以调用Buffer.reset()方法将mark的值恢复到position中。
也就是说,Buffer.mark()和Buffer.reset()方法是配套使用的。两种方法都需要内部mark属性的支持。
在前面重复读取缓冲区的示例代码中,读到第3个元素(i= =2时),调用mark()方法,把当前位置position的值保存到mark属性中,这时mark属性的值为2。
接下来,就可以调用reset方法,将mark属性的值恢复到position中。然后可以从位置2(第三个元素)开始读。
继续接着前面的重复读取的代码,进行reset的示例演示,代码如下:
package com.crazymakercircle.bufferDemo;
//...
public class UseBuffer
{
static IntBufferintBuffer = null;
//省略了缓冲区之前的mark等代码,具体看源代码工程
public static void afterReset() {
Logger.info("------------after reset------------------");
//把前面保存在mark中的值恢复到position
intBuffer.reset();
//输出缓冲区的属性值
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
//读取并且输出元素
for (int i =2; i< 5; i++) {
int j = intBuffer.get();
Logger.info("j = " + j);
}
}
//...
}
在上面的代码中,首先调用reset()把mark中的值恢复到position中,因此读取的位置position就是2,表示可以再次开始从第3个元素开始读取数据。上面的程序代码的输出结果是:
afterReset |> ------------after reset------------------
afterReset |> position=2
afterReset |> limit=5
afterReset |> capacity=20
afterReset |> j = 2
afterReset |> j = 3
afterReset |> j = 4
调用reset方法之后,position的值为2。此时去读取缓冲区,输出后面的三个元素为2、3、4。
3.3.7 clear( )清空缓冲区
在读取模式下,调用clear()方法将缓冲区切换为写入模式。此方法会将position清零,limit设置为capacity最大容量值,可以一直写入,直到缓冲区写满。
接着上面的实例,演示一下clear方法。代码如下:
package com.crazymakercircle.bufferDemo;
//...
public class UseBuffer
{
static IntBufferintBuffer = null;
//省略了之前的buffer操作代码,具体看源代码工程
public static void clearDemo() {
Logger.info("------------after clear------------------");
//清空缓冲区,进入写入模式
intBuffer.clear();
//输出缓冲区的属性值
Logger.info("position=" + intBuffer.position());
Logger.info("limit=" + intBuffer.limit());
Logger.info("capacity=" + intBuffer.capacity());
}
//...
}
这个程序运行之后,结果如下:
main |>清空
clearDemo |> ------------after clear------------------
clearDemo |> position=0
clearDemo |> limit=20
clearDemo |> capacity=20
在缓冲区处于读取模式时,调用clear(),缓冲区会被切换成写入模式。调用clear()之后,我们可以看到清空了position的值,即设置写入的起始位置为0,并且写入的上限为最大容量。
3.3.8 使用Buffer类的基本步骤
总体来说,使用Java NIO Buffer类的基本步骤如下:
(1)使用创建子类实例对象的allocate()方法,创建一个Buffer类的实例对象。
(2)调用put方法,将数据写入到缓冲区中。
(3)写入完成后,在开始读取数据前,调用Buffer.flip()方法,将缓冲区转换为读模式。
(4)调用get方法,从缓冲区中读取数据。
(5)读取完成后,调用Buffer.clear() 或Buffer.compact()方法,将缓冲区转换为写入模式。
3.4 详解NIO Channel(通道)类
前面讲到,NIO中一个连接就是用一个Channel(通道)来表示。大家知道,从更广泛的层面来说,一个通道可以表示一个底层的文件描述符,例如硬件设备、文件、网络连接等。然而,远远不止如此,除了可以对应到底层文件描述符,Java NIO的通道还可以更加细化。例如,对应不同的网络传输协议类型,在Java中都有不同的NIO Channel(通道)实现。
3.4.1 Channel(通道)的主要类型
这里不对纷繁复杂的Java NIO通道类型进行过多的描述,仅仅聚焦于介绍其中最为重要的四种Channel(通道)实现:FileChannel、SocketChannel、ServerSocketChannel、DatagramChannel。
对于以上四种通道,说明如下:
(1)FileChannel文件通道,用于文件的数据读写。
(2)SocketChannel套接字通道,用于Socket套接字TCP连接的数据读写。
(3)ServerSocketChannel服务器嵌套字通道(或服务器监听通道),允许我们监听TCP连接请求,为每个监听到的请求,创建一个SocketChannel套接字通道。
(4)DatagramChannel数据报通道,用于UDP协议的数据读写。
这个四种通道,涵盖了文件IO、TCP网络、UDP IO基础IO。下面从Channel(通道)的获取、读取、写入、关闭四个重要的操作,来对四种通道进行简单的介绍。
3.4.2 FileChannel文件通道
FileChannel是专门操作文件的通道。通过FileChannel,既可以从一个文件中读取数据,也可以将数据写入到文件中。特别申明一下,FileChannel为阻塞模式,不能设置为非阻塞模式。
下面分别介绍:FileChannel的获取、读取、写入、关闭四个操作。
1. 获取FileChannel通道
可以通过文件的输入流、输出流获取FileChannel文件通道,示例如下:
//创建一条文件输入流
FileInputStreamfis = new FileInputStream(srcFile);
//获取文件流的通道
FileChannelinChannel = fis.getChannel();
//创建一条文件输出流
FileOutputStreamfos = new FileOutputStream(destFile);
//获取文件流的通道
FileChanneloutchannel = fos.getChannel();
也可以通过RandomAccessFile文件随机访问类,获取FileChannel文件通道:
// 创建RandomAccessFile随机访问对象
RandomAccessFileaFile = new RandomAccessFile("filename.txt","rw");
//获取文件流的通道
FileChannelinChannel = aFile.getChannel();
2. 读取FileChannel通道
在大部分应用场景,从通道读取数据都会调用通道的int read(ByteBufferbuf)方法,它从通道读取到数据写入到ByteBuffer缓冲区,并且返回读取到的数据量。
RandomAccessFileaFile = new RandomAccessFile(fileName, "rw");
//获取通道
FileChannelinChannel=aFile.getChannel();
//获取一个字节缓冲区
ByteBufferbuf = ByteBuffer.allocate(CAPACITY);
int length = -1;
//调用通道的read方法,读取数据并买入字节类型的缓冲区
while ((length = inChannel.read(buf)) != -1) {
//省略……处理读取到的buf中的数据
}
注 意
虽然对于通道来说是读取数据,但是对于ByteBuffer缓冲区来说是写入数据,这时,ByteBuffer缓冲区处于写入模式。
3. 写入FileChannel通道
写入数据到通道,在大部分应用场景,都会调用通道的int write(ByteBufferbuf)方法。此方法的参数——ByteBuffer缓冲区,是数据的来源。write方法的作用,是从ByteBuffer缓冲区中读取数据,然后写入到通道自身,而返回值是写入成功的字节数。
//如果buf刚写完数据,需要flip翻转buf,使其变成读取模式
buf.flip();
int outlength = 0;
//调用write方法,将buf的数据写入通道
while ((outlength = outchannel.write(buf)) != 0) {
System.out.println("写入的字节数:" + outlength);
}
注 意
此时的ByteBuffer缓冲区要求是可读的,处于读模式下。
4.关闭通道
当通道使用完成后,必须将其关闭。关闭非常简单,调用close方法即可。
//关闭通道
channel.close();
5.强制刷新到磁盘
在将缓冲区写入通道时,出于性能原因,操作系统不可能每次都实时将数据写入磁盘。如果需要保证写入通道的缓冲数据,最终都真正地写入磁盘,可以调用FileChannel的force()方法。
//强制刷新到磁盘
channel.force(true);
3.4.3 使用FileChannel完成文件复制的实践案例
下面是一个简单的实战案例:使用文件通道复制文件。其功能是:使用FileChannel文件通道,将原文件复制一份,也就是把原文中的数据都复制到目标文件中。完整代码如下:
package com.crazymakercircle.iodemo.fileDemos;
//...省略import的类,具体请参见源代码工程
public class FileNIOCopyDemo {
public static void main(String[] args) {
//演示复制资源文件
nioCopyResouceFile();
}
/**
* 复制两个资源目录下的文件
*/
public static void nioCopyResouceFile() {
String sourcePath = NioDemoConfig.FILE_RESOURCE_SRC_PATH;
String srcPath = IOUtil.getResourcePath(sourcePath);
Logger.info("srcPath=" + srcPath);
String destPath = NioDemoConfig.FILE_RESOURCE_DEST_PATH;
String destDecodePath = IOUtil.builderResourcePath(destPath);
Logger.info("destDecodePath=" + destDecodePath);
nioCopyFile(srcDecodePath, destDecodePath);
}
/**
* nio方式复制文件
* @param srcPath
* @param destPath
*/
public static void nioCopyFile(String srcPath, String destPath) {
File srcFile = new File(srcPath);
File destFile = new File(destPath);
try {
//如果目标文件不存在,则新建
if (!destFile.exists()) {
destFile.createNewFile();
}
long startTime = System.currentTimeMillis();
FileInputStreamfis = null;
FileOutputStreamfos = null;
FileChannelinChannel = null;
FileChanneloutchannel = null;
try {
fis = new FileInputStream(srcFile);
fos = new FileOutputStream(destFile);
inChannel = fis.getChannel();
outchannel = fos.getChannel();
int length = -1;
ByteBufferbuf = ByteBuffer.allocate(1024);
//从输入通道读取到buf
while ((length = inChannel.read(buf)) != -1) {
//第一次切换:翻转buf,变成读取模式
buf.flip();
int outlength = 0;
//将buf写入到输出的通道
while ((outlength = outchannel.write(buf)) != 0) {
System.out.println("写入的字节数:" + outlength);
}
//第二次切换:清除buf,变成写入模式
buf.clear();
}
//强制刷新到磁盘
outchannel.force(true);
} finally {
//关闭所有的可关闭对象
IOUtil.closeQuietly(outchannel);
IOUtil.closeQuietly(fos);
IOUtil.closeQuietly(inChannel);
IOUtil.closeQuietly(fis);
}
long endTime = System.currentTimeMillis();
Logger.info("base复制毫秒数:" + (endTime - startTime));
} catch (IOException e) {
e.printStackTrace();
}
}
特别强调一下,除了FileChannel的通道操作外,还需要注意ByteBuffer的模式切换。新建的ByteBuffer,默认是写入模式,可以作为inChannel.read(ByteBuffer)的参数。inChannel.read方法将从通道inChannel读到的数据写入到ByteBuffer。
此后,需要调用缓冲区的flip方法,将ByteBuffer切换成读取模式,才能作为outchannel.write(ByteBuffer)方法的参数,从ByteBuffer读取数据,再写入到outchannel输出通道。
如此,便是完成一次复制。在进入下一次复制前,还要进行一次缓冲区的模式切换。ByteBuffer数据读完之后,需要将通过clear方法切换成写入模式,才能进入下一次的复制。
在示例代码中,外层的每一轮while循环,都需要两次模式ByteBuffer切换:第一次切换时,翻转buf,变成读取模式;第二次切换时,清除buf,变成写入模式。
上面的示例代码,主要的目的在于:演示文件通道以及字节缓冲区的使用。作为文件复制的程序来说,实战代码的效率不是最高的。
更高效的文件复制,可以调用文件通道的transferFrom方法。具体的代码,可以参见源代码工程中的FileNIOFastCopyDemo类,完整源文件的路径为:
com.crazymakercircle.iodemo.fileDemos.FileNIOFastCopyDemo
3.4.4 SocketChannel套接字通道
在NIO中,涉及网络连接的通道有两个,一个是SocketChannel负责连接传输,另一个是ServerSocketChannel负责连接的监听。
NIO中的SocketChannel传输通道,与OIO中的Socket类对应。
NIO中的ServerSocketChannel监听通道,对应于OIO中的ServerSocket类。
ServerSocketChannel应用于服务器端,而SocketChannel同时处于服务器端和客户端。换句话说,对应于一个连接,两端都有一个负责传输的SocketChannel传输通道。
无论是ServerSocketChannel,还是SocketChannel,都支持阻塞和非阻塞两种模式。如何进行模式的设置呢?调用configureBlocking方法,具体如下:
(1)socketChannel.configureBlocking(false)设置为非阻塞模式。
(2)socketChannel.configureBlocking(true)设置为阻塞模式。
在阻塞模式下,SocketChannel通道的connect连接、read读、write写操作,都是同步的和阻塞式的,在效率上与Java旧的OIO的面向流的阻塞式读写操作相同。因此,在这里不介绍阻塞模式下的通道的具体操作。在非阻塞模式下,通道的操作是异步、高效率的,这也是相对于传统的OIO的优势所在。下面详细介绍在非阻塞模式下通道的打开、读写和关闭操作等操作。
1. 获取SocketChannel传输通道
在客户端,先通过SocketChannel静态方法open()获得一个套接字传输通道;然后,将socket套接字设置为非阻塞模式;最后,通过connect()实例方法,对服务器的IP和端口发起连接。
//获得一个套接字传输通道
SocketChannelsocketChannel = SocketChannel.open();
//设置为非阻塞模式
socketChannel.configureBlocking(false);
//对服务器的IP和端口发起连接
socketChannel.connect(new InetSocketAddress("127.0.0.1",80));
非阻塞情况下,与服务器的连接可能还没有真正建立,socketChannel.connect方法就返回了,因此需要不断地自旋,检查当前是否是连接到了主机:
while(! socketChannel.finishConnect() ){
//不断地自旋、等待,或者做一些其他的事情……
}
在服务器端,如何获取传输套接字呢?
当新连接事件到来时,在服务器端的ServerSocketChannel能成功地查询出一个新连接事件,并且通过调用服务器端ServerSocketChannel监听套接字的accept()方法,来获取新连接的套接字通道:
//新连接事件到来,首先通过事件,获取服务器监听通道
ServerSocketChannel server = (ServerSocketChannel) key.channel();
//获取新连接的套接字通道
SocketChannelsocketChannel = server.accept();
//设置为非阻塞模式
socketChannel.configureBlocking(false);
强调一下,NIO套接字通道,主要用于非阻塞应用场景。所以,需要调用configureBlocking(false),从阻塞模式设置为非阻塞模式。
2. 读取SocketChannel传输通道
当SocketChannel通道可读时,可以从SocketChannel读取数据,具体方法与前面的文件通道读取方法是相同的。调用read方法,将数据读入缓冲区ByteBuffer。
ByteBufferbuf = ByteBuffer.allocate(1024);
int bytesRead = socketChannel.read(buf);
在读取时,因为是异步的,因此我们必须检查read的返回值,以便判断当前是否读取到了数据。read()方法的返回值,是读取的字节数。如果返回-1,那么表示读取到对方的输出结束标志,对方已经输出结束,准备关闭连接。实际上,通过read方法读数据,本身是很简单的,比较困难的是,在非阻塞模式下,如何知道通道何时是可读的呢?这就需要用到NIO的新组件——Selector通道选择器,稍后介绍。
3. 写入到SocketChannel传输通道
和前面的把数据写入到FileChannel文件通道一样,大部分应用场景都会调用通道的int write(ByteBufferbuf)方法。
//写入前需要读取缓冲区,要求ByteBuffer是读取模式
buffer.flip();
socketChannel.write(buffer);
4. 关闭SocketChannel传输通道
在关闭SocketChannel传输通道前,如果传输通道用来写入数据,则建议调用一次shutdownOutput()终止输出方法,向对方发送一个输出的结束标志(-1)。然后调用socketChannel.close()方法,关闭套接字连接。
//终止输出方法,向对方发送一个输出的结束标志
socketChannel.shutdownOutput();
//关闭套接字连接
IOUtil.closeQuietly(socketChannel);
3.4.5 使用SocketChannel发送文件的实践案例
下面的实践案例是使用FileChannel文件通道读取本地文件内容,然后在客户端使用SocketChannel套接字通道,把文件信息和文件内容发送到服务器。客户端的完整代码如下:
package com.crazymakercircle.iodemo.socketDemos;
//...
public class NioSendClient {
private Charset charset = Charset.forName("UTF-8");
/**
* 向服务器端传输文件
*/
public void sendFile() throws Exception {
try {
String sourcePath = NioDemoConfig.SOCKET_SEND_FILE;
String srcPath = IOUtil.getResourcePath(sourcePath);
Logger.info("srcPath=" + srcPath);
String destFile = NioDemoConfig.SOCKET_RECEIVE_FILE;
Logger.info("destFile=" + destFile);
File file = new File(srcPath);
if (!file.exists()) {
Logger.info("文件不存在");
return;
}
FileChannelfileChannel = new FileInputStream(file).getChannel();
SocketChannelsocketChannel = SocketChannel.open();
socketChannel.socket().connect(
InetSocketAddress(NioDemoConfig.SOCKET_SERVER_IP,
NioDemoConfig.SOCKET_SERVER_PORT));
socketChannel.configureBlocking(false);
while(! socketChannel.finishConnect() ){
//不断地自旋、等待,或者做一些其他的事情
}
Logger.info("Client成功连接服务器端");
//发送文件名称
ByteBufferfileNameByteBuffer = charset.encode(destFile);
socketChannel.write(fileNameByteBuffer);
//发送文件长度
ByteBuffer buffer = ByteBuffer.allocate
(NioDemoConfig.SEND_BUFFER_SIZE);
buffer.putLong(file.length());
buffer.flip();
socketChannel.write(buffer);
buffer.clear();
//发送文件内容
Logger.info("开始传输文件");
int length = 0;
long progress = 0;
while ((length = fileChannel.read(buffer)) > 0) {
buffer.flip();
socketChannel.write(buffer);
buffer.clear();
progress += length;
Logger.info("| "+(100 * progress / file.length()) + "% |");
}
if (length == -1) {
IOUtil.closeQuietly(fileChannel);
//在SocketChannel传输通道关闭前,尽量发送一个输出结束标志到对端
socketChannel.shutdownOutput();
IOUtil.closeQuietly(socketChannel);
}
Logger.info("======== 文件传输成功 ========");
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
NioSendClient client = new NioSendClient(); // 启动客户端连接
client.sendFile(); // 传输文件
}
}
以上代码中的文件发送过程:首先发送目标文件名称(不带路径),然后发送文件长度,最后是发送文件内容。代码中的配置项,如服务器的IP、服务器端口、待发送的源文件名称(带路径)、远程的目标文件名称等配置信息,都是从system.properties配置文件中读取的,通过自定义的NioDemoConfig配置类来完成配置。
在运行以上客户端的程序之前,需要先运行服务器端的程序。服务器端的类与客户端的源代码在同一个包下,类名为NioReceiveServer,具体参见源代码工程,我们稍后再详细介绍这个类。
3.4.6 DatagramChannel数据报通道
和Socket套接字的TCP传输协议不同,UDP协议不是面向连接的协议。使用UDP协议时,只要知道服务器的IP和端口,就可以直接向对方发送数据。在Java中使用UDP协议传输数据,比TCP协议更加简单。在Java NIO中,使用DatagramChannel数据报通道来处理UDP协议的数据传输。
1. 获取DatagramChannel数据报通道
获取数据报通道的方式很简单,调用DatagramChannel类的open静态方法即可。然后调用configureBlocking(false)方法,设置成非阻塞模式。
//获取DatagramChannel数据报通道
DatagramChannel channel = DatagramChannel.open();
//设置为非阻塞模式
datagramChannel.configureBlocking(false);
如果需要接收数据,还需要调用bind方法绑定一个数据报的监听端口,具体如下:
//调用bind方法绑定一个数据报的监听端口
channel.socket().bind(new InetSocketAddress(18080));
2. 读取DatagramChannel数据报通道数据
当DatagramChannel通道可读时,可以从DatagramChannel读取数据。和前面的SocketChannel的读取方式不同,不是调用read方法,而是调用receive(ByteBufferbuf)方法将数据从DatagramChannel读入,再写入到ByteBuffer缓冲区中。
//创建缓冲区
ByteBufferbuf = ByteBuffer.allocate(1024);
//从DatagramChannel读入,再写入到ByteBuffer缓冲区
SocketAddressclientAddr= datagramChannel.receive(buffer);
通道读取receive(ByteBufferbuf)方法的返回值,是SocketAddress类型,表示返回发送端的连接地址(包括IP和端口)。通过receive方法读数据非常简单,但是,在非阻塞模式下,如何知道DatagramChannel通道何时是可读的呢?和SocketChannel一样,同样需要用到NIO的新组件—Selector通道选择器,稍后介绍。
3. 写入DatagramChannel数据报通道
向DatagramChannel发送数据,和向SocketChannel通道发送数据的方法也是不同的。这里不是调用write方法,而是调用send方法。示例代码如下:
//把缓冲区翻转到读取模式
buffer.flip();
//调用send方法,把数据发送到目标IP+端口
dChannel.send(buffer, new InetSocketAddress(NioDemoConfig.SOCKET_SERVER_IP,
NioDemoConfig.SOCKET_SERVER_PORT));
//清空缓冲区,切换到写入模式
buffer.clear();
由于UDP是面向非连接的协议,因此,在调用send方法发送数据的时候,需要指定接收方的地址(IP和端口)。
4. 关闭DatagramChannel数据报通道
这个比较简单,直接调用close()方法,即可关闭数据报通道。
//简单关闭即可
dChannel.close();
3.4.7 使用DatagramChannel数据包通道发送数据的实践案例
下面是一个使用DatagramChannel数据包通到发送数据的客户端示例程序代码。其功能是:获取用户的输入数据,通过DatagramChannel数据报通道,将数据发送到远程的服务器。客户端的完整程序代码如下:
package com.crazymakercircle.iodemo.udpDemos;
//...
public class UDPClient {
public void send() throws IOException {
//获取DatagramChannel数据报通道
DatagramChanneldChannel = DatagramChannel.open();
//设置为非阻塞
dChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(NioDemoConfig.SEND_BUFFER_
SIZE);
Scanner scanner = new Scanner(System.in);
Print.tcfo("UDP客户端启动成功!");
Print.tcfo("请输入发送内容:");
while (scanner.hasNext()) {
String next = scanner.next();
buffer.put((Dateutil.getNow() + " >>" + next).getBytes());
buffer.flip();
//通过DatagramChannel数据报通道发送数据
dChannel.send(buffer, new InetSocketAddress(NioDemoConfig.
SOCKET_SERVER_IP, NioDemoConfig.SOCKET_SERVER_PORT));
buffer.clear();
}
//操作四:关闭DatagramChannel数据报通道
dChannel.close();
}
public static void main(String[] args) throws IOException {
new UDPClient().send();
}
}
通过示例程序代码可以看出,在客户端使DatagramChannel数据报通道发送数据,比起在客户端使用套接字SocketChannel发送数据,简单很多。
接下来看看在服务器端应该如何使用DatagramChannel数据包通道接收数据呢?
下面贴出服务器端通过DatagramChannel数据包通道接收数据的程序代码,可能大家目前不一定可以看懂,因为代码中用到了Selector选择器,但是不要紧,下一个小节就介绍它。
服务器端的接收功能是:通过DatagramChannel数据报通道,绑定一个服务器地址(IP+端口),接收客户端发送过来的UDP数据报。服务器端的完整代码如下:
package com.crazymakercircle.iodemo.udpDemos;
//...
public class UDPServer {
public void receive() throws IOException {
//获取DatagramChannel数据报通道
DatagramChanneldatagramChannel = DatagramChannel.open();
//设置为非阻塞模式
datagramChannel.configureBlocking(false);
//绑定监听地址
datagramChannel.bind(new InetSocketAddress(NioDemoConfig.SOCKET
_SERVER_IP, NioDemoConfig.SOCKET_SERVER_PORT));
Print.tcfo("UDP服务器启动成功!");
//开启一个通道选择器
Selector selector = Selector.open();
//将通道注册到选择器
datagramChannel.register(selector, SelectionKey.OP_READ);
//通过选择器,查询IO事件
while (selector.select() > 0) {
Iterator<SelectionKey> iterator = selector.selectedKeys()
.iterator();
ByteBuffer buffer = ByteBuffer.allocate(NioDemoConfig.SEND
_BUFFER_SIZE);
//迭代IO事件
while (iterator.hasNext()) {
SelectionKeyselectionKey = iterator.next();
//可读事件,有数据到来
if (selectionKey.isReadable()) {
//读取DatagramChannel数据报通道的数据
SocketAddress client = datagramChannel.receive(buffer);
buffer.flip();
Print.tcfo(new String(buffer.array(), 0, buffer.limit()));
buffer.clear();
}
}
iterator.remove();
}
//关闭选择器和通道
selector.close();
datagramChannel.close();
}
public static void main(String[] args) throws IOException {
new UDPServer().receive();
}
}
在服务器端,首先调用了bind方法绑定datagramChannel的监听端口。当数据到来后,调用了receive方法,从datagramChannel数据包通道接收数据,再写入到ByteBuffer缓冲区中。
除此之外,在服务器端代码中,为了监控数据的到来,使用了Selector选择器。什么是选择器?如何使用选择器呢?欲知后事如何,请听下节分解。
3.5 详解NIO Selector选择器
Java NIO的三大核心组件:Channel(通道)、Buffer(缓冲区)、Selector(选择器)。其中通道和缓冲区,二者的联系也比较密切:数据总是从通道读到缓冲区内,或者从缓冲区写入到通道中。
至此,前面两个组件已经介绍完毕,下面迎来了最后一个非常重要的角色——选择器(Selector)。
3.5.1 选择器以及注册
选择器(Selector)是什么呢?选择器和通道的关系又是什么?
简单地说:选择器的使命是完成IO的多路复用。一个通道代表一条连接通路,通过选择器可以同时监控多个通道的IO(输入输出)状况。选择器和通道的关系,是监控和被监控的关系。
选择器提供了独特的API方法,能够选出(select)所监控的通道拥有哪些已经准备好的、就绪的IO操作事件。
一般来说,一个单线程处理一个选择器,一个选择器可以监控很多通道。通过选择器,一个单线程可以处理数百、数千、数万、甚至更多的通道。在极端情况下(数万个连接),只用一个线程就可以处理所有的通道,这样会大量地减少线程之间上下文切换的开销。
通道和选择器之间的关系,通过register(注册)的方式完成。调用通道的Channel.register(Selector sel,int ops)方法,可以将通道实例注册到一个选择器中。register方法有两个参数:第一个参数,指定通道注册到的选择器实例;第二个参数,指定选择器要监控的IO事件类型。
可供选择器监控的通道IO事件类型,包括以下四种:
(1)可读:SelectionKey.OP_READ
(2)可写:SelectionKey.OP_WRITE
(3)连接:SelectionKey.OP_CONNECT
(4)接收:SelectionKey.OP_ACCEPT
事件类型的定义在SelectionKey类中。如果选择器要监控通道的多种事件,可以用“按位或”运算符来实现。例如,同时监控可读和可写IO事件:
//监控通道的多种事件,用“按位或”运算符来实现
int key = SelectionKey.OP_READ | SelectionKey.OP_WRITE ;
什么是IO事件呢?这个概念容易混淆,这里特别说明一下。这里的IO事件不是对通道的IO操作,而是通道的某个IO操作的一种就绪状态,表示通道具备完成某个IO操作的条件。
比方说,某个SocketChannel通道,完成了和对端的握手连接,则处于“连接就绪”(OP_CONNECT)状态。
再比方说,某个ServerSocketChannel服务器通道,监听到一个新连接的到来,则处于“接收就绪”(OP_ACCEPT)状态。
还比方说,一个有数据可读的SocketChannel通道,处于“读就绪”(OP_READ)状态;一个等待写入数据的,处于“写就绪”(OP_WRITE)状态。
3.5.2 SelectableChannel可选择通道
并不是所有的通道,都是可以被选择器监控或选择的。比方说,FileChannel文件通道就不能被选择器复用。判断一个通道能否被选择器监控或选择,有一个前提:判断它是否继承了抽象类SelectableChannel(可选择通道)。如果继承了SelectableChannel,则可以被选择,否则不能。
简单地说,一条通道若能被选择,必须继承SelectableChannel类。
SelectableChannel类,是何方神圣呢?它提供了实现通道的可选择性所需要的公共方法。Java NIO中所有网络链接Socket套接字通道,都继承了SelectableChannel类,都是可选择的。而FileChannel文件通道,并没有继承SelectableChannel,因此不是可选择通道。
3.5.3 SelectionKey选择键
通道和选择器的监控关系注册成功后,就可以选择就绪事件。具体的选择工作,和调用选择器Selector的select()方法来完成。通过select方法,选择器可以不断地选择通道中所发生操作的就绪状态,返回注册过的感兴趣的那些IO事件。换句话说,一旦在通道中发生了某些IO事件(就绪状态达成),并且是在选择器中注册过的IO事件,就会被选择器选中,并放入SelectionKey选择键的集合中。
这里出现一个新的概念——SelectionKey选择键。SelectionKey选择键是什么呢?简单地说,SelectionKey选择键就是那些被选择器选中的IO事件。前面讲到,一个IO事件发生(就绪状态达成)后,如果之前在选择器中注册过,就会被选择器选中,并放入SelectionKey选择键集合中;如果之前没有注册过,即使发生了IO事件,也不会被选择器选中。SelectionKey选择键和IO的关系,可以简单地理解为:选择键,就是被选中了的IO事件。
在编程时,选择键的功能是很强大的。通过SelectionKey选择键,不仅仅可以获得通道的IO事件类型,比方说SelectionKey.OP_READ;还可以获得发生IO事件所在的通道;另外,也可以获得选出选择键的选择器实例。
3.5.4 选择器使用流程
使用选择器,主要有以下三步:
(1)获取选择器实例;(2)将通道注册到选择器中;(3)轮询感兴趣的IO就绪事件(选择键集合)。
第一步:获取选择器实例
选择器实例是通过调用静态工厂方法open()来获取的,具体如下:
//调用静态工厂方法open()来获取Selector实例
Selector selector = Selector.open();
Selector选择器的类方法open()的内部,是向选择器SPI(SelectorProvider)发出请求,通过默认的SelectorProvider(选择器提供者)对象,获取一个新的选择器实例。Java中SPI全称为(Service Provider Interface,服务提供者接口),是JDK的一种可以扩展的服务提供和发现机制。Java通过SPI的方式,提供选择器的默认实现版本。也就是说,其他的服务提供商可以通过SPI的方式,提供定制化版本的选择器的动态替换或者扩展。
第二步:将通道注册到选择器实例
要实现选择器管理通道,需要将通道注册到相应的选择器上,简单的示例代码如下:
// 2.获取通道
ServerSocketChannelserverSocketChannel = ServerSocketChannel.open();
// 3.设置为非阻塞
serverSocketChannel.configureBlocking(false);
// 4.绑定连接
serverSocketChannel.bind(new InetSocketAddress(SystemConfig.SOCKET_SERVER_PORT));
// 5.将通道注册到选择器上,并制定监听事件为:“接收连接”事件
serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);
上面通过调用通道的register()方法,将ServerSocketChannel通道注册到了一个选择器上。当然,在注册之前,首先要准备好通道。
这里需要注意:注册到选择器的通道,必须处于非阻塞模式下,否则将抛出IllegalBlockingModeException异常。这意味着,FileChannel文件通道不能与选择器一起使用,因为FileChannel文件通道只有阻塞模式,不能切换到非阻塞模式;而Socket套接字相关的所有通道都可以。
其次,还需要注意:一个通道,并不一定要支持所有的四种IO事件。例如服务器监听通道ServerSocketChannel,仅仅支持Accept(接收到新连接)IO事件;而SocketChannel传输通道,则不支持Accept(接收到新连接)IO事件。
如何判断通道支持哪些事件呢?可以在注册之前,可以通过通道的validOps()方法,来获取该通道所有支持的IO事件集合。
第三步:选出感兴趣的IO就绪事件(选择键集合)
通过Selector选择器的select()方法,选出已经注册的、已经就绪的IO事件,保存到SelectionKey选择键集合中。SelectionKey集合保存在选择器实例内部,是一个元素为SelectionKey类型的集合(Set)。调用选择器的selectedKeys()方法,可以取得选择键集合。
接下来,需要迭代集合的每一个选择键,根据具体IO事件类型,执行对应的业务操作。大致的处理流程如下:
//轮询,选择感兴趣的IO就绪事件(选择键集合)
while (selector.select() > 0) {
Set selectedKeys = selector.selectedKeys();
Iterator keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
//根据具体的IO事件类型,执行对应的业务操作
if(key.isAcceptable()) {
// IO事件:ServerSocketChannel服务器监听通道有新连接
} else if (key.isConnectable()) {
// IO事件:传输通道连接成功
} else if (key.isReadable()) {
// IO事件:传输通道可读
} else if (key.isWritable()) {
// IO事件:传输通道可写
}
//处理完成后,移除选择键
keyIterator.remove();
}
}
处理完成后,需要将选择键从这个SelectionKey集合中移除,防止下一次循环的时候,被重复的处理。SelectionKey集合不能添加元素,如果试图向SelectionKey选择键集合中添加元素,则将抛出java.lang.UnsupportedOperationException异常。
用于选择就绪的IO事件的select()方法,有多个重载的实现版本,具体如下:
(1)select():阻塞调用,一直到至少有一个通道发生了注册的IO事件。
(2)select(long timeout):和select()一样,但最长阻塞时间为timeout指定的毫秒数。
(3)selectNow():非阻塞,不管有没有IO事件,都会立刻返回。
select()方法返回的整数值(int整数类型),表示发生了IO事件的通道数量。更准确地说,是从上一次select到这一次select之间,有多少通道发生了IO事件。强调一下,select()方法返回的数量,指的是通道数,而不是IO事件数,准确地说,是指发生了选择器感兴趣的IO事件的通道数。
3.5.5 使用NIO实现Discard服务器的实践案例
Discard服务器的功能很简单:仅仅读取客户端通道的输入数据,读取完成后直接关闭客户端通道;并且读取到的数据直接抛弃掉(Discard)。Discard服务器足够简单明了,作为第一个学习NIO的通信实例,较有参考价值。
下面的Discard服务器代码,将选择器使用流程中的步骤进行了细化:
package com.crazymakercircle.iodemo.NioDiscard;
//...
public class NioDiscardServer {
public static void startServer() throws IOException {
// 1.获取选择器
Selector selector = Selector.open();
// 2.获取通道
ServerSocketChannelserverSocketChannel = ServerSocketChannel.open();
// 3.设置为非阻塞
serverSocketChannel.configureBlocking(false);
// 4.绑定连接
serverSocketChannel.bind(newInetSocketAddress(NioDemoConfig
.SOCKET_SERVER_PORT));
Logger.info("服务器启动成功");
// 5.将通道注册的“接收新连接”IO事件,注册到选择器上
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
// 6.轮询感兴趣的IO就绪事件(选择键集合)
while (selector.select() > 0) {
// 7.获取选择键集合
Iterator<SelectionKey>selectedKeys = selector.selectedKeys().
iterator();
while (selectedKeys.hasNext()) {
// 8.获取单个的选择键,并处理
SelectionKeyselectedKey = selectedKeys.next();
// 9.判断key是具体的什么事件
if (selectedKey.isAcceptable()) {
// 10.若选择键的IO事件是“连接就绪”事件,就获取客户端连接
SocketChannelsocketChannel = serverSocketChannel.accept();
// 11.切换为非阻塞模式
socketChannel.configureBlocking(false);
// 12.将该新连接的通道的可读事件,注册到选择器上
socketChannel.register(selector, SelectionKey.OP_READ);
} else if (selectedKey.isReadable()) {
// 13.若选择键的IO事件是“可读”事件, 读取数据
SocketChannelsocketChannel = (SocketChannel) selectedKey.
channel();
// 14.读取数据,然后丢弃
ByteBufferbyteBuffer = ByteBuffer.allocate(1024);
int length = 0;
while ((length = socketChannel.read(byteBuffer)) >0) {
byteBuffer.flip();
Logger.info(new String(byteBuffer.array(), 0, length));
byteBuffer.clear();
}
socketChannel.close();
}
// 15.移除选择键
selectedKeys.remove();
}
}
// 16.关闭连接
serverSocketChannel.close();
}
public static void main(String[] args) throws IOException {
startServer();
}
}
实现DiscardServer一共分为16步,其中第7到第15步是循环执行的。不断选择感兴趣的IO事件到选择器的选择键集合中,然后通过selector.selectedKeys()获取该选择键集合,并且进行迭代处理。对于新建立的socketChannel客户端传输通道,也要注册到同一个选择器上,使用同一个选择线程,不断地对所有的注册通道进行选择键的选择。
在DiscardServer程序中,涉及到两次选择器注册:一次是注册serverChannel服务器通道;另一次,注册接收到的socketChannel客户端传输通道。serverChannel服务器通道注册的,是新连接的IO事件SelectionKey.OP_ACCEPT;客户端socketChannel传输通道注册的,是可读IO事件SelectionKey.OP_READ。
DiscardServer在对选择键进行处理时,通过对类型进行判断,然后进行相应的处理
(1)如果是SelectionKey.OP_ACCEPT新连接事件类型,代表serverChannel服务器通道发生了新连接事件,则通过服务器通道的accept方法,获取新的socketChannel传输通道,并且将新通道注册到选择器。
(2)如果是SelectionKey.OP_READ可读事件类型,代表某个客户端通道有数据可读,则读取选择键中socketChannel传输通道的数据,然后丢弃。
客户端的DiscardClient代码,则更为简单。客户端首先建立到服务器的连接,发送一些简单的数据,然后直接关闭连接。代码如下:
package com.crazymakercircle.iodemo.NioDiscard;
//...
public class NioDiscardClient {
public static void startClient() throws IOException {
InetSocketAddress address =new InetSocketAddress(NioDemoConfig.
SOCKET_SERVER_IP, NioDemoConfig.SOCKET_SERVER_PORT);
// 1.获取通道
SocketChannelsocketChannel = SocketChannel.open(address);
// 2.切换成非阻塞模式
socketChannel.configureBlocking(false);
//不断地自旋、等待连接完成,或者做一些其他的事情
while (!socketChannel.finishConnect()) {
}
Logger.info("客户端连接成功");
// 3.分配指定大小的缓冲区
ByteBufferbyteBuffer = ByteBuffer.allocate(1024);
byteBuffer.put("hello world".getBytes());
byteBuffer.flip();
//发送到服务器
socketChannel.write(byteBuffer);
socketChannel.shutdownOutput();
socketChannel.close();
}
public static void main(String[] args) throws IOException {
startClient();
}
}
如果需要执行整个程序,首先要执行前面的服务器端程序,然后执行后面的客户端程序。
通过Discard服务器的开发实践,大家对NIO Selector(选择)的使用流程,应该了解得非常清楚了。
下面来看一个稍微复杂一点的案例:在服务器端接收文件和内容。
3.5.6 使用SocketChannel在服务器端接收文件的实践案例
本示例演示文件的接收,是服务器端的程序。和前面介绍的文件发送的SocketChannel客户端程序是相互配合使用的。由于在服务器端,需要用到选择器,所以在介绍完选择器后,才开始介绍NIO文件传输的Socket服务器端程序。服务器端接收文件的示例代码如下所示:
package com.crazymakercircle.iodemo.socketDemos;
//...
public class NioReceiveServer {
private Charset charset = Charset.forName("UTF-8");
/**
* 内部类,服务器端保存的客户端对象,对应一个客户端文件
*/
static class Client {
//文件名称
String fileName;
//长度
long fileLength;
//开始传输的时间
long startTime;
//客户端的地址
InetSocketAddressremoteAddress;
//输出的文件通道
FileChanneloutChannel;
}
private ByteBuffer buffer
= ByteBuffer.allocate(NioDemoConfig.SERVER_BUFFER_SIZE);
//使用Map保存每个文件传输,当OP_READ可读时,根据通道找到对应的对象
Map<SelectableChannel, Client>clientMap = new HashMap<SelectableChannel, Client>();
public void startServer() throws IOException {
// 1.获取选择器
Selector selector = Selector.open();
// 2.获取通道
ServerSocketChannelserverChannel = ServerSocketChannel.open();
ServerSocketserverSocket = serverChannel.socket();
// 3.设置为非阻塞
serverChannel.configureBlocking(false);
// 4.绑定连接
InetSocketAddress address
= new InetSocketAddress(NioDemoConfig.SOCKET_SERVER_PORT);
serverSocket.bind(address);
// 5.将通道注册到选择器上,并注册的IO事件为:“接收新连接”
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
Print.tcfo("serverChannel is listening...");
// 6.选择感兴趣的IO就绪事件(选择键集合)
while (selector.select() > 0) {
// 7.获取选择键集合
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
while (it.hasNext()) {
// 8.获取单个的选择键,并处理
SelectionKey key = it.next();
// 9.判断key是具体的什么事件,是否为新连接事件
if (key.isAcceptable()) {
// 10.若接受的事件是“新连接”事件,就获取客户端新连接
ServerSocketChannel server
= (ServerSocketChannel) key.channel();
SocketChannelsocketChannel = server.accept();
if (socketChannel == null) continue;
// 11.客户端新连接,切换为非阻塞模式
socketChannel.configureBlocking(false);
// 12.将客户端新连接通道注册到选择器上
SelectionKeyselectionKey =
socketChannel.register(selector, SelectionKey.OP_READ);
// 为每一条传输通道,建立一个Client客户端对象,放入map,供后面使用
Client client = new Client();
client.remoteAddress
= (InetSocketAddress) socketChannel.getRemoteAddress();
clientMap.put(socketChannel, client);
Logger.info(socketChannel.getRemoteAddress() + "连接成功...");
} else if (key.isReadable()) {
// 13.若接收的事件是“数据可读”事件,就读取客户端新连接
processData(key);
}
// NIO的特点只会累加,已选择的键的集合不会删除
// 如果不删除,下一次又会被select函数选中
it.remove();
}
}
}
/**
* 处理客户端传输过来的数据
*/
private void processData(SelectionKey key) throws IOException {
Client client = clientMap.get(key.channel());
SocketChannelsocketChannel = (SocketChannel) key.channel();
int num = 0;
try {
buffer.clear();
while ((num = socketChannel.read(buffer)) > 0) {
buffer.flip();
if (null == client.fileName) {
//客户端发送过来的,首先是文件名
//根据文件名,创建服务器端的文件,将文件通道保存到客户端
String fileName = charset.decode(buffer).toString();
String destPath = IOUtil.getResourcePath(
NioDemoConfig.SOCKET_RECEIVE_PATH);
File directory = new File(destPath);
if (!directory.exists()) {
directory.mkdir();
}
client.fileName = fileName;
String fullName = directory.getAbsolutePath()
+ File.separatorChar + fileName;
Logger.info("NIO 传输目标文件:" + fullName);
File file = new File(fullName);
FileChannelfileChannel
= new FileOutputStream(file).getChannel();
client.outChannel = fileChannel;
}else if (0 == client.fileLength) {
//客户端发送过来的,其次是文件长度
long fileLength = buffer.getLong();
client.fileLength = fileLength;
client.startTime = System.currentTimeMillis();
Logger.info("NIO 传输开始:");
} else {
//客户端发送过来的,最后是文件内容,写入文件内容
client.outChannel.write(buffer);
}
buffer.clear();
}
key.cancel();
} catch (IOException e) {
key.cancel();
e.printStackTrace();
return;
}
// 读取数量-1,表示客户端传输结束标志到了
if (num == -1) {
IOUtil.closeQuietly(client.outChannel);
System.out.println("上传完毕");
key.cancel();
Logger.info("文件接收成功,File Name:" + client.fileName);
Logger.info(" Size:" +
IOUtil.getFormatFileSize(client.fileLength));
long endTime = System.currentTimeMillis();
Logger.info("NIO IO传输毫秒数:" + (endTime - client.startTime));
}
}
public static void main(String[] args) throws Exception {
NioReceiveServer server = new NioReceiveServer();
server.startServer();
}
}
由于客户端每次传输文件,都会分为多次传输:
(1)首先传入文件名称。
(2)其次是文件大小。
(3)然后是文件内容。
对应于每一个客户端socketChannel,创建一个Client客户端对象,用于保存客户端状态,分别保存文件名、文件大小和写入的目标文件通道outChannel。
socketChannel和Client对象之间是一对一的对应关系:建立连接的时候,以socketChannel作为键(Key),Client对象作为值(Value),将Client保存在map中。当socketChannel传输通道有数据可读时,通过选择键key.channel()方法,取出IO事件所在socketChannel通道。然后通过socketChannel通道,从map中取到对应的Client对象。
接收到数据时,如果文件名为空,先处理文件名称,并把文件名保存到Client对象,同时创建服务器上的目标文件;接下来再读到数据,说明接收到了文件大小,把文件大小保存到Client对象;接下来再接到数据,说明是文件内容了,则写入Client对象的outChannel文件通道中,直到数据读取完毕。
运行方式:启动这个NioReceiveServer服务器程序后,再启动前面介绍的客户端程序NioSendClient,即可以完成文件的传输。
3.6 本章小结
在编程难度上,Java NIO编程的难度比同步阻塞Java OIO编程大很多。请注意,前面的实践案例,是比较简单的,并不是复杂的通信程序,没有看到“粘包”和“拆包”等问题。如果加上这些问题,代码将会更加复杂。
与Java OIO相比,Java NIO编程大致的特点如下:
(1)在NIO中,服务器接收新连接的工作,是异步进行的。不像Java的OIO那样,服务器监听连接,是同步的、阻塞的。NIO可以通过选择器(也可以说成:多路复用器),后续不断地轮询选择器的选择键集合,选择新到来的连接。
(2)在NIO中,SocketChannel传输通道的读写操作都是异步的。如果没有可读写的数据,负责IO通信的线程不会同步等待。这样,线程就可以处理其他连接的通道;不需要像OIO那样,线程一直阻塞,等待所负责的连接可用为止。
(3)在NIO中,一个选择器线程可以同时处理成千上万个客户端连接,性能不会随着客户端的增加而线性下降。
总之,有了Linux底层的epoll支持,有了Java NIO Selector选择器这样的应用层IO复用技术,Java程序从而可以实现IO通信的高TPS、高并发,使服务器具备并发数十万、数百万的连接能力。Java的NIO技术非常适合用于高性能、高负载的网络服务器。鼎鼎大名的通信服务器中间件Netty,就是基于Java的NIO技术实现的。
当然,Java NIO技术仅仅是基础,如果要实现通信的高性能和高并发,还离不开高效率的设计模式。下一章将开始为大家介绍高性能服务必备的设计模式:Reactor反应器模式。