浅尝JavaNIO

时间：2019-11-14 22:21:20 阅读：75 评论：0 收藏：0 [点我收藏+]

标签：inter als ddr host client rac truncate 成员 ntb

本文将通过JavaNIO实现文件的局域网发送功能。

目的：

1.为了体验零拷贝

2.为了体验NIO的Channel，Buffer，Selector

3.为了体验NIO方式网络传输文件和传统网络方式传输文件的差异（性能差异）。

什么是零拷贝：

　　那先来说说传统的文件传输：在本地读取文件（FileInputStream）通过网络传输(socket.getInputStream，socket.getOutputstream)发送到另一台设备进行存储（FileOutPutStream）过程，以及在操作这些流所用到的缓存（自定义byte数组或者各种BufferdInputStream/BufferdOutputStream）中，这些操作都在我们浅显来看，是这样的：磁盘——网络——磁盘的过程；稍微深一层次的，应该是这样的：磁盘—内存—网络—内存—磁盘的过程；再深一层次的，应该是这样：磁盘—操作系统内存（内核态-从磁盘读取过来）—应用程序内存（用户态）—操作系统内存（内核态-网络准备发送）—网络中传输—操作系统内存（内核态-网络接收）—应用程序内存（用户态）—操作系统内存（内核态-准备存磁盘）—磁盘。

　　重点来了，零拷贝（需要操作系统支持）就是用来减少甚至杜绝操作系统内存（内核态）到应用程序内存（用户态）的拷贝过程的。让cpu不要浪费在这种内存间拷贝的操作上，而是用在其他高效的计算上，零拷贝通过内存地址映射的方式，让网络/磁盘直接到系统内存中（内核态）中读取缓存的数据进行网络传输/本地存储，而不用再让数据到应用程序内存中来”转一圈“。

　　java中NIO中的的FileChannel类的transferTo、transferFrom就能够实现这个操作，直接操作内核态，不经过用户态。

接下来就是NIO了：

1.Channel：

　　我一直把Channel看成是双向流，这一点在其实现类的API中可以很明显的看出来，在使用过程中，通过对API的调用，更能感同身受。

　　最常用的实现类：

