NIO是Java 1.4开始引入的,目的是替代标准IO,它采用了与标准IO完全不同的设计模式和工作方式,这里就来总结一下。
1.Buffer
正如他的名字,就是一个缓存,实际上是内存的一块区域,它是NIO体系的重要组成部分,主要和通道进行交互。 Buffer本身是一个抽象类,它有以下几个子类:
ByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer
根据缓存的数据类型不同创建的不同子类,大致功能都类似,我们不一个一个介绍,只介绍Buffer的关键点。
Buffer的使用一般如下:将数据写入buffer,准备从buffer中读数据,读取数据,清空buffer。下面先简单演示一下然后再解释:
public static void main(String[] args) throws IOException {
try (RandomAccessFile file = new RandomAccessFile("file/test.txt","rw");
FileChannel channel = file.getChannel()){
ByteBuffer buffer = ByteBuffer.allocate(5);
int len;
while((len = channel.read(buffer))!=-1){
buffer.flip();
while (buffer.hasRemaining())
System.out.println((char)buffer.get());
buffer.compact();
}
}catch (Exception e){
e.printStackTrace();
}
}
先看构造,Buffer的子类也都是抽象类,不能直接实例化,都需要调用静态方法生成,可以看源码,调用不同的静态方法实例化不同的实现类。以ByteBuffer为例,可以这样实例化:
allocate(int capacity) //分配一个大小为capacity的字节数组作为缓冲,但是在堆中
allocateDirect(int capacity) //和上面类似,不过直接借助系统在内存中创建,速度较快,但消耗性能
wrap(byte[] array) //直接从外部指定一个数组,不适用默认创建的,但是双方一方改动就会影响另一方
wrap(byte[] array, int offset, int length) //和上面一个一样,但是能指定偏移量和长度
我们获得一个Buffer实例后就可以使用,首先向buffer中写东西需要channel配合,调用read方法即可。之后在读之前需要准备一下,从代码看就是调用flip方法,这个方法有什么作用呢?从文档上看,就是将limit设置为position,然后将position 置零。这样有什么用呢?下面就来介绍一下buffer的几个成员变量:capacity,limit,position。
capacity就是一个buffer的固定大小,表示他的容量
position表示当前指针的位置,也就是当前读或写到的位置
limit这个值在写模式下表示最多能写多少,写模式下等于capacity。读模式下表示能读到多少,调用flip将limit等于position,表示最多能读到之前写入的所有内容
看flip的实现也很好理解,如下:
public final Buffer flip() {
limit = position;
position = 0;
mark = -1;
return this;
}
刚才是准备读数据,下面就是从中读了,buffer有一系列get方法,和流的read方法类似。既然有get就有put方法,除了上面和channel配合写入东西,还可以用一系列put方法写入。读之前可以判断一下是否还有数据:hasRemaining();读完之后为了使下次还能用,需要清空buffer,可以用clear方法或者compact方法。可以看一下他们的实现:
public final Buffer clear() {
position = 0;
limit = capacity;
mark = -1;
return this;
}
public ByteBuffer compact() {
System.arraycopy(hb, ix(position()), hb, ix(0), remaining());
position(remaining());
limit(capacity());
discardMark();
return this;
}
clear方法很清晰,就是将几个游标归为为原始状态。compact也是设置几个游标位置,不过有点特殊,remaining方法是获取剩余数据数量就是limit - position,然后将该值赋给position,然后将capacity赋值给limit。他和clear的区别就是在position上的处理。但是,如果我们已经将buffer内容读完,这时limit = position,那么 position(remaining())的效果就是position = limit - position = 0.这时compact方法效果等于clear。否则虽然也可以继续写内容进去,但容量减少,但好处是未读的数据以后可以继续读。
再来看其他方法:
public final Buffer rewind() {
position = 0;
mark = -1;
return this;
}
rewind是将position 置零,也就是buffer中内容可以重新读取。
public final Buffer mark() {
mark = position;
return this;
}
public final Buffer reset() {
int m = mark;
if (m < 0)
throw new InvalidMarkException();
position = m;
return this;
}
mark和reset是配合使用的。mark标记一个位置,reset使游标回到这个位置。mark成员变量初始为-1.所以不要没有调用mark方法就去调用reset。
可以看到buffer的主要操作就是针对几个指针的,毕竟他是依赖于数组实现的。
2.Channel
Channel用来实现通道的概念,他类似于流,但是不能直接操作数据,需要借助于Buffer,它本身是一个接口,一般有以下几个重要实现:
FileChannel
DatagramChannel
SocketChannel
ServerSocketChannel
2.1 FileChannel
FileChannel是一个用于读写操作文件的channel。首先看怎么获得实例化对象,FileChannel也是一个抽象类,所以不能通过构造获得。一般获得的途径有,RandomAccessFile、FileOutputStream、FileInputStream等一些类的getChannel()方法,如上文中示例。
另外在Java 1.7 中提供了几个静态的open方法用来直接打开或创建文件获取Channel:
public static void main(String[] args) throws IOException {
try (FileChannel channel = FileChannel.open(Paths.get("file/test.txt"), StandardOpenOption.READ)){
ByteBuffer buffer = ByteBuffer.allocate(5);
int len;
while((len = channel.read(buffer))!=-1){
buffer.flip();
while (buffer.hasRemaining())
System.out.println((char)buffer.get());
buffer.compact();
}
}catch (Exception e){
e.printStackTrace();
}
}
Channel是不能直接读数据的,需要借助于buffer,同样写内容也是要借助于buffer,下面演示一下传统的复制文件。
public static void main(String[] args) throws IOException {
try (FileChannel readChannel = FileChannel.open(Paths.get("file/test.png"), StandardOpenOption.READ)){
FileChannel writeChannel = FileChannel.open(Paths.get("file/copy.png"), StandardOpenOption.READ,StandardOpenOption.WRITE,StandardOpenOption.CREATE);
ByteBuffer buffer = ByteBuffer.allocate(1024);
while (readChannel.read(buffer)!=-1){
buffer.flip();
while (buffer.hasRemaining())
writeChannel.write(buffer);
buffer.clear();
}
}catch (Exception e){
e.printStackTrace();
}
}
有一点需要注意的是,有时并不能保证把整个buffer的内容写入,为了严谨起见,需要循环判断buffer中是否有内容未写入。
除了上面传统的写法,channel还有自己特有的传输方法:
try (FileChannel readChannel = FileChannel.open(Paths.get("file/test.png"), StandardOpenOption.READ)){
FileChannel writeChannel = FileChannel.open(Paths.get("file/copy.png"), StandardOpenOption.READ,StandardOpenOption.WRITE,StandardOpenOption.CREATE);
//以下两句话效果一样
//writeChannel.transferFrom(readChannel,0,readChannel.size());
readChannel.transferTo(0,readChannel.size(),writeChannel);
}catch (Exception e){
e.printStackTrace();
}
transferFrom和transferTo都是把一个channel的内容传输到另一个,但是注意两个方法的区别,即方向性。
上面示例用到了size()方法,是用来获取所关联文件的大小。
position()和position(long)方法用来获取指针位置和设置指针位置。设置position是可以将指针设置到文件结束符之后的,但是中间会有空洞。
truncate(long)方法可一截取一个文件并返回FileChannel,从文件开始截取到指定位置。
2.2 DatagramChannel
DatagramChannel是Java UDP通信中传输数据的通道。关于Java中传统UDP的实现见这里,下面简单用DatagramChannel实现一下UDP通信
服务端
public class UDPService {
public static final String SERVICE_IP = "127.0.0.1";
public static final int SERVICE_PORT = 10101;
public static void main(String[] args) {
UDPService service = new UDPService();
service.startService(SERVICE_IP,SERVICE_PORT);
}
private void startService(String ip, int port){
try (DatagramChannel channel = DatagramChannel.open()){
channel.bind(new InetSocketAddress(ip,port));
while (true){
ByteBuffer buffer = ByteBuffer.allocate(1024);
SocketAddress socketAddress = channel.receive(buffer);
String receive = new String(buffer.array(),"UTF-8").trim();
System.out.println("address: " + socketAddress.toString()+ " msg: "+ receive);
buffer.clear();
buffer.put((receive + "hello world").getBytes());
buffer.flip();
channel.send(buffer,socketAddress);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
客户端
public class UDPClient {
public static void main(String[] args){
UDPClient client = new UDPClient();
Scanner scanner = new Scanner(System.in);
while(true){
String msg = scanner.nextLine();
if("##".equals(msg))
break;
System.out.println(client.sendAndReceive(UDPService.SERVICE_IP,UDPService.SERVICE_PORT,msg));
}
}
private String sendAndReceive(String serviceIp, int servicePort, String msg) {
try (DatagramChannel channel = DatagramChannel.open()){
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.put(msg.getBytes());
System.out.println(buffer.position());
buffer.flip();
SocketAddress address = new InetSocketAddress(serviceIp,servicePort);
System.out.println( channel.send(buffer,address));
buffer.clear();
SocketAddress socketAddress = channel.receive(buffer);
return "address: " + socketAddress.toString()+ " msg: "+ new String(buffer.array(),"UTF-8");
} catch (IOException e) {
e.printStackTrace();
}
return "null";
}
}
一般获得一个DatagramChannel 需要使用静态方法open,DatagramChannel 也是配合buffer使用的。之后服务端需要绑定一个地址和端口。接下来就可以收发数据了,收用receive方法,返回发送方的地址信息,发生使用send方法,返回成功发生的字节数。
有一点特别重要,由于是配合buffer操作,无论是客户端还是服务端,在发送前都需要调用flip方法,否则发送的都是空数据(因为都是从position到limit,flip之前position是当前写的位置,limit为capacity)。还有一点,在接受时,由于不知道buffer中有效字节数,所以limit为capacity,直观的看就是转为字符串时末尾有大量空内容,需要trim一下。
默认情况下是阻塞的,也可以设置为非阻塞的,channel.configureBlocking(true);,此时receive方法会立刻返回,可能为null。channel也有connect方法,但是UDP是非连接的,所以只是绑定一个远端地址,收发智能从指定地址来。connect之后就可以用read或者write收发数据。
2.3 SocketChannel 与 ServerSocketChannel
这两类是和Socket与 ServerSocket对应的两个雷,也是专为TCP通信设计的,ServerSocketChannel代表服务端,SocketChannel 代表一个连接。关于Java的传统TCP实现见这里,下面简单实现一下TCP通信
服务端:
public class TCPService {
public static final String SERVICE_IP = "127.0.0.1";
public static final int SERVICE_PORT = 10101;
public static void main(String[] args) {
TCPService service = new TCPService();
service.startService();
}
private void startService(){
try (ServerSocketChannel service = ServerSocketChannel.open()){
service.bind(new InetSocketAddress(SERVICE_IP,SERVICE_PORT));
while (true){
SocketChannel channel = service.accept();
ByteBuffer buffer = ByteBuffer.allocate(1024);
StringBuilder msg = new StringBuilder();
while ((len = channel.read(buffer)) > 0) {
receive.append(new String(buffer.array(), 0, len));
buffer.clear();
}
System.out.println("address: " + channel.getRemoteAddress().toString() + " msg: " + msg.toString());
buffer.clear();
buffer.put((msg + "hello world").getBytes());
buffer.flip();
while (buffer.hasRemaining())
channel.write(buffer);
channel.shutdownOutput();
}
}catch (Exception e){
e.printStackTrace();
}
}
}
客户端
public class TCPClient {
public static void main(String[] args) {
Scanner scanner = new Scanner(System.in);
TCPClient client = new TCPClient();
while(true){
System.out.println(client.sendAndReceive(TCPService.SERVICE_IP,TCPService.SERVICE_PORT,scanner.nextLine()));
}
}
private String sendAndReceive(String address,int port,String msg){
try (SocketChannel channel = SocketChannel.open()){
channel.connect(new InetSocketAddress(address,port));
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.put(msg.getBytes());
buffer.flip();
while (buffer.hasRemaining())
channel.write(buffer);
channel.shutdownOutput();
buffer.clear();
StringBuilder receive = new StringBuilder();
int len = 0;
while ((len = channel.read(buffer)) > 0) {
receive.append(new String(buffer.array(), 0, len));
buffer.clear();
}
return "address: " + channel.getRemoteAddress().toString() + " msg: " + receive.toString();
}catch (Exception e){
e.printStackTrace();
}
return "null";
}
}
同样都是利用open获取一个示例,服务端要绑定地址和端口,然后监听连接,客户端只需去连接服务端即可。同样的都可以设置为非阻塞的,收发数据使用read和write方法。需要注意的还是buffer操作问题以及即使关流或者做控制就行。
3.Selector
Selector 可以同时监控多个Channel 的 IO 状况,也就是说,利用 Selector可使一个单独的线程管理多个 Channel,selector 是非阻塞 IO 的核心。简单示例
客户端
public class TCPClient {
public static void main(String[] args) {
Scanner scanner = new Scanner(System.in);
TCPClient client = new TCPClient();
while(true){
client.sendAndReceive(TCPService.SERVICE_IP,TCPService.SERVICE_PORT,scanner.nextLine());
}
}
private void sendAndReceive(String address,int port,String msg){
try (SocketChannel channel = SocketChannel.open()){
channel.connect(new InetSocketAddress(address, port));
ByteBuffer buf = ByteBuffer.allocate(1024);
channel.configureBlocking(false);
buf.put((new Date() + ":" + msg).getBytes());
buf.flip();
channel.write(buf);
buf.clear();
channel.shutdownOutput();
}catch (Exception e){
e.printStackTrace();
}
}
}
服务端
public class TCPService {
public static final String SERVICE_IP = "127.0.0.1";
public static final int SERVICE_PORT = 10101;
private String msg;
public static void main(String[] args) {
TCPService service = new TCPService();
service.startService();
}
private void startService(){
try (ServerSocketChannel service = ServerSocketChannel.open()){
service.bind(new InetSocketAddress(SERVICE_IP,SERVICE_PORT));
service.configureBlocking(false);
Selector selector = Selector.open();
service.register(selector, SelectionKey.OP_ACCEPT);
while (selector.select() > 0) {
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
while (it.hasNext()) {
SelectionKey key = it.next();
if (key.isAcceptable()) {
System.out.println("isAcceptable");
SocketChannel sc = service.accept();
sc.configureBlocking(false);
sc.register(selector, SelectionKey.OP_READ );
} else if (key.isReadable()) {
System.out.println("isReadable");
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer buf = ByteBuffer.allocate(1024);
int len = 0;
StringBuilder sb = new StringBuilder();
while ((len = channel.read(buf)) > 0) {
sb.append(new String(buf.array(), 0, len));
buf.clear();
}
System.out.println(sb.toString());
channel.close();
}
it.remove();
}
}
}catch (Exception e){
e.printStackTrace();
}
}
}
主要是服务端的应用,基本流程就是先利用open()方法获取一个Selector ,设置ServerSocketChannel 为非阻塞的,之后注册事件,一般有以下几种
SelectionKey.OP_CONNECT
SelectionKey.OP_ACCEPT
SelectionKey.OP_READ
SelectionKey.OP_WRITE
之后调用select()返回就绪通道数,然后根据时间类型执行具体操作即可。
4.Pipe
传统IO为我们提供了线程间通信的类,PipedInputStream与PipedOutputStream。NIO作为IO的替代者,自然也有线程通信的方法,就是Pipe,使用起来很简单,如下:
public class Receiver extends Thread{
private Pipe pipe;
public void setPipe(Pipe pipe){
this.pipe = pipe;
}
@Override
public void run() {
super.run();
try (Pipe.SourceChannel channel = pipe.source()){
ByteBuffer buf = ByteBuffer.allocate(1024);
int len = 0;
StringBuilder sb = new StringBuilder();
while((len = channel.read(buf))!=-1){
sb.append(new String(buf.array(),0,len));
buf.clear();
}
System.out.println(sb.toString());
} catch (IOException e) {
e.printStackTrace();
}
}
}
public class Sender extends Thread{
private Pipe pipe;
public void setPipe(Pipe pipe){
this.pipe = pipe;
}
@Override
public void run() {
super.run();
try (Pipe.SinkChannel channel = pipe.sink()){
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.put("hello world".getBytes());
buffer.flip();
while (buffer.hasRemaining())
channel.write(buffer);
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws IOException {
Receiver receiver = new Receiver();
Sender sender = new Sender();
Pipe pipe = Pipe.open();
receiver.setPipe(pipe);
sender.setPipe(pipe);
receiver.start();
sender.start();
}
Pipe只是一个管理者,收发数据还是通过两个通道:SinkChannel 和SourceChannel 。都是单项的,SinkChannel 负责写数据,SourceChannel 负责收数据。
网友评论