第22章 NIO编程-编程知识

在本章中需要掌握NIO中的缓冲区的作用，并理解缓冲区中的数据处理模型，掌握Channel的作用，并结合缓冲区实现数据I/O操作，理解文件锁的作用，并且掌握字符编码处理支持类的使用，掌握Reactor设计模型，可以实现同步非则色I/O通信模型，掌握异步非阻塞I/O通信模型。
NIO(Non-blocking I/O,非阻塞I/O，或被称为NewIO)是在JDK1.4后提供的意向重要开发包-因为有了NIO的出现才使得Java底层通信的性能得到了大幅度的提升。在NIO中猜中非阻塞模式并结合通道(Channel)与缓冲区(Buffer)实现I/O通信操作，本章将讲解NIO模型及后续推出的AIO模型
提示：BIO、NIO、AIO的区别
JavaBIO:同步阻塞I/O。服务器为每一个连接的客户端分配一个线程，可以通过线程池提高线程的管理机制
javaNIO:同步阻塞I/O。服务器为每一个请求分配一个线程，所有的请求都会注册到多路复用器中，多路复用器采用轮训的形式针对每次请求创建处理线程。
javaAIO(NIO.2):异步非阻塞I/O，客户端的I/O请求都是由系统先处理，当处理完成后再通知服务器启动线程进行处理。

22.1 NIO简介

NIO提供了一个全新的底层I/O模型。与最初的java.io包中面向流（Stream Oriented）的概念不同，NIO中采用了面向缓冲区(Buffer Oriented)的概念，这样所有的数据被读取到缓冲区后，可以直接利用指针实现缓冲区的数据读取控制，同时在进行更多数据缓存时也可以保证不覆盖原始数据的前提下进行。
传统的JavaIO采用的是阻塞形式，这样当进行读或写时，该线程一直处于阻塞状态，一直到读取或写入完成前都无法进行其他任何的操作，这样在进行网络通信过程中就会因为线程阻塞而影响到程序性能，其模式如下图在Java发展初期，由于JVM的优化性能不高，所以Java程序的运行速度很慢，此时对于I/O的性能没有过多的要求。然而随着JVM的不断优化，JVM中的字节码程序的执行性能已经接近或超过本地系统程序的处理性能，同时随着硬件技术的发展，大部分的Java程序已经不在受CPU的束缚，所以此时I/O通信的性能问题就尤为突出。在这样的背景下，Java退出了java.nio开发包，在NIO中采用可非阻塞设计模型，这样在没有数据进行读写的情况下不会产生阻塞，同时线程可以继续完成其他操作，并且一个线程可以通过选择器(Selector)同时管理多个输入\输出通道。

22.2 Buffer

缓冲区(Buffer)是一个线性的、有序的数据集，一个缓冲区只能够容纳一种数据类型，Buffer类定义如下
public abstract class Buffer extends Object{}
Buffer是一个抽象类，所以提供了一个缓存操作的标准，如果要保存不同的数据类型，则应该使用Buffer的不同子类，常见的Buffer子类为ByteBuffer、CharBuffer、ShortBuffer、InBuffer、LongBuffer、FloatBuffer、DoubleBuffer在Buffer中存在一系列的状态变量，这些状态变量随着写入或读取都有可能发生改变。在缓冲区可以使用3个值表示缓冲区的状态
poison：表示下一个缓冲区读取或写入的操作指针，每箱缓冲区写入数据的时候此指针就会改变，指针永远放到写入的最后一个元素后，即如果写入了4个位置的数据，则position会执行第5个数据。
limit:表示还有多少个数据需要存储或读取，position<=limit
capacity:表示缓冲区的最大容量，limit<=capacity.此值再分配缓冲区时被设置，一般不会更改，为了进一步说明这3个指针的作用，下面通过一个具体的案例进行说明
public static void main(String[]args)
{

String str="AAA";//定义操作数据，长度为3
ByteBuffer buffer=ByteBuffer.allocate(20);创建缓冲区，容量为20
System.out.println("【1】没有存放数据:capacity="+buffer.capacity()+"limit="+buffer.limit()+"position"+buffer.position());
buffer.put(str.getBytes());
System.out.println("【2】保存数据:capacity="+buffer.capacity()+"limit="+buffer.limit()+"position"+buffer.position());
buffer.flip();
System.out.println("【3】没有数据:capacity="+buffer.capacity()+"limit="+buffer.limit()+"position"+buffer.position());
while(buffer.hasRemaining())
{
System.out.println(buffer.get()+"");返回字节数据
}
System.out.println();//换行
buffer.clear();//清空缓冲区
System.out.println("【4】清空缓冲区:capacity="+buffer.capacity()+"limit="+buffer.limit()+"position"+buffer.position());
}

