Java IO
1.linux网络IO模型
Linux的内核将所有的外部设备都看作一个文件来操作,对一个文件的读写操作会调用内核提供的系统命令,返回一个fd(file descriptor,文件描述符),对socket的操作也会有相应的描述符,成为socketfd(socket描述符),它指向内存中的一个结构体(文件路径,数据区等一些属性)
1.1 阻塞I/O模型
应用进程调用recvfrom后,系统调用直到数据包到达并被复制到应用缓存中或发生错误才返回,期间应用进程一直被阻塞.
1.2 非阻塞I/O模型
调用recvfrom后,若系统缓冲区没有数据,则直接返回一个ewouldblock错误,之后轮询检查该状态,若数据到达则复制到应用缓存中后返回.
1.3 I/O复用模型
linux提供select/poll,应用程序将一个或多个fd传递给select/poll系统调用,然后阻塞在select上,这样可以同时侦测多个fd是否处于就绪状态,select/poll顺序扫描fd是否就绪,且支持到fd数量有限.此外Linux提供了epoll基于事件模式的驱动,当有fd就绪时,就立即调用回掉函数rollback.数据拷贝到应用缓存区期间,应用进程被阻塞.
1.4 信号驱动I/O模型
首先开启信号驱动I/O功能,并通过系统调用sigaction执行一个信号处理函数(立即返回,非阻塞),数据准备就绪后,为该进程生成一个sigio信号,并通知应用程序调用recvfrom读取数据(拷贝数据到应用缓存期间阻塞),拷贝完成返回.
1.5 异步I/O
告知内核启动某个操作,并在操作完成后通知我们(包括将数据从内核复制到应用缓冲区).与信号驱动模式到区别是:信号I/O由内核告诉我们何时开始一个I/O操作;异步I/O由内核告诉我们何时完成了I/O操作.
1.6 epoll
epoll改进的select缺点如下:
- 支持一个进程打开的socket描述符(FD)不受限制(仅受限于操作系统的最大文件句柄数)
- select最大缺陷是单个进程打开的FD是有限制的,默认值1024个,太多网络效率会下降
- epoll在1G内存大概支持10万个
- IO效率不会随着FD数目的增加而线性下降
- select/poll每次调用都会线性扫描全部的集合,导致效率呈现线性下降.
- epoll只有活跃的socket才会主动调用callback函数,epoll实现了一个伪AIO.极端情况当全部socket都活跃时,epoll和select性能差不多.
- 使用mmap加速内核与用户空间的消息传递.
- 无论select,pool还是epoll都需要内核把FD消息通知给用户空间
- epoll通过内核和用户空间mmap同一块内存来避免不必要的内存复制
- epoll的API更加简单.
- 包括创建一个epoll描述符、添加监听事件、阻塞等待所监听的事件发生、关闭epoll描述符等
2.NIO入门
socket传输网络数据的过程
I/O 对比 | 对比项 | 同步阻塞(BIO) | 伪异步I/O | 非阻塞(NIO) | 异步(AIO) | | -------- | -------- | -------- | -------- | -------- | | 客户端个数(I/O)线程 | 1:1 | S:C,S可以大于C | S:1,一个IO线程可以处理多个客户端连接 | M:0 无需启动额外的线程,被动回调 | | I/O类型(阻塞) | 阻塞 | 阻塞 | 非阻塞 | 非阻塞 | | I/O类型(同步) | 同步 | 同步 | 同步 | 异步 | | API使用难度 | 简单 | 较简单 | 非常复杂 | 复杂且非常繁琐 | | 调试难度 | 简单 | 简单 | 复杂 | 复杂 | | 可靠性 | 非常差 | 差 | 高 | 高 | | 吞吐量 | 非常低 | 中 | 高 | 高 |
2.1 BIO编程
ServerSocket: 负责绑定IP地址,启动监听端口;Socket负责发起连接操作.经过三次握手,连接成功之后,Client和Server双方通过输入流和输出流进行同步阻塞式通信.- 同步阻塞IO服务端通信模型:称做一个客户端连接一个线程.通常由一个独立的Acceptor线程负责监听客户端的连接,它接收到连接请求后为每个客户端创建一个新的线程进行链路处理,处理完成之后,通过输出流返回应答给客户端,线程销毁.
- 该模型最大问题:缺乏弹性伸缩能力,服务端线程数和客户端访问数呈1:1的正比关系,当线程数膨胀,性能急剧下降,导致宕机和僵死.
服务端通信模型
2.2 伪异步IO编程
是对传统BIO一个连接一个线程的简单优化,服务端通过一个线程池来处理多个客户端的请求接入,形成客户端个数M:线程池最大线程数N的比例关系,M可以远远大于N.由于底层依然使用同步阻塞IO,所以被称为"伪异步".
- 优点:资源占用是可控的,不会导致耗尽和宕机.
- 缺点:只是简单优化,无法从根本上解决同步IO导致的通信线程阻塞问题.
服务端通信模型
通信对方防护应答时间过长会引起的级联故障
- 消息接收方接收缓慢,导致发送TCP的发送窗口不断减少,直到为0,那么此时write就会被无限期阻塞
- 返回应答过长,所有的可用线程都被故障服务器阻塞,后续所有的IO消息都将在队列中排队
- 由于线程池采用阻塞队列实现,当队列积满之后,后续入队列的操作将会被阻塞
- 前端只有一个Accptor线程接手客户端接入,它被阻塞在线程池的同步阻塞队列中,新的客户端请求消息将被拒绝,客户端会发生大佬的连接超时
- 由于几乎所有的连接都超时,调用者会认为系统已经崩溃,无法接收新的请求消息
2.3 NIO编程
官方叫法是New I/O.而被大多数人接受的更准确叫法是非阻塞IO(Non-block I/O).
2.3.1 NIO类库简介
- 缓冲区
Buffer- 缓冲区实质是一个数组,提供对数据的结构化访问以及维护读写位置等.所有数据都是用缓冲区处理,任何时候访问NIO中的数据,都是通过缓冲区进行操作.
- 常见缓冲区类有7个:
- ByteBuffer(最常用):字节缓冲区
- CharBuffer:字符缓冲区
- ShortBuffer:短整型缓冲区
- IntBuffer:整型缓冲区
- LongBuffer:长整型缓冲区
- FloatBuffer:浮点型缓冲区
- DoubleBuffer:双精度浮点型缓冲区
java bio是面向流操作的.
- 通道
Channel- Channel是一个全双工的双向通道,可以读写操作同时进行,能更好的映射底层操作系统的API,因为Unix底层操作系统通道都是全双工的.而java bio 流是单向的,一个流必须是InputStream或OutputString的子类
- Channel可以分为两大类:用于网络多写的
SelectableChannel和用于文件操作的FileChannel ServerSocketChannel和SocketChannel都是SelectableChannel的子类
- 多路复用器
Selector- 多路复用器提供选择以及就绪的任务的能力.Selector会不断地轮询注册在其上地Channel,如果某个Channel上面发生读或写事件,这个Channel就处于就绪状态,会被Selector轮询出来,然后通过SelectionKey可以获取就绪Channel的集合,进行后续的IO操作.
- JDK使用epoll实现.
2.3.2 NIO编程优缺点
- 缺点:NIO编程难度比BIO大很大,编码复杂.
- 优点
- 客户端发起的连接操作都是异步的,通过在多路复用器注册OP_CONNECT等待后续结果.
- SocketChannel的读写操作都是异步的.
- 线程模型的优化,一个Selector线程可以同时处理成千上万个客户端连接.
2.3.3 NIO服务端
@startuml
autonumber
NioServer -> NioServer: 打开ServerSocketChannel
NioServer -> NioServer: 绑定监听地址 InetSocketAddress
ReactorThread -> ReactorThread: 创建Selector,启动线程
NioServer -> ReactorThread: 将ServerSocketChannel注册到Selector上\n,监听SelectionKey.OP_ACCEPT
ReactorThread -> ReactorThread: Selector轮询就绪的Key
ReactorThread -> IoHandler: handleAccept()处理新的客户端接入
IoHandler -> IoHandler: 设置新建连接的Socket
IoHandler -> ReactorThread: 向Selector注册监听读操作SelectionKey.OP_READ
ReactorThread -> IoHandler: handleRead()异步读请求消息到ByteBuffer
IoHandler -> IoHandler: decode请求
IoHandler -> ReactorThread: 异步写ByteBuffer到SocketChannel
@enduml
服务端序列图
源码分析
// 1. 打开ServerSocketChannel,用于监听客户端的连接
ServerSocketChannel open = ServerSocketChannel.open();
// 2. 绑定监听端口,设置连接为非阻塞模式
open.accept().bind(new InetSocketAddress(InetAddress.getByName("IP"), port));
open.configureBlocking(false);
// 3. 创建reactor线程,创建多路复用器并启动线程
Selector selector = Selector.open();
new Thread(new ReactorTask()).start();
// 4. 将ServerSocketChannel注册到Reactor线程的多路复用器Selector上,监听ACCEPT事件
SelectionKey key = open.register(selector, SelectionKey.OP_ACCEPT, ioHandler);
// 5. 多路复用器在线程run方法的无限循环体内轮询准备就绪的key
int num = selector.select();
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectionKeys.iterator();
while (it.hasNext()){
SelectionKey next = it.next();
}
// 6. 多路复用监听到有新的客户端接入.处理新的接入请求,完成TCP三次握手,建立物理链路
SocketChannel channel = open.accept();
// 7. 设置客户端链路为非阻塞模式
channel.configureBlocking(false);
channel.socket().setReuseAddress(true);
// 8. 将新接入的客户端来你就饿注册到Reactor线程的多路复用器上,监听读操作,读取客户端发送的网络消息
SelectionKey key = socketChannel.register(selector, SelectionKey.OP_READ, ioHandler);
// 9. 异步读取客户端请求消息到缓冲区
int readNumber = channel.read(receivedBuffer);
// 10. 对ByteBuffer进行编解码,如果有半包消息指针reset,继续读取后续的报文,将解码成功的消息封装成Task,投递到业务线程池中,进行业务逻辑编排
Object message = null;
while (buffer.hasRemain()){
byteBuffer.mark();
Object message = decode(byteBuffer);
if (message == null){
ByteBuffer.reset();
break;
}
messageList.add(message);
}
if (!byteBuffer.hasRemain()){
byteBuffer.clear();
}else {
byteBuffer.compact();
}
if (messageList != null && !messageList.isEmpty()){
if (Object messageE : messageList){
handlerTask(messageE);
}
}
// 11. 将POJO对象encode成ByteBuffer,调用SocketChannel的异步write接口,将消息异步发送给客户端
socketChannel.write(buffer);
// 注意: 如果发送区TCP缓冲区满,会导致写半包,此时,需要注册监听写操作位,循环写,直到整包消息写入TCP缓冲区
2.3.3 NIO客户端
@startuml
autonumber
NioClient -> NioClient: 打开SocketChannel
NioClient -> NioClient: 设置SocketChannel为非阻塞模式,同时设置TCP参数
NioClient -> NioServer: 异步连接服务端
NioClient -> NioClient: 判断连接结果,如果连接成功,调到10,否知执行5
NioClient -> ReactorThread: 向Reactor线程的多路复用器注册OP CONNECT 事件
ReactorThread -> ReactorThread: 创建Selector,启动线程
ReactorThread -> ReactorThread: Selector轮询就绪的Key
ReactorThread -> IoHandler: handleConnect()
IoHandler -> IoHandler: 判断连接完成
IoHandler -> ReactorThread: 向Selector注册监听读操作SelectionKey.OP_READ
ReactorThread -> IoHandler: handleRead()异步读请求消息到ByteBuffer
IoHandler -> IoHandler: decode请求
IoHandler -> ReactorThread: 异步写ByteBuffer到SocketChannel
@enduml
客户端序列图
源码分析
// 1.打开SocketChannel,绑定客户端本地地址(可选,默认系统会随机分配一个可用的本地地址)
SocketChannel clientChannel = SocketChannel.open();
// 2.设置SocketChannel为非阻塞模式,同时设置客户端连接的TCP参数
clientChannel.configureBlocking(false);
socket.setReuseAddress(true);
socket.setReceiveBufferSize(BUFFER_SIZE);
socket.setSendBufferSize(BUFFER_SIZE);
// 3.异步连接服务端
boolean connected = clientChannel.connect(new InetSocketAddress("ip",port));
// 4.判断是否连接成功,如果连接成功,则直接注册读状态位到多路复用器中,如果当前没有连接成功(异步连接,返回false,说明客户端已经发送sync包,服务端没有返回ack包,物理链路还没有建立)
if (connected) {
clientChannel.register(selector, SelectionKey.OP_READ, ioHandler);
} else {
clientChannel.register(selector, SelectionKey.OP_CONNECT, ioHandler);
}
// 5.向Reactor线程的多路复用器注册OP_CONNECT状态位,监听服务端的TCP ACK应答
clientChannel.register(selector, SelectionKey.OP_CONNECT, ioHandler);
// 6.创建Reactor线程,创建多路复用器并启动线程
Selector selector = Selector.open();
New Thread(new ReactorTask()).start();
// 7.多路复用器在线程run方法的无限循环体内轮询准备就绪的Key
int num = selector.select();
Set selectedKeys = selector.selectedKeys();
Iterator it = selectedKeys.iterator();
while (it.hasNext()) {
SelectedKey key = (SelectedKey) it.next();
}
// 8.接收 connect事件进行处理
if (key.isConnectable())
//handlerConnect();
// 9.判断连接结果,如果连接成功,注册读事件到多路复用器
if (channel.finishConnect())
registerRead();
// 10.注册读事件到多路复用器:
clientChannel.register(selector, SelectionKey.OP_READ, ioHandler);
// 11.异步读客户端请求消息到缓冲区
int readNumber = channel.read(receivedBuffer);
// 12.对ByteBuffer进行编解码,如果有半包消息接收缓冲区Reset,继续读取后续的报文,将解码成功的消息封装成Task,投递到业务线程池中,进行业务逻辑编排
Object message = null;
while(buffer.hasRemain()){
byteBuffer.mark();
Object message = decode(byteBuffer);
if (message == null){
byteBuffer.reset();
break;
}
messageList.add(message);
}
if (!byteBuffer.hasRemain())
byteBuffer.clear();
else
byteBuffer.compact();
if (messageList != null & !messageList.isEmpty()){
for(Object messageE : messageList)
handlerTask(messageE);
}
// 13.将POJO对象encode成ByteBuffer,调用SocketChannel的异步write接口,将消息异步发送给客户端
socketChannel.write(buffer);
2.4 AIO编程
JDK1.7(NIO 2.0)引入了新的异步通道的概念,并提供了异步文件通道和异步套接字通道的实现,是真正的异步IO(因此NIO2.0也称作异步非阻塞IO,而NIO 1.0称作非阻塞IO).
NIO2.0引入了新的异步通道的概念,并提供了异步文件通道和异步套接字通道的实现..异步通道提供两种方式获取获取操作结果.
- 通过
java.util.concurrent.Future类来表示异步操作的结果 - 在执行异步操作的时候传入一个
java.nio.channels.CompletionHandler接口的实现类作为操作完成的回调.
其中异步套接字通道是真正的异步非阻塞IO,对应于Unix网络编程中的事件驱动IO(AIO).它不需要通过多路复用器Selector对注册的通道进行轮询操作即可实现异步读写,从而简化了NIO的编程模型.
2.4.1 服务端的代码
public class TimeServer {
public static void main(String[] args) throws IOException {
//首先创建异步的时间服务器处理类,然后启动线程将AsyncTimeServerHandler启动
AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(8080);
new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
}
}
public class AsyncTimeServerHandler implements Runnable {
CountDownLatch latch;
AsynchronousServerSocketChannel asynchronousServerSocketChannel;
public AsyncTimeServerHandler(int port) {
//在构造方法中,我们首先创建一个异步的服务端通道AsynchronousServerSocketChannel,
//然后调用它的bind方法绑定监听端口,如果端口合法且没被占用,绑定成功,打印启动成功提示到控制台.
try {
asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
//在线程的run方法中,初始化CountDownLatch对象,
//它的作用是在完成一组正在执行的操作之前,允许当前的线程一直阻塞.
//在本例程中,我们让线程在此阻塞,防止服务端执行完成退出.
//在实际项目应用中,不需要启动独立的线程来处理AsynchronousServerSocketChannel,这里仅仅是个demo演示.
latch = new CountDownLatch(1);
doAccept();
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//用于接收客户端的连接,由于是异步操作,
//我们可以传递一个CompletionHandler<AsynchronousSocketChannel,? super A>类型的handler实例接收accept操作成功的通知消息,
//在本例程中我们通过AcceptCompletionHandler实例作为handler来接收通知消息,
public void doAccept() {
asynchronousServerSocketChannel.accept(this, new CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler>() {
@Override
public void completed(AsynchronousSocketChannel result,
AsyncTimeServerHandler attachment) {
//我们从attachment获取成员变量AsynchronousServerSocketChannel,然后继续调用它的accept方法.
//在此可能会心存疑惑:既然已经接收客户端成功了,为什么还要再次调用accept方法呢?
//原因是这样的:当我们调用AsynchronousServerSocketChannel的accept方法后,
//如果有新的客户端连接接入,系统将回调我们传入的CompletionHandler实例的completed方法,
//表示新的客户端已经接入成功,因为一个AsynchronousServerSocket Channel可以接收成千上万个客户端,
//所以我们需要继续调用它的accept方法,接收其他的客户端连接,最终形成一个循环.
//每当接收一个客户读连接成功之后,再异步接收新的客户端连接.
attachment.asynchronousServerSocketChannel.accept(attachment, this);
//链路建立成功之后,服务端需要接收客户端的请求消息,
//创建新的ByteBuffer,预分配1M的缓冲区.
ByteBuffer buffer = ByteBuffer.allocate(1024);
//通过调用AsynchronousSocketChannel的read方法进行异步读操作.
//下面我们看看异步read方法的参数.
//ByteBuffer dst:接收缓冲区,用于从异步Channel中读取数据包;
//A attachment:异步Channel携带的附件,通知回调的时候作为入参使用;
//CompletionHandler<Integer,? super A>:接收通知回调的业务handler,本例程中为ReadCompletionHandler.
result.read(buffer, buffer, new ReadCompletionHandler(result));
}
@Override
public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
exc.printStackTrace();
attachment.latch.countDown();
}
});
}
}
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {
private AsynchronousSocketChannel channel;
public ReadCompletionHandler(AsynchronousSocketChannel channel) {
//将AsynchronousSocketChannel通过参数传递到ReadCompletion Handler中当作成员变量来使用
//主要用于读取半包消息和发送应答.本例程不对半包读写进行具体说明
if (this.channel == null)
this.channel = channel;
}
@Override
public void completed(Integer result, ByteBuffer attachment) {
//读取到消息后的处理,首先对attachment进行flip操作,为后续从缓冲区读取数据做准备.
attachment.flip();
//根据缓冲区的可读字节数创建byte数组
byte[] body = new byte[attachment.remaining()];
attachment.get(body);
try {
//通过new String方法创建请求消息,对请求消息进行判断,
//如果是"QUERY TIME ORDER"则获取当前系统服务器的时间,
String req = new String(body, "UTF-8");
System.out.println("The time server receive order : " + req);
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) ? new java.util.Date(
System.currentTimeMillis()).toString() : "BAD ORDER";
//调用doWrite方法发送给客户端.
doWrite(currentTime);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
private void doWrite(String currentTime) {
if (currentTime != null && currentTime.trim().length() > 0) {
//首先对当前时间进行合法性校验,如果合法,调用字符串的解码方法将应答消息编码成字节数组,
//然后将它复制到发送缓冲区writeBuffer中,
byte[] bytes = (currentTime).getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
//最后调用AsynchronousSocketChannel的异步write方法.
//正如前面介绍的异步read方法一样,它也有三个与read方法相同的参数,
//在本例程中我们直接实现write方法的异步回调接口CompletionHandler.
channel.write(writeBuffer, writeBuffer,
new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer buffer) {
//对发送的writeBuffer进行判断,如果还有剩余的字节可写,说明没有发送完成,需要继续发送,直到发送成功.
if (buffer.hasRemaining())
channel.write(buffer, buffer, this);
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
//关注下failed方法,它的实现很简单,就是当发生异常的时候,对异常Throwable进行判断,
//如果是I/O异常,就关闭链路,释放资源,
//如果是其他异常,按照业务自己的逻辑进行处理,如果没有发送完成,继续发送.
//本例程作为简单demo,没有对异常进行分类判断,只要发生了读写异常,就关闭链路,释放资源.
try {
channel.close();
} catch (IOException e) {
// ingnore on close
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
this.channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
2.4.1 客户端的代码
public class TimeClient {
public static void main(String[] args) {
//通过一个独立的I/O线程创建异步时间服务器客户端handler,
//在实际项目中,我们不需要独立的线程创建异步连接对象,因为底层都是通过JDK的系统回调实现的.
new Thread(new AsyncTimeClientHandler("127.0.0.1", 8080), "AIO-AsyncTimeClientHandler-001").start();
}
}
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;
public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>, Runnable {
private AsynchronousSocketChannel client;
private String host;
private int port;
private CountDownLatch latch;
//首先通过AsynchronousSocketChannel的open方法创建一个新的AsynchronousSocketChannel对象.
public AsyncTimeClientHandler(String host, int port) {
this.host = host;
this.port = port;
try {
client = AsynchronousSocketChannel.open();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
//创建CountDownLatch进行等待,防止异步操作没有执行完成线程就退出.
latch = new CountDownLatch(1);
//通过connect方法发起异步操作,它有两个参数,
//A attachment：AsynchronousSocketChannel的附件,用于回调通知时作为入参被传递,调用者可以自定义;
//CompletionHandler<Void,? super A> handler：异步操作回调通知接口,由调用者实现.
client.connect(new InetSocketAddress(host, port), this, this);
try {
latch.await();
} catch (InterruptedException e1) {
e1.printStackTrace();
}
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
//异步连接成功之后的方法回调——completed方法
@Override
public void completed(Void result, AsyncTimeClientHandler attachment) {
//创建请求消息体,对其进行编码,然后复制到发送缓冲区writeBuffer中,
//调用Asynchronous SocketChannel的write方法进行异步写.
//与服务端类似,我们可以实现CompletionHandler <Integer, ByteBuffer>接口用于写操作完成后的回调.
byte[] req = "QUERY TIME ORDER".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
writeBuffer.put(req);
writeBuffer.flip();
client.write(writeBuffer, writeBuffer,
new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer buffer) {
//如果发送缓冲区中仍有尚未发送的字节,将继续异步发送,如果已经发送完成,则执行异步读取操作.
if (buffer.hasRemaining()) {
client.write(buffer, buffer, this);
} else {
//客户端异步读取时间服务器服务端应答消息的处理逻辑
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
//调用AsynchronousSocketChannel的read方法异步读取服务端的响应消息.
//由于read操作是异步的,所以我们通过内部匿名类实现CompletionHandler<Integer,ByteBuffer>接口,
//当读取完成被JDK回调时,构造应答消息.
client.read(readBuffer,readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result,ByteBuffer buffer) {
//从CompletionHandler的ByteBuffer中读取应答消息,然后打印结果.
buffer.flip();
byte[] bytes = new byte[buffer.remaining()];
buffer.get(bytes);
String body;
try {
body = new String(bytes,"UTF-8");
System.out.println("Now is : " + body);
latch.countDown();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
//当读取发生异常时,关闭链路,
//同时调用CountDownLatch的countDown方法让AsyncTimeClientHandler线程执行完毕,客户端退出执行.
try {
client.close();
latch.countDown();
} catch (IOException e) {
// ingnore on close
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
// ingnore on close
}
}
});
}
@Override
public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
exc.printStackTrace();
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
}