　　1.FileChannel，常用API如下（先列在这，后期再来逐一梳理）：

 1 public abstract class FileChannel extends AbstractInterruptibleChannel
 2     implements SeekableByteChannel, GatheringByteChannel, ScatteringByteChannel
 3 {
 4     //打开或者创建（无则创建）一个文件，成功后返回该文件的Channel对象
 5     public static FileChannel open(Path path, OpenOption... options)
 6 
 7     public static FileChannel open(Path path,Set<? extends OpenOption> options, FileAttribute<?>... attrs)
 8     //从文件读
 9     public abstract int read(ByteBuffer dst) throws IOException;
10     public abstract long read(ByteBuffer[] dsts, int offset, int length)throws IOException;
11     public final long read(ByteBuffer[] dsts) throws IOException
12     //写文件
13     public abstract int write(ByteBuffer src) throws IOException;
14     public abstract long write(ByteBuffer[] srcs, int offset, int length)
15     public final long write(ByteBuffer[] srcs) throws IOException
16     //截取指定大小的文件
17     public abstract FileChannel truncate(long size) throws IOException;
18     //强制将所有对此通道的文件更新写入包含该文件的存储设备中
19     public abstract void force(boolean metaData) throws IOException;
20     //将字节从此通道的文件传输到给定的可写入字节通道(支持零拷贝)
21     public abstract long transferTo(long position, long count,WritableByteChannel target) throws IOException
22     //将字节从给定的可读取字节通道传输到此通道的文件中(支持零拷贝)
23     public abstract long transferFrom(ReadableByteChannel src, long position,long count) throws IOException
24     //将指定的文件映射到内存中
25     public abstract MappedByteBuffer map(MapMode mode, long position, long size) throws IOException;

　　2.SocketChannel

　　2.1 SocketChannel,常用API如下：

 1 public abstract class SocketChannel
 2     extends AbstractSelectableChannel
 3     implements ByteChannel, ScatteringByteChannel, GatheringByteChannel, NetworkChannel
 4 {
 5     //打开套接字通道并将其连接到远程地址。
 6     public static SocketChannel open(SocketAddress remote) throws IOException
 7     //开启连接，可以使阻塞和非阻塞，在普通socket中，此处是阻塞的
 8     public abstract boolean connect(SocketAddress remote) throws IOException
 9     //完成套接字通道的连接过程。
10     public abstract boolean finishConnect() throws IOException
11     //将字节序列从此通道中读入给定的缓冲区。
12     public abstract int read(ByteBuffer dst)throws IOException   
13     public abstract long read(ByteBuffer[] dsts,int offset,int length) throws IOException
14     public final long read(ByteBuffer[] dsts) throws IOException
15     //将字节序列从给定的缓冲区中写入此通道。 
16     public abstract int write(ByteBuffer src) throws IOException
17     public abstract long write(ByteBuffer[] srcs,int offset, int length) throws IOException
18     public final long write(ByteBuffer[] srcs) throws IOException
19 }

　　2.2SocketServerChannel,常用API如下：

 1 public abstract class ServerSocketChannel extends AbstractSelectableChannel  implements NetworkChannel
 2 {
 3     //打开服务器套接字通道，开启网络服务器第一步
 4     public static ServerSocketChannel open() throws IOException
 5     //接受到此通道套接字的连接
 6     public abstract SocketChannel accept() throws IOException
 7     //继承自java.nio.channels.spi.AbstractSelectableChannel，设置阻塞模式（阻塞或者非阻塞）
 8     public final SelectableChannel configureBlocking(boolean block) throws IOException
 9     //继承自java.nio.channels.SelectableChannel，实现选择网络模型
10     register(Selector sel, int ops) throws ClosedChannelException
11 }

2.Buffer

　　Buffer是一个虚类，他的子类有ByteBuffer,ShortBuffer,IntBuffer,LongBuffer,FloatBuffer,DoubleBuffer,CharBuffer基本类型都有对应的Buffer，但是bollean没有，这些子类也是虚类。

　　在Buffer类中这，需要注意几个成员变量，及几个实现方法：

 1 public abstract class Buffer {
 2     //标志位，可以通过设置此变量的值，reset()方法可以将操作指针下标回滚到该位置
 3     private int mark = -1;
 4     //当前操作指针的下标
 5     private int position = 0;
 6     //当前有效数据的最大范围，也可以说是当前操作指针能操作到的最大范围，类似集合的size()方法的返回值。
 7     private int limit;
 8     //当前buffer的最大值（容量），类似于集合的capacity
 9     private int capacity;
10     
11     
12     //重绕缓冲区-读取后操作指针回到0位置—针对多线程读取的时候，position位置改变。
13     public final Buffer rewind()
14     //重置缓冲区——position回到上一个mark点，mark回到-1
15     //类似于回滚到指定位置
16     public final Buffer reset()
17     //清除缓冲区——position回到0，mark回到-1，limit回到capacity
18     //一般用在读取后，清空缓冲区（缓冲区数据还在，只是操作指针归零了）
19     public final Buffer clear()
20     //翻转缓冲区——limit来到position,positio回到0，mark回到-1
21     //一般用在read（）方法前
22     public final Buffer flip()

　　接下来上代码—客户端：

 1 public class Client {
 2     public static void main(String[] args) throws IOException, InterruptedException {
 3         //创建客户端的通道
 4         SocketChannel sc = SocketChannel.open();
 5         //设置为非阻塞
 6         sc.configureBlocking(false);
 7         Socket s;
 8         //发起连接
 9         //连接过程在BIO中会产生阻塞
10         sc.connect(new InetSocketAddress("localhost", 60253));
11         //判断连接是否建立
12         while (!sc.isConnected()) {
13             //如果没连上，试图再次连接
14             //如果连接多次依然失败，则意味着这个连接失败
15             //sc.finishConnect底层会自动计数，计数多次依然失败则抛出异常
16             sc.finishConnect();
17         }
18 
19         //发送数据
20         sc.write(ByteBuffer.wrap("hello server".getBytes()));
21 
22         //读数据
23         Thread.sleep(50);
24         FileChannel fc = new FileOutputStream(new File("D:\\FeiQ.exe")).getChannel();
25         ByteBuffer buffer = ByteBuffer.allocate(500);
26         int len = 0;
27         while ((len = sc.read(buffer)) > 0) {
28             System.out.println(len);
29             //
30             buffer.flip();
31             fc.write(buffer);
32             buffer.clear();
33             
34         }
35         //关流
36         sc.close();
37     }
38 }

　　服务器端代码

 1 public class Server {
 2     public static void main(String[] args) throws IOException {
 3         
 4         /**选择器要求通道必须是非阻塞的*/
 5         
 6         //开启服务器端的通道
 7         ServerSocketChannel ssc = ServerSocketChannel.open();        
 8         //绑定
 9         ssc.bind(new InetSocketAddress(60253));        
10         //开启选择器
11         Selector selc = Selector.open();        
12         //将服务器注册到选择器上
13         ssc.configureBlocking(false);
14         ssc.register(selc, SelectionKey.OP_ACCEPT);        
15         //
16         while(true){
17             //进行选择
18             selc.select();
19             //获取选择出来的事件
20             Set<SelectionKey> set = selc.selectedKeys();
21             //遍历集合，根据事件类型不同，进行处理
22             Iterator< SelectionKey> it = set.iterator();
23             while (it.hasNext()) {
24                 SelectionKey key = it.next();
25                 
26                 //可接受
27                 if (key.isAcceptable()) {
28                     //真正需要处理的事件，需要通道来完成
29                     //从这个事件中需要获取到需要进行accept的通道
30                     ServerSocketChannel sscx =  (ServerSocketChannel) key.channel();
31                     //接受连接
32                     SocketChannel sc = sscx.accept();
33                     //需要给这个通道注册可读或者可写事件
34                     //如果需要注册可读，也主要注册可写——在注册的时候，后注册的事件会覆盖之前的事件 -或者|
35                     //将sc设置为非阻塞
36                     sc.configureBlocking(false);
37                     sc.register(selc, SelectionKey.OP_WRITE + SelectionKey.OP_READ);
38                 }                
39                 //可读
40                 if (key.isReadable()) {
41                     //先从事件中获取通道
42                     SocketChannel sc = (SocketChannel) key.channel();
43                     System.out.println("11111");
44                     //读取数据
45                     ByteBuffer dst = ByteBuffer.allocate(1024);
46                     sc.read(dst);
47                     dst.flip();
48                     System.out.println(new String(dst.array(), 0, dst.limit()));
49                     
50                     //注销read事件 +或者^
51                     sc.register(selc, key.interestOps() ^ SelectionKey.OP_READ);
52                 }                
53                 //可写
54                 if (key.isWritable()) {
55                     //先从事件中获取通道
56                     SocketChannel sc = (SocketChannel) key.channel();
57                     
58                     //写入事件
59                     //sc.write(ByteBuffer.wrap("hello client".getBytes()));
60                     FileChannel fc = FileTrans.getFileChannel();
61                     
62                     long len = fc.size();
63                     int postion = 0;
64                     while (postion < len) {
65                         long readlen = fc.transferTo(postion, fc.size(), sc);
66                         postion += readlen;
67                         
68                     }
69                     System.out.println(fc.size());
70                     
71                     //注销掉writer
72                     sc.register(selc, key.interestOps() ^ SelectionKey.OP_WRITE);
73                 }
74                 it.remove();
75             }
76         }
77     }
78 }
79 
80 class FileTrans{
81     public static FileChannel getFileChannel() throws FileNotFoundException{
82         FileInputStream fos = new FileInputStream(new File("E:\\FeiQ.exe"));
83         return fos.getChannel();    
84     }
85 }