22.3 Channel

通道(Channel可以用来读取和写入数据，通道类似于之前的输入、输出流，但是程序不会直接操作通道，所有的内容都是先读取或写入缓冲区，在通过缓冲区取得或写入。
通道与传统的流操作不同，传统的流操作分为输入流和输出流，而通道本身是双向操作的，既可以完成输入也可以玩成输出，Channel本身是一个接口标准。

22.3.1 FileChannel

FileChannel是Channel的子类，可以进行文件的通道读写操作，如果要想使用FileChannel，则可以依靠FileInputStream或FileOutputStream类中的getChannel()方法获得输入或输出的通道。下面通过文件通道实现数据读取。

范例：使用文件通道读取数据

package cn.mldn.demo;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
public class FileChannels.FileChannel;
public class FileChannelDemo
{
public static void main(String[]args)throws Exception
{
File file=new File("D:"+File.separator+"info.txt");//定义文件路径
FileInputStream input=new FileInputStream(file);//文件输入流
FileChannel channel=input.getChannel();//获得文件通道
ByteBuffer buffer=ByteBuffer.allocate(20);//开辟缓冲大小
ByteArrayOutputStream bos=new ByteArrayOutputStream();//内存输出流
int count=0;//保存读取个数
while((count=channel.read(buffer))!=-1)//缓冲区读取
{
buffer.flip();//重置缓冲区
while(buffer.hasRemaining())
{
//是否还有数据
bos.write(buffer.get());//内容写入内存流
}
buffer.clear();
}
System.out.println(new String(bos.toByteArray()));//字节数组转字符串
channel.close();
input.close();
}
}

本程序通过文件通道实现了文件数据的读取操作，在读取中需要通过缓存来接收返回的数据内容。由于读取的文件较多，所以在每一次进行缓存操作时都需要进行缓存的重置与清空操作。

22.3.2 Pipe

Channel针对线程管道的I/O操作也提供有专门的通道，为此定义了俩个类型：Pipe.SinkChannel(管道数据输入)。下面江界河多线程实现Pipe通道操作
public static void main(String[]args)throws Exception
{
//打开管道流
Pipe pipe=Pipe.open();
new Thread(()->
{
Pipe.SourceChannel sourceChannel=pipe.source();//打开管道输入流
ByteBuffer buffer=ByteBuffer.allocate(50);//开辟缓存空间
int count=sourceChannel.read(buffer);//返回读取个数
buffer.flip();//重置缓冲区
System.out.println("接收端"+new String(buffer.array(),0,count));
},"接收线程").start();
new Thread(()->
{
String msg=""+Thread.currentThread().getName()+"AAA";//信息
Pipe.SinkChannel sinkChannel=pipe.sink();//获取管道输出流
ByteBuffer buffer=ByteBuffer.allocate(50);//开辟缓冲区
buffer.put(msg.getBytes());//设置要发送的数据
buffer.flip();//重置缓冲区
while(buffer.hasRemainning())//缓冲区有数据
{
sinkChannek.write(buffer);
}
},"发送线程").start();
}

22.4 文件锁

在Java新I/O中提供了文件锁的功能，这样当一个线程将文件锁定后，其他线程是无法操作此文件的。要想进行文件的锁定操作，则要使用FileLock类操作，FileLock类依靠FileChannel进行实例化操作。

范例：锁定文件

public class FileLockDemo
{
public static void main(String[]args)throws Exception
{
File file=new File("D:"+File.separator+"info.txt");//定义锁定文件路径
FileOutputStream output=new FileOutputStream(file);//获取文件输出流
FileChannel channel=output.getChannel();//文件通道
FileLock lock=channel.tryLock();//获取文件锁
if(lock!=null)
{
TimeUnit.SECONDS.sleep(30);//释放锁
}
channel.close();
output.close();
}
}
本程序获取了一个指定文件的文件锁，并且延迟30秒后才进行锁释放，在这期间文件无法操作。

22.5 字符集

在java语言中所有的信息都是以Unicode进行编码的，但是在计算机的世界里并不只是单单存在一种编码，而是多个，在I/O通信过程中若是编码处理不恰当，则就可能产生乱码。在java的新i/O包中提供了Charset类来处理编码问题，该类还包括了创建编码器(CharsetEncoder)和创建解码器(CharsetDecoder)的操作。

提示：编码器和解码器
编码和解码实际上是从最早的电报发展起来的，所有的内容如果需要使用电报传送，则必须变为对应的编码，之后再通过指定的编码进行解码的操作。在新i/O中为了保证程序可以适应不同的编码，所以提供了编码器和解码器，通过解码器程序可以方便的读取各个平台平台上不同编码的数据，之后在通过编码器讲程序的内容以正确的形式进行输出。

范例：编码和解码

public class EncodeAndDecodeDemo
{
publid static void main(String[]args)throws Exception
{
Charset charset=Charset.forName("UTF-8");//创建指定编码
CharsetEncoder encoder=charset.newEncoder();//获取编码类对象
CharsetDecoder decoder=charset.newDecoder();//获取解码类对象
String str="AAA";//字符数据
CharBuffer buf=CharBuffer.allocate(20);//缓冲区
buf.put(str);//向字符穿冲去保存数据
buf.flip();//缓冲区重置
ByteBuffer buffer=encoder.encode(buf);//编码处理
System.out.println(decoder.decode(buffer));//解码处理
}
}

本程序采用UTF-8编码获取了编码器和解码器，这样就可以直接进行缓冲区数据的编码与解码操作。

22.6 同步非阻塞I/O通信模型

NIO推出的主要目的就是解决I/O处理性能问题，而传统I/O中最大的情况在于它属于同步阻塞I/O通信，也就是说一个线程在进行操作的时候，其他的线程都无法进行处理。如果说现在只是一个单机版的程序，那么没有任何问题，而如果该程序用于网络通信，那么这个问题比较严重，所以真正的NIO应该应用在高效的网络传输程序中。
网络通信就是一个基本的通道连接，在NIO中提供有两个新的通道类：ServerSocketChannel、SocketChannel，为了方便的进行所有通道的管理，NIO提供了一个Selector通道管理类，这样所有的通道都可以直接向Selector进行注册，并采用统一的模式进行读、写操作，这样的设计称为Reactor模式。
在进行非阻塞网络开发的时候需要使用SelectableChannel类向Select类注册，SelectableChannel提供了注册Selector的方法和阻塞模式。ServerSocketChannel描述了服务器通道。
在使用register()方法的时候需要指定一个选择器(Selector对象)以及Select域、Selector对象可以通过Selector中的open()方法取得，而Selector域则在SelectionKey类中定义。
下面将结合以上所给出的操作类，利用NIO实现一个Echo程序模型
class SocketClientChannelThread implements Runnable
{
//客户端处理线程
private SocketChannel clientChannel;//客户端通道
private boolean flag=true;//循环标记
public SocketClientChannelThread(SocketChannel clientChannel)throws Exception
{
this.clientChannel=clientChannel;//保存客户端连接
System.out.println("客户端连接成功，该客户端的地址为"+clientChannel.getRemoteAddress());
}
@Override
public void run()
{
//线程任务
ByteBuffer buffer=ByteBuffer.allocate(50);//创建缓冲区
while(this.flag)
{
//不断与客户端交互
buffer.clear();
int readCount=this.clientChannel.read(buffer);//接收客户端发送的数据
String readMessage=new String(buffer.array(),0,readCount).trim();//数据变为字符串
System.out.println("服务器接收到消息"+readMessage);//提示消息
String writeMessage="ECHO"+readMessage+"\n";
if("exit".equals(readMessage))
{
writeMessage="exit拜拜，下次再见";//结束消息
this.flag=false;
}
buffer.clear();//清空缓冲区
buffer.put(writeMessage.getBytes());//缓冲区保存数据
buffer.clip();//重置缓冲区
this.clientChannel.write(buffer);
}
this.clientChannel.write(buffer);//回应消息
}
}

public class EchoServer
{
public static final int PORT=9999;//绑定端口
public static void main(String[]args)throws Exception
{
//考虑到性能的优化所以只允许5个用户进行访问
ExecutorService executorService=Executors.newFixedThreadPool(5);
//打开一个服务器端的Socket连接通道
ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);//设置非阻塞模式
serverSocketChannel.bind(new InetSocketAddress(PORT));//服务绑定端口
//打开一个选择器，随后所有的Channel都要注册到此选择器中
Selector selector=Selector.open();
//将当前ServerSocketChannel统一注册到Selector中，接受统一管理
serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);
System.out.println("服务器端启动程序，该程序"+port+"端口上监听，等待客户端连接")
//所哟肚饿连接都需要被Selector多管理，也就是说，只要有新的用户连接，那么就通过Selector处理
int keySelect=0;//接收连接状态
while((keySelect=selector.select())>0)
{
//持续等待接收
Set<SelectionKey>selectedKeys=selector.selectedKeys();//获取全部连接通道
Iterator<SelectionKey>selectionLey=selectedKeys.iterator();
while(selectionIter.hasNext())
{
SelectionKey.isAcceptionIter.next();//获取每一个通道
if(selectionKey.isAcceptable())
{
//模式为接收连接模式
SocketChannel clientChannel=serverSocketChannel.accept();//等待接收
if(clientChannel!=null)
{
executorService.submit(new SocketClientChannelThread(clientChannel));
}
}
selectionIter.remove();
}
}
executoeService.shutdown();
serverSocketChannel.close();//关闭服务器端通道
}
}

