本篇文章观点和例子来自 《Java网络编程精解》， 作者为孙卫琴， 出版社为电子工业出版社。
[转]http://blog.csdn.net/xhh198781/article/details/6635775
对于用ServerSocket 及 Socket 编写的服务器程序和客户程序, 他们在运行过程中常常会阻塞. 例如, 当一个线程执行 ServerSocket 的accept() 方法时, 假如没有客户连接, 该线程就会一直等到有客户连接才从 accept() 方法返回. 再例如, 当线程执行 Socket 的 read() 方法时, 如果输入流中没有数据, 该线程就会一直等到读入足够的数据才从 read() 方法返回.
假如服务器程序需要同时与多个客户通信, 就必须分配多个工作线程, 让他们分别负责与一个客户通信, 当然每个工作线程都有可能经常处于长时间的阻塞状态.
从 JDK1.4 版本开始, 引入了非阻塞的通信机制. 服务器程序接收客户连接, 客户程序建立与服务器的连接, 以及服务器程序和客户程序收发数据的操作都可以按非阻塞的方式进行. 服务器程序只需要创建一个线程, 就能完成同时与多个客户通信的任务.
非阻塞的通信机制主要由 java.nio 包(新I/O包) 中的类实现, 主要的类包括 ServerSocketChannel, SocketChannel, Selector, SelectionKey 和 ByteBuffer 等.
本章介绍如何用 java.nio 包中的类来创建服务器程序和客户程序, 并且 分别采用阻塞模式和非阻塞模式来实现它们. 通过比较不同的实现方式, 可以帮助读者理解它们的区别和适用范围.
一. 线程阻塞的概念
在生活中, 最常见的阻塞现象是公路上汽车的堵塞. 汽车在公路上快速行驶, 如果前方交通受阻, 就只好停下来等待, 等到交通畅顺, 才能恢复行驶.
线程在运行中也会因为某些原因而阻塞. 所有处于阻塞状态的线程的共同特征是: 放弃CPU, 暂停运行, 只有等到导致阻塞的原因消除, 才能恢复运行; 或者被其他线程中断, 该线程会退出阻塞状态, 并且抛出 InterruptedException.
1.1 线程阻塞的原因
导致线程阻塞的原因主要有以下几方面.
- 线程执行了 Thread.sleep(int n) 方法, 线程放弃 CPU, 睡眠 n 毫秒, 然后恢复运行.
- 线程要执行一段同步代码, 由于无法获得相关的同步锁, 只好进入阻塞状态, 等到获得了同步锁, 才能恢复运行.
- 线程执行了一个对象的 wait() 方法, 进入阻塞状态, 只有等到其他线程执行了该对象的 notify() 和 notifyAll() 方法, 才可能将其呼醒.
- 线程执行 I/O 操作或进行远程通信时, 会因为等待相关的资源而进入阻塞状态. 例如, 当线程执行 System.in.read() 方法时, 如果用户没有向控制台输入数据, 则该线程会一直等读到了用户的输入数据才从 read() 方法返回.
进行远程通信时, 在客户程序中, 线程在以下情况可能进入阻塞状态.
- 请求与服务器建立连接时, 即当线程执行 Socket 的带参数构造方法, 或执行 Socket 的 connect() 方法时, 会进入阻塞状态, 直到连接成功, 此线程才从 Socket 的构造方法或 connect() 方法返回.
- 线程从 Socket 的输入流读入数据时, 如果没有足够的数据, 就会进入阻塞状态, 直到读到了足够的数据, 或者到达输入流的末尾, 或者出现了异常, 才从输入流的 read() 方法返回或异常中断. 输入流中有多少数据才算足够呢? 这要看线程执行的 read() 方法的类型.
＞ int read(): 只要输入流中有一个字节, 就算足够.
＞ int read( byte[] buff): 只要输入流中的字节数目与参数buff 数组的长度相同, 就算足够.
＞ String readLine(): 只要输入流中有一行字符串, 就算足够. 值得注意的是, InputStream 类并没有 readLine() 方法, 在过滤流 BufferedReader 类中才有此方法.
- 线程向 Socket 的输出流写一批数据时, 可能会进入阻塞状态, 等到输出了所有的数据, 或者出现异常, 才从输出流 的 write() 方法返回或异常中断.
- 调用 SOcket 的setSoLinger() 方法设置了关闭 Socket 的延迟时间, 那么当线程执行 Socket 的 close() 方法时, 会进入阻塞状态, 直到底层 Socket 发送完所有剩余数据, 或者超过了 setSoLinger() 方法设置的延迟时间, 才从 close() 方法返回.
在服务器程序中, 线程在以下情况下可能会进入阻塞状态.
- 线程执行 ServerSocket 的 accept() 方法, 等待客户的连接, 直到接收到了客户连接, 才从 accept() 方法返回.
- 线程从 Socket 的输入流读入数据时, 如果输入流没有足够的数据, 就会进入阻塞状态.
- 线程向 Socket 的输出流写一批数据时, 可能会进入阻塞状态, 等到输出了所有的数据, 或者出现异常, 才从输出流的 write() 方法返回或异常中断.
由此可见, 无论在服务器程序还是客户程序中, 当通过 Socket 的输入流和输出流来读写数据时, 都可能进入阻塞状态. 这种可能出现阻塞的输入和输出操作被称为阻塞 I/O. 与此对照, 如果执行输入和输出操作时, 不会发生阻塞, 则称为非阻塞 I/O.
1.2 服务器程序用多线程处理阻塞通信的局限
本书第三章的第六节(创建多线程的服务器) 已经介绍了服务器程序用多线程来同时处理多个客户连接的方式. 服务器程序的处理流程如图 4-1 所示. 主线程负责接收客户的连接. 在线程池中有若干工作线程, 他们负责处理具体的客户连接. 每当主线程接收到一个客户连接, 就会把与这个客户交互的任务交给一个空闲的工作线程去完成, 主线程继续负责接收下一个客户连接.
图4-1 服务器程序用多线程处理阻塞通信
在图4-1 总, 用粗体框标识的步骤为可能引起阻塞的步骤. 从图中可以看出, 当主线程接收客户连接, 以及工作线程执行 I/O 操作时, 都有可能进入阻塞状态.
服务器程序用多线程来处理阻塞 I/O, 尽管能满足同时响应多个客户请求的需求, 但是有以下局限:
⑴ Java 虚拟机会为每个线程分配独立的堆栈空间, 工作线程数目越多, 系统开销就越大, 而且增加了 Java虚拟机调度线程的负担, 增加了线程之间同步的复杂性, 提高了线程死锁的可能性;
⑵ 工作线程的许多时间都浪费在阻塞 I/O 操作上, Java 虚拟机需要频繁地转让 CPU 的使用权, 使进入阻塞状态的线程放弃CPU, 再把CPU 分配给处于可运行状态的线程.
由此可见, 工作线程并不是越多越好. 如图 4-2 所示, 保持适量的工作线程, 会提高服务器的并发性能, 但是当工作线程的数目达到某个极限, 超出了系统的负荷时, 反而会减低并发性能, 使得多数客户无法快速得到服务器的响应.
图4-2 线程数目与并发性能的更新
1.3 非阻塞通信的基本思想
假如要同时做两件事: 烧开水和烧粥. 烧开水的步骤如下:
锅里放水, 打开煤气炉;
等待水烧开; //阻塞
关闭煤气炉, 把开水灌到水壶里;
烧粥的步骤如下:
锅里放水和米, 打开煤气炉;
等待粥烧开; //阻塞
调整煤气炉, 改为小火;
等待粥烧熟; //阻塞
关闭煤气炉;
为了同时完成两件事, 一个方案是同时请两个人分别做其中的一件事, 这相当于采用多线程来同时完成多个任务. 还有一种方案是让一个人同时完成两件事, 这个人应该善于利用一件事的空闲时间去做另一件事, 一刻也不应该闲着:
锅子里放水, 打开煤气炉; //开始烧水
锅子力放水和米, 打开煤气炉; //开始烧粥
while(一直等待, 直到有水烧开, 粥烧开或粥烧熟事件发生){ //阻塞
if(水烧开)
关闭煤气炉, 把开水灌到水壶里;
if(粥烧开)
调整煤气炉, 改为小火;
if(粥烧熟)
关闭煤气炉;
if(水已经烧开并且粥已经烧熟)
退出循环;
} //这里的煤气炉我可以理解为每件事就有一个煤气炉配给吧, 这也是一部分的开销呢
//并且if里面的动作必须要能快速完成的才行, 不然后面的就要排队了
//如是太累的工作还是不要用这个好
这个人不断监控烧水及烧粥的状态, 如果发生了 "水烧开", "粥烧开" 或 "粥烧熟" 事件, 就去处理这些事件, 处理完一件事后进行监控烧水及烧粥的状态, 直到所有的任务都完成.
以上工作方式也可以运用到服务器程序中, 服务器程序只需要一个线程就能同时负责接收客户的连接, 接收各个客户发送的数据, 以及向各个客户发送响应数据. 服务器程序的处理流程如下:
while(一直等待, 直到有接收连接就绪事件, 读就绪事件或写就绪事件发生){ //阻塞
if(有客户连接)
接收客户的连接; //非阻塞
if(某个 Socket 的输入流中有可读数据)
从输入流中读数据; //非阻塞
if(某个 Socket 的输出流可以写数据)
向输出流写数据; //非阻塞
}
以上处理流程采用了轮询的工作方式, 当某一种操作就绪时, 就执行该操作, 否则就查看是否还有其他就绪的操作可以执行. 线程不会因为某一个操作还没有就绪, 就进入阻塞状态, 一直傻傻地在那里等待这个操作就绪.
为了使轮询的工作方式顺利进行, 接收客户的连接, 从输入流读数据, 以及向输出流写数据的操作都应该以非阻塞的方式运行. 所谓非阻塞, 就是指当线程执行这些方法时, 如果操作还没有就绪, 就立即返回, 而不会一直等到操作就绪. 例如, 当线程接收客户连接时, 如果没有客户连接, 就立即返回; 再例如, 当线程从输入流中读数据时, 如果输入流中还没有数据, 就立即返回, 或者如果输入流还没有足够的数据, 那么就读取现有的数据, 然后返回. 值得注意的是, 以上 while 学校条件中的操作还是按照阻塞方式进行的, 如果未发生任何事件, 就会进入阻塞状态, 直到接收连接就绪事件, 读就绪事件或写就绪事件中至少有一个事件发生时, 才会执行 while 循环体中的操作. 在while 循环体中, 一般会包含在特定条件下退出循环的操作.
二. java.nio 包中的主要类
java.nio 包提供了支持非阻塞通信的类.
- ServerSocketChannel: ServerSocket 的替代类, 支持阻塞通信与非阻塞通信.
- SocketChannel: Socket 的替代类, 支持阻塞通信与非阻塞通信.
- Selector: 为ServerSocketChannel 监控接收连接就绪事件, 为 SocketChannel 监控连接就绪, 读就绪和写就绪事件.
- SelectionKey: 代表 ServerSocketChannel 及 SocketChannel 向 Selector 注册事件的句柄. 当一个 SelectionKey 对象位于Selector 对象的 selected-keys 集合中时, 就表示与这个 SelectionKey 对象相关的事件发生了.
ServerSocketChannel 及 SocketChannel 都是 SelectableChannel 的子类, 如图 4-3 所示. SelectableChannel 类及其子类都能委托 Selector 来监控他们可能发生的一些事件, 这种委托过程也称为注册事件过程.
图4-3 SelectableChannel 类及其子类的类框图
ServerSocketChannel 向 Selector 注册接收连接就绪事件的代码如下:
SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
SelectionKey 类的一些静态常量表示事件类型, ServerSocketChannel 只可能发生一种事件.
- SelectionKey.OP_ACCEPT: 接收连接就绪事件, 表示至少有了一个客户连接, 服务器可以接收这个连接.
SocketChannel 可能发生以下 3 种事件.
- SelectionKey.OP_CONNECT: 连接就绪事件, 表示客户与服务器的连接已经建立成功.
- SelectionKey.OP_READ: 读就绪事件, 表示输入流中已经有了可读数据, 可以执行读操作了
- SelectionKey.OP_WRITE: 写就绪事件, 表示已经可以向输入流写数据了.
SocketChannel 提供了接收和发送数据的方法.
- read(ByteBuffer buffer): 接收数据, 把它们存放到参数指定的 ByteBuffer 中.
- write(ByteBuffer buffer): 把参数指定的 ByteBuffer 中的数据发送出去.
ByteBuffer 表示字节缓冲区, SocketChannel 的 read() 和 write() 方法都会操纵 ByteBuffer. ByteBuffer 类继承于 Buffer 类. ByteBuffer 中存放的是字节, 为了把它们转换为字符串, 还需要用到 Charset 类, Charset 类代表字符编码, 它提供了把字节流转换为字符串(解码过程) 和把字符串转换为字节流(编码过程) 的实用方法.
下面几小节分别介绍 Buffer, Charset, Channel, SelectableChannel, ServerSocketChannel, SocketChannel, Selector 和 SelectionKey 的用法.
2.1 缓冲区 Buffer
数据输入和输出往往是比较耗时的操作. 缓冲区从两个方面提高 I/O 操作的效率:
- 减少实际的物理读写次数;
- 缓存区在创建时被分配内存, 这块内存区域一直被重用, 这可以减少动态分配和回收内存区域的次数.
旧I/O 类库(对应 java.nio包) 中的 BufferedInputStream, BufferedOutputStream, BufferedReader 和 BufferedWriter 在其实现中都运用了缓冲区. java.nio 包公开了 Buffer API, 使得Java 程序可以直接控制和运用缓冲区. 如图 4-4 所示, 显示了 Buffer 类的层次结构.
图 4-4 Buffer 类的层次结构
所有的缓冲区都有以下属性:
- 容量(capacity): 表示该缓冲区可以保存多少数据．
- 极限(limit): 表示缓冲区的当前终点, 不能对缓冲区中超过极限的区域进行读写操作. 极限是可以修改的, 这有利于缓冲区的重用. 例如, 假定容量100 的缓冲区已经填满了数据, 接着程序在重用缓冲区时, 仅仅将 10 个新的数据写入缓冲区中从位置0 到10 的区域, 这时可以将极限设为 10, 这样就不能读取先前的数据了. 极限是一个非负整数, 不应该大于容量.
- 位置(position): 表示缓冲区中下一个读写单元的位置, 每次读写缓冲区的数据时, 都会改变该值, 为下一次读写数据作准备. 位置是一个非负整数, 不应该大于极限.
如图 4-5 所示, 以上 3 个属性的关系为: 容量 ≥ 极限 ≥ 位置 ≥ 0
图 4-5 缓冲区的 3 个属性
缓冲区提供了用于改变以上 3 个属性的方法.
- clear(): 把极限设为容量, 再把位置设为 0;
- flip(): 把极限设为位置, 再把位置设为 0;
- rewind(): 不改变极限, 把位置设为 0.
Buffer 类的remaining() 方法返回缓冲区的剩余容量, 取值等于极限-位置. Buffer 类的 compact() 方法删除缓冲区内从 0 到当前位置position 的内容, 然后把从当前位置position 到极限limit 的内容复制到 0 到 limit-position 的区域内, 当前位置position 和极限limit 的取值也作相应的变化, 如图 4-6 所示.
图4-6 Buffer 类的 compact() 的作用
java.nio.Buffer 类是一个抽象类, 不能被实例化. 共有 8 个具体的缓冲区类, 其中最基本的缓冲区是 ByteBuffer, 它存放的数据单元是字节. ByteBuffer 类并没有提供公开的构造方法, 但是提供了两个获得 ByteBuffer 实例的静态工厂方法.
- allocate(int capacity): 返回一个 ByteBuffer 对象, 参数capacity 指定缓冲区的容量.
- directAllocate(int capacity):返回一个 ByteBuffer 对象, 参数capacity 指定缓冲区的容量. 该方法返回的缓冲区称为直接缓冲区, 它与当前操作系统能够更好地耦合, 因此能进一步提高 I/O 操作的速度. 但是直接分配缓冲区的系统开销很大, 因此只有在缓冲区较大并且长期存在, 或者需要经常重用时, 才使用这种缓冲区.
除 boolean 类型以外, 每种基本类型都有对应的缓冲区类, 包括 CharBuffer, DoubleBuffer, FloatBuffer, IntBuffer, LongBuffer 和 ShortBuffer. 这几个缓冲区类都有一个能够返回自身实例的静态工厂方法allocate(int capacity). 在 CharBuffer 中存放的数据单元为字符, 在 DoubleBuffer 中存放的数据单元为 double 数据, 依此类推. 还有一个缓冲区是 MappedByteBuffer, 它是 ByteBuffer 的子类. MappedByteBuffer 能够把缓冲区和文件的某个区域直接映射.
所有具有缓冲区类都提供了读写缓冲区的方法:
- get(): 相对读. 从缓冲区的当前位置读取一个单元的数据, 读完后把位置加 1;
- get(int index): 绝对读. 从参数 index 指定的位置读取一个单元的数据;
- put(): 相对写. 向缓冲区的当前位置写入一个单元的数据, 写完后把位置加 1;
- put(int index): 绝对写. 向参数 index 指定的位置写入一个单元的数据.
2.2 字符编码 Charset
java.nio.Channel 类的每个实例代表特定的字符编码类型. 如图 4-7 所示, 把字节序列转换为字符串的过程称为解码; 把字符串转换为字节序列的过程称为编码.
图 4-7 编码与解码
Charset 类提供了编码与解码的方法:
- ByteBuffer encode(String str): 对参数 Str 指定的字符串进行编码, 把得到的字节序列存放在一个 ByteBuffer 对象中, 并将其返回;
- ByteBuffer encode(CharBuffer cb): 对参数 cb 指定的字符缓冲区中的字符进行编码,把得到的字节序列存放在一个 ByteBuffer 对象中, 并将其返回;
- CharBuffer decode(ByteBuffer bb): 把参数 bb 指定的 ByteBuffer 中的字节序列进行解码, 把得到的字符序列存放在一个 CharBuffer 对象中, 并将其返回.
Charset 类的静态 forName(String encode) 方法返回一个 Charset 对象, 它代表参数 encode 指定的编码类型. 例如, 以下代码创建了一个代表"GBK" 编码的 Charset对象:
Charset charset = Charset.forName("GBK");
Charset 类还有一个静态方法 defaultCharset(), 它返回代表本地平台的默认字符编码的 Charset 对象.
2.3 通道Channel
通道 Channel 用来连接缓冲区与数据源或数据汇(数据目的地). 如图4-8 所示, 数据源的数据经过管道到达缓冲区, 缓冲区的数据经过通道到达数据汇.
图4-8 通道的作用
如图 4-9 所示, 显示了 Channel 的主要层次结构.
图4-9 Channel 的主要层次结构
java.nio.channels.Channel 接口只声明了两个方法.
- close(): 关闭通道;
- isOpen(): 判断通道是否打开.
通道在创建时被打开, 一旦关闭通道, 就不能重新打开了.
Channel 接口的两个最重要的子接口是 ReadableByteChannel 和 WritableByteChannel. ReadableByteChannel 接口声明了 read(ByteBuffer dst) 方法, 该方法把数据源的数据读入参数指定的 ByteBuffer 缓冲区中; WritableByteChannel 接口声明了 write(ByteBuffer src)方法, 该方法把参数指定的 ByteBuffer 缓冲区中的数据写到数据汇中. 如图4-10 所示, 显示了 Channel 与 Buffer 的关系. ByteChannel 接口是一个便利接口, 它扩展了 ReadByteChannel 和 WritableByteChannel 接口, 因而同时支持读写操作.
图4-10 Channel 与 Buffer 的关系
ScatteringByteChannel 接口扩展了 ReadByteChannel 接口, 允许分散地读取数据. 分散读取数据是指单个读取操作能填充多个缓冲区. ScatteringByteChannel 接口声明了 read(ByteBuffer[] dsts)方法, 该方法把从数据源读取的数据依次填充到参数指定的 ByteBuffer 数组的各个 ByteBuffer 中. GatheringByteChannel 接口扩展了 WritableByteChannel 接口, 允许集中地写入数据. 集中写入数据是指单个写操作能把多个缓冲区的数据写入数据汇. GatheringByteChannel 接口声明了 write(ByteBuffer[] srcs)方法, 该方法依次把参数指定的 ByteBuffer 数组的每个 ByteBuffer 中的数据写入数据汇. 分散读取和集中写数据能够进一步提高输入和输出操作的速度.
FileChannel 类是 Channel 接口的实现类, 代表一个与文件相连的通道. 该类实现了 ByteChannel, ScatteringByteChannel, GatheringByteChannel 接口, 支持读操作, 写操作, 分散读操作和集中写操作. FileChannel 类没有提供公开的构造方法, 一次客户程序不能用 new 语句来构造它的实现. 不过, 在 FileInputStream, FileOutputStream 和 RandomAccessFile 类中提供了 getChannel() 方法, 该方法返回对应的 FileChannel 对象.
SelectableChannel 也是一种通道, 它不仅支持阻塞的 I/O 操作, 还支持非阻塞的 I/O 操作. SelectableChannel 有两个子类: ServerSocketChannel 和 SocketChannel. SocketChannel 还实现了 ByteChannel 接口, 具有 read(ByteBuffer dst) 和 write(ByteBuffer src) 方法.
注意上面的图4-9 Channel 的主要层次结构, 这个跟原书有点区别, 里面的类都是 jdk1.5的, 其中 SocketChannel 是实现了 ByteChannel, ScatteringByteChannel, GatheringByteChannel 接口, SocketChannel 还有一个子类SocketChannelImpl, SocketChannelImpl 的源代码看不到呢.
2.4 SelectableChannel 类
SelectableChannel 类是一种支持阻塞 I/O 和非阻塞 I/O 的通道. 在非阻塞模式下, 读写数据不会阻塞, 并且SelectableChannel 可以向 Selector 注册读就绪和写就绪等事件. Selector 负责监控这些事件, 等到事件发生时, 比如发生了读就绪事件, SelectableChannel 就可以执行读操作了.
SelectableChannel 的主要方法如下:
- public SelecotableChannel configureBlocking(boolean block) throws IOException
数block 为true 时, 表示把 SelectableChannel 设为阻塞模式; 如果参数block 为false, 表示把 SelectableChannel 设为非阻塞模式. 默认情况下, SelectableChannel 采用阻塞模式. 该方法返回 SelectableChannel 对象本身的引用, 相当于" return this".
- public SelectionKey register(Selector sel, int ops) throws ClosedChannelException
- public SelectionKey register(Selector sel, int ops, Object attachment) throws ClosedChannelException
后两个方法都向 Selector 注册时间, 如以下 socketChannel( SelectableChannel 的一个子类) 向 Selector 注册读就绪和写就绪事件:
SelectionKey key = socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
register() 方法返回一个 SelectionKey 对象, SelectionKey 用来跟踪被注册的事件. 第二个register() 方法还有一个Object 类型的参数 attachment, 它用于为 SelectionKey 关联一个附件, 当被注册事件发生后, 需要处理该事件时, 可以从 SelectionKey 中获得这个附件, 该附件可用来包含与处理这个事件相关的信息. 以下这两段代码是等价的:
MyHandler handler = new MyHandler(); //负责处理事件的对象
SelectionKey key = socketChannel.register(selector, SelectioinKey.OP_READ | SelectionKey.OP_WRITE, handler );
等价于:
MyHandler handler = new MyHandler(); //负责处理事件的对象
SelectionKey key = socketChannel.register(selector, SelectioinKey.OP_READ | SelectionKey.OP_WRITE);
key.attach(handler ); //为SelectionKey 关联一个附件
2.5 ServerSocketChannel 类
SeverSocketChannel 从 SeletableChannel 中继承了 configureBlocking() 和 register()方法. ServerSocketChannel 是 ServerSocket 的替换类, 也具有负责接收客户连接的 accept() 方法. ServerSocket 并没有 public 类型的构造方法, 必须通过它的静态方法open() 来创建 ServerSocketChannel 对象. 每个ServerSocketChannel 对象都与一个ServerSocket 对象关联. ServerSocketChannel 的 socket() 方法返回与它关联的 ServerSocket 对象. 可通过以下方法把服务器进程绑定到一个本地端口:
serverSocketChannel.socket().bind(port);