在本服务器程序中首先定义了一个可以处理的线程池，随后将当前服务器通道设置为非阻塞模式，对于NIO实现的服务器端来讲需要采用轮训的模式获取每一个连接的客户端细腻，并且将客户端的处理操作包装在一个线程类中，同时将此线程交给线程池进行调度和管理，实现Echo操作。

范例：定义一个输入工具类，实现键盘数据接收
public class InputUtil
{
public static final BufferedReader KEYBOARD_INPUT=new BufferReader(new InputStreamReader(System.in));//键盘缓冲输入流
private InputUtil(){}
public static String getString(String prompt)throws Exception
{
//键盘接收数据
boolean flag=true;//输入标记
String str=null;//接收输入字符串
while(flag)
{
System.out.print(prompt);//提示信息
str=KEYBOARD_INPUT.readLine();//读取数据
if(str==null||"".equals(str))
{
System.out.println("");
}else
{
flag=false;
}
}
return str;
}
}

本程序将由键盘输入信息，而后通过客户端向服务器端发送消息

范例：实现客户端Socket
public class EchoClient
{
public static final String HOST="localhost";//连接主机
public static final int PORT=9999;//绑定端口
public static void main(String[]args)throws Exception
{
SocketChannel clientChannel=SocketChannel.open();//获取客户端通道
clientChannel.connect(new InetSocketAddress(HOST,PORT));//连接服务器端
ByteBuffer buffer=ByteBuffer.allocate(50);//开辟缓存
boolean flag=true;
while(flag)
{
//持续输入消息
buffer.clear();//清空缓冲区
String msg=InputUtil.getString("请输入要发送的消息:");//提示信息
buffer.put(msg.getBytes());//数据保存在缓冲区
buffer.flip();//重设缓冲区
clientChannel.write(buffer);//发送消息
buffer.clear();//清空缓冲区
int readCount=clientChannel.read(buffer);//读取服务器端回应
buffer.flip();//重置缓冲区
System.err.println(new String(buffer.array(),0,readCount));
if("exit".equals(msg)){flag=false;}
}
clientChannel.close();
}

}

22.7 异步非阻塞I/O通信模型

NIO是基于事件驱动模式实现的通信操作，主要的问题是解决BIO并发访问量高的性能问题，而所有的通信操作依然是通过程序来完成，在进行通信处理中如果I/O操作性能较差会影响到执行性能，所以从JDK1.7开始提供有AIO模型。与NIO不同的是，AIO当前I/O操作是由操作系统进行I/O操作，而应用程序只是调用给定的类库实现读或写的操作调用，例如，当有数据流可以读取或写入时，会由操作系统讲课操作的流传入read或write()方法的缓冲区并发出操作通知，整个操作完全是异步处理实现的，
在进行异步操作时通过CompletionHandler获取异步执行结果，在Echo案例中数据读取与数据写入都可以编写一个CompletionHandler作为回调操作实现。

范例：定义服务器端

class EchoHandler implements CompletionHandler<Interger,ByteBuffer>
{
//实现回调处理
private AsynchronousSocketChannel clientChannel;//客户端对象
private boolean exit=false;//结束标记
public EchoHandler(AsynchronousSocketChannel clientChannel)
{
this.clientChannel=clientChannel;
}
@Override
public void completed(Interger result,ByteBuffer buffer)
{
//回调任务
buffer.flip();//重置缓冲区
String readMessage=new String(buffer.array(),0,buffer,remaining()).trim();//接收读取数据
System.err.println("服务器端读取到数据"+readMessage);//回应信息
String resultMessage="ECHO"+readMessage;//回应信息
if("exit".equalsIgnoreCase(readMessage))
{
resultMessage="exit";
this.exit=true;
}
this.echoWrite(resultMessage);
}
@Override
public void failed(Throwbale exp,ByteBuffer buffer)
{
//异步处理失败
//关闭连接
this.closeClient();
}
private void closeClient()
{
System.out.println("客户端连接有错误");
this.clientChannel.close();
}
private void echoWrite(String result)
{
//数据回应
ByteBuffer buffer=ByteBuffer.allocate(100);//回应缓冲区
buffer.put(result.getBytes());//回应处理
buffer.flip();//重置缓冲区
this.clientChannel.write(buffer,buffer,new CompletionHandler<Interger,ByteBuffer>()
{
@Override
public void completed(Interger result,ByteBuffer buffer)
{
if(buffer.hasRemaining())
{
EchoHandler.this.clientChannel.write(buffer,buffer,this);//输出
}else{
if(EchoHandler.this.exit==false)
{
ByteBuffer readBuffer=ByteBuffer.allocate(100);
EchoHandler.this.clientChannel.read(readBuffer,readBuffer,new EchoHandler(EchoHandler.this.clientChannel));
}
}
}
@Override
public void failed(Throwable exp,ByteBuffer buffer){EchoHandler.this.closeClient();}
});
}
}
class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel,AIOServerThread>
{
@Override
public void completed(AsynchronousSocketChannel channel,AIOServerThread aioThread)
{
aioThread.getServerChannel().accept(aioThread,this);//接收连接
ByteBuffer buffer=ByteBuffer.allocate(100);//接收缓冲区
channel.read(buffer,buffer,new EchoHandler(channel));//交由其他异步处理
}
@Override
public void failed(Throwable exp,AIOServerThread aiuThread)
{
aioThread.getLatch().countDown();//接触阻塞状态
}
}

class AIOServerThread implements Runnable
{
//AIO处理线程
private static final int PORT=9999;//监听端口
private CountDownLatch latch=null;//服务器端关闭阻塞
private AsynchronousServerSocketChannel serverChannel=null;//服务器端关闭阻塞
public AIOServerThread()throws Exception
{
//服务器端阻塞线程数
this.latch=new CountDownLatch(1);//服务器端阻塞线程数
this.serverChannel=AsynchronousServerSocketChannel.open();//异步通道
this.serverChannel.bind(new InetSocketAddress(PORT));//绑定端口
}
public AsynchronousServerSocketChannel getServerChannel()
{
return serverChannel;
}
public CountDownLatch getLatch(){return latch;}
@Override
public void run()
{
//启动线程
this.serverChannel.accept(this,new AcceptHandler());//等待连接
this.latch.await();//保持连接
}
}
public class AIOEchoServer
{
public static void main(String[]args)throws Exception
{
new Thread(new AIOServerThread()).start();
}
}