深入浅出使用Java NIO

悠悠 2022-03-17 13:28 186阅读 0赞

> java NIO的使用越来重要,找个时间学习了一下进行一下Test

package com.test2;
    
    import org.junit.Test;
    
    import java.io.*;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.CharBuffer;
    import java.nio.MappedByteBuffer;
    import java.nio.channels.*;
    import java.nio.charset.Charset;
    import java.nio.charset.CharsetDecoder;
    import java.nio.charset.CharsetEncoder;
    import java.nio.file.Paths;
    import java.nio.file.StandardOpenOption;
    import java.util.Date;
    import java.util.Iterator;
    import java.util.Scanner;
    
    /** * @program: test01 * @description: * @author: houyu * @create: 2019-02-24 16:17 */
    public class TestNIO { 
    
        /** * 一、缓冲区（Buffer）：在 Java NIO 中负责数据的存取。缓冲区就是数组。用于存储不同数据类型的数据 * * 根据数据类型不同（boolean 除外），提供了相应类型的缓冲区： * ByteBuffer * CharBuffer * ShortBuffer * IntBuffer * LongBuffer * FloatBuffer * DoubleBuffer * * 上述缓冲区的管理方式几乎一致，通过 allocate() 获取缓冲区 * * 二、缓冲区存取数据的两个核心方法： * put() : 存入数据到缓冲区中 * get() : 获取缓冲区中的数据 * * 三、缓冲区中的四个核心属性： * capacity : 容量，表示缓冲区中最大存储数据的容量。一旦声明不能改变。 * limit : 界限，表示缓冲区中可以操作数据的大小。（limit 后数据不能进行读写） * position : 位置，表示缓冲区中正在操作数据的位置。 * * mark : 标记，表示记录当前 position 的位置。可以通过 reset() 恢复到 mark 的位置 * * 0 <= mark <= position <= limit <= capacity * * 四、直接缓冲区与非直接缓冲区： * 非直接缓冲区：通过 allocate() 方法分配缓冲区，将缓冲区建立在 JVM 的内存中 * 直接缓冲区：通过 allocateDirect() 方法分配直接缓冲区，将缓冲区建立在物理内存中。可以提高效率 */
    
        @Test
        public void testByteBuffer() throws UnsupportedEncodingException { 
    
            // 1.0 新建 非直接缓冲区,并指定大小
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            // 2.0 往缓冲区put数据
            byte[] src = "abcde".getBytes("UTF-8");
            byteBuffer.put(src);
            // 3.0 缓冲区get数据
            // 3.1 切换为读取模式,实质的意思就是byteBuffer.position()修改为0,
            byteBuffer.flip();
            // 3.2 新建 byte[],数据存储到byte[]
            byte[] dst = new byte[byteBuffer.limit()];// limit():实质缓冲区可读大小
            byteBuffer.get(dst);
            // 4.0 输出数据
            System.out.println(new String(dst, "UTF-8"));
            // 5.0 清空缓冲区. 但是缓冲区中的数据依然存在，但是处于“被遗忘”状态
            byteBuffer.clear();
    
            /** * 补充: * byteBuffer底层实质还是一个byte[] 查看源码得知final byte[] hb; * 读取之前一定需要flip():目的是使得Buffer的属性position修改为0 * rewind()和flip()类似,都是position修改为0可以读取 * * 创建直接缓冲区(调用的是操作系统的内存) * ByteBuffer directByteBuffer = ByteBuffer.allocateDirect(1024); * * 以下两行代码可以优化 * byte[] dst = new byte[byteBuffer.limit()]; * byteBuffer.get(dst); * * byteBuffer.array():底层是返回byte[] hb; */
        }
    
        /** * 一、通道（Channel）：用于源节点与目标节点的连接。在 Java NIO 中负责缓冲区中数据的传输。Channel 本身不存储数据，因此需要配合缓冲区进行传输。 * * 二、通道的主要实现类 * java.nio.channels.Channel 接口： * |--FileChannel * |--SocketChannel * |--ServerSocketChannel * |--DatagramChannel * * 三、获取通道 * 1. Java 针对支持通道的类提供了 getChannel() 方法 * 本地 IO： * FileInputStream/FileOutputStream * RandomAccessFile * * 网络IO： * Socket * ServerSocket * DatagramSocket * * 2. 在 JDK 1.7 中的 NIO.2 针对各个通道提供了静态方法 open() * 3. 在 JDK 1.7 中的 NIO.2 的 Files 工具类的 newByteChannel() * * 四、通道之间的数据传输 * transferFrom() * transferTo() * * 五、分散(Scatter)与聚集(Gather) * 分散读取（Scattering Reads）：将通道中的数据分散到多个缓冲区中 * 聚集写入（Gathering Writes）：将多个缓冲区中的数据聚集到通道中 * * 六、字符集：Charset * 编码：字符串 -> 字节数组 * 解码：字节数组 -> 字符串 * */
        @Test
        public void testChannelByFileInputStream() throws IOException { 
    
            // 1.0 获取文件输入流
            FileInputStream inputStream = new FileInputStream("D:/1.txt");
            // 2.0 文件输入流获取通道
            FileChannel fileChannel = inputStream.getChannel();
            // 3.0 读取文件信息
            // 3.1 创建非直接缓冲区
            ByteBuffer dst = ByteBuffer.allocate(1024);
            while (fileChannel.read(dst) != -1){ // 管道循环读取文件到缓冲区
                dst.flip();// 读取文件必须要切换为读取模式
                String tempString = new String(dst.array(), "GBK");
                System.out.println(tempString);// window默认编码为GBK,否则乱码
            }
            // 4.0 关闭资源
            fileChannel.close();
            inputStream.close();
    
        }
    
        /** * 通过缓冲区复制文件 */
        @Test
        public void testChannelTOCopyFile_1() throws IOException { 
    
            // 1.0 获取流
            FileInputStream inputStream = new FileInputStream("D:/1.txt");
            FileOutputStream outputStream = new FileOutputStream("D:/2.txt");
            // 2.0 流获取通道
            FileChannel inChannel = inputStream.getChannel();
            FileChannel outChannel = outputStream.getChannel();
            // 3.0 创建非直接缓冲区
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            // 4.0 inChannel读取文件到缓冲区并写入到outChannel
            while (inChannel.read(byteBuffer) != -1){ 
                byteBuffer.flip();
                outChannel.write(byteBuffer);
                byteBuffer.clear();
            }
            // 5.0 关闭资源
            outChannel.close();
            inputStream.close();
            outChannel.close();
            inputStream.close();
            /** * 补充: * 这种复制文件使用的非直接缓冲区,速度方面比不上使用直接缓冲区 */
        }
    
        /** * 通过内存映射文件复制文件 */
        @Test
        public void testChannelTOCopyFile_2() throws IOException { 
            // 使用FileChannel静态方法open()获取通道
            FileChannel inChannel = FileChannel.open(Paths.get("D:/1.pdf"), StandardOpenOption.READ);
            FileChannel outChannel = FileChannel.open(Paths.get("D:/2.pdf"), StandardOpenOption.WRITE, StandardOpenOption.READ, StandardOpenOption.CREATE);
            // 获取内存映射文件
            MappedByteBuffer inMappedBuffer = inChannel.map(FileChannel.MapMode.READ_ONLY, 0, inChannel.size());
            MappedByteBuffer outMappedBuffer = outChannel.map(FileChannel.MapMode.READ_WRITE, 0, inChannel.size());
            // 直接对缓冲区进行数据的读写操作
            byte[] dst = new byte[inMappedBuffer.limit()];
            inMappedBuffer.get(dst);
            outMappedBuffer.put(dst);
            // 关闭资源
            outMappedBuffer.clear();
            inMappedBuffer.clear();
            outChannel.close();
            inChannel.close();
    
        }
    
        /** * 通过管道对管道复制文件 */
        @Test
        public void testChannelTOCopyFile_3() throws IOException { 
            // 1.0 使用FileChannel静态方法open()获取通道
            FileChannel inChannel = FileChannel.open(Paths.get("D:/1.pdf"), StandardOpenOption.READ);
            FileChannel outChannel = FileChannel.open(Paths.get("D:/2.pdf"), StandardOpenOption.WRITE, StandardOpenOption.READ, StandardOpenOption.CREATE);
            // 2.0 管道对接管道
            inChannel.transferTo(0, inChannel.size(), outChannel);
            // outChannel.transferFrom(inChannel, 0, inChannel.size());
            // 3.0 关闭资源
            inChannel.close();
            outChannel.close();
        }
    
        /** * 分散和聚集 */
        @Test
        public void testScatterAndGather() throws IOException { 
            // 1.0 创建随机访问文件对象
            RandomAccessFile randomAccessFile = new RandomAccessFile("D:/1.txt", "rw");
            // 2.0 获取管道
            FileChannel inChannel = randomAccessFile.getChannel();
            // 3.0 创建指定大小的缓冲区
            ByteBuffer buf1 = ByteBuffer.allocate(50);
            ByteBuffer buf2 = ByteBuffer.allocate(1024);
            ByteBuffer[] bufs = { buf1, buf2};
            inChannel.read(bufs);
            // 4.0 切换读模式
            buf1.flip();
            buf2.flip();
            // 5.0 输出查看效果
            System.out.println(new String(buf1.array()));
            System.out.println("-----------------");
            System.out.println(new String(buf2.array()));
            // 6.0 创建输出随机访问对象
            RandomAccessFile outFile = new RandomAccessFile("D:/2.txt", "rw");
            // 7.0 获取管道
            FileChannel outChannel = outFile.getChannel();
            outChannel.write(bufs);
    
            // 8.0 关闭资源
            outChannel.close();
            outFile.close();
            buf2.clear();
            buf1.clear();
            inChannel.close();
            randomAccessFile.close();
    
        }
    
        //字符集
        @Test
        public void testCharset() throws IOException{ 
            Charset cs1 = Charset.forName("GBK");
            //获取编码器
            CharsetEncoder ce = cs1.newEncoder();
            //获取解码器
            CharsetDecoder cd = cs1.newDecoder();
            CharBuffer cBuf = CharBuffer.allocate(1024);
            cBuf.put("我是数据");
            cBuf.flip();
            //编码
            ByteBuffer bBuf = ce.encode(cBuf);
            for (int i = 0; i < 8; i++) { 
                System.out.println(bBuf.get());
            }
            //解码
            bBuf.flip();
            CharBuffer cBuf2 = cd.decode(bBuf);
            System.out.println(cBuf2.toString());
        }
    
        /* * 一、使用 NIO 完成网络通信的三个核心： * * 1. 通道（Channel）：负责连接 * * java.nio.channels.Channel 接口： * |--SelectableChannel * |--SocketChannel * |--ServerSocketChannel * |--DatagramChannel * * |--Pipe.SinkChannel * |--Pipe.SourceChannel * * 2. 缓冲区（Buffer）：负责数据的存取 * * 3. 选择器（Selector）：是 SelectableChannel 的多路复用器。用于监控 SelectableChannel 的 IO 状况 * */
        //客户端
        @Test
        public void testBlockingNIOClient_1() throws IOException{ 
            //1. 获取通道
            SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9898));
            FileChannel inChannel = FileChannel.open(Paths.get("D:/1.txt"), StandardOpenOption.READ);
            //2. 分配指定大小的缓冲区
            ByteBuffer buf = ByteBuffer.allocate(1024);
            //3. 读取本地文件，并发送到服务端
            while(inChannel.read(buf) != -1){ 
                buf.flip();
                sChannel.write(buf);
                buf.clear();
            }
            //4. 关闭通道
            inChannel.close();
            sChannel.close();
        }
    
        //服务端
        @Test
        public void testBlockingNIOServer_1() throws IOException{ 
            //1. 获取通道
            ServerSocketChannel ssChannel = ServerSocketChannel.open();
            FileChannel outChannel = FileChannel.open(Paths.get("D:/2.txt"), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
            //2. 绑定连接
            ssChannel.bind(new InetSocketAddress(9898));
            //3. 获取客户端连接的通道
            SocketChannel sChannel = ssChannel.accept();
            //4. 分配指定大小的缓冲区
            ByteBuffer buf = ByteBuffer.allocate(1024);
            //5. 接收客户端的数据，并保存到本地
            while(sChannel.read(buf) != -1){ 
                buf.flip();
                outChannel.write(buf);
                buf.clear();
            }
            //6. 关闭通道
            sChannel.close();
            outChannel.close();
            ssChannel.close();
        }
    
        //客户端
        @Test
        public void testBlockingNIOClient_2() throws IOException{ 
            SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9898));
            FileChannel inChannel = FileChannel.open(Paths.get("1.jpg"), StandardOpenOption.READ);
            ByteBuffer buf = ByteBuffer.allocate(1024);
            while(inChannel.read(buf) != -1){ 
                buf.flip();
                sChannel.write(buf);
                buf.clear();
            }
            sChannel.shutdownOutput();
            //接收服务端的反馈
            int len;
            while((len = sChannel.read(buf)) != -1){ 
                buf.flip();
                System.out.println(new String(buf.array(), 0, len));
                buf.clear();
            }
    
            inChannel.close();
            sChannel.close();
        }
    
        //服务端
        @Test
        public void testBlockingNIOServer_2() throws IOException{ 
            ServerSocketChannel ssChannel = ServerSocketChannel.open();
            ssChannel.bind(new InetSocketAddress(9898));
            FileChannel outChannel = FileChannel.open(Paths.get("2.jpg"), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
            SocketChannel sChannel = ssChannel.accept();
            ByteBuffer buf = ByteBuffer.allocate(1024);
            while(sChannel.read(buf) != -1){ 
                buf.flip();
                outChannel.write(buf);
                buf.clear();
            }
            //发送反馈给客户端
            buf.put("服务端接收数据成功".getBytes());
            buf.flip();
            sChannel.write(buf);
            sChannel.close();
            outChannel.close();
            ssChannel.close();
        }
    
    
        //客户端
        @Test
        public void testNonBlockingNIOClient_1() throws IOException{ 
            //1. 获取通道
            SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9898));
            //2. 切换非阻塞模式
            sChannel.configureBlocking(false);
            //3. 分配指定大小的缓冲区
            ByteBuffer buf = ByteBuffer.allocate(1024);
            //4. 发送数据给服务端
            Scanner scan = new Scanner(System.in);
            while(scan.hasNext()){ 
                String str = scan.next();
                buf.put((new Date().toString() + "\n" + str).getBytes());
                buf.flip();
                sChannel.write(buf);
                buf.clear();
            }
            //5. 关闭通道
            sChannel.close();
        }
    
        //服务端
        @Test
        public void testNonBlockingNIOServer_1() throws IOException{ 
            //1. 获取通道
            ServerSocketChannel ssChannel = ServerSocketChannel.open();
            //2. 切换非阻塞模式
            ssChannel.configureBlocking(false);
            //3. 绑定连接
            ssChannel.bind(new InetSocketAddress(9898));
            //4. 获取选择器
            Selector selector = Selector.open();
            //5. 将通道注册到选择器上, 并且指定“监听接收事件”
            ssChannel.register(selector, SelectionKey.OP_ACCEPT);
            //6. 轮询式的获取选择器上已经“准备就绪”的事件
            while(selector.select() > 0){ // 这个方法是阻塞的
                //7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while(it.hasNext()){ 
                    //8. 获取准备“就绪”的是事件
                    SelectionKey sk = it.next();
                    //9. 判断具体是什么事件准备就绪
                    if(sk.isAcceptable()){ 
                        //10. 若“接收就绪”，获取客户端连接
                        SocketChannel sChannel = ssChannel.accept();
                        //11. 切换非阻塞模式
                        sChannel.configureBlocking(false);
                        //12. 将该通道注册到选择器上
                        sChannel.register(selector, SelectionKey.OP_READ);
                    }else if(sk.isReadable()){ 
                        //13. 获取当前选择器上“读就绪”状态的通道
                        SocketChannel sChannel = (SocketChannel) sk.channel();
                        //14. 读取数据
                        ByteBuffer buf = ByteBuffer.allocate(1024);
                        int len = 0;
                        while((len = sChannel.read(buf)) > 0 ){ 
                            buf.flip();
                            System.out.println(new String(buf.array(), 0, len));
                            buf.clear();
                        }
                    }
                    //15. 取消选择键 SelectionKey
                    it.remove();
                }
            }
        }
    
    
    
        @Test
        public void testNonBlockingNIOSend_2() throws IOException{ 
            DatagramChannel dc = DatagramChannel.open();
            dc.configureBlocking(false);
            ByteBuffer buf = ByteBuffer.allocate(1024);
            Scanner scan = new Scanner(System.in);
            while(scan.hasNext()){ 
                String str = scan.next();
                buf.put((new Date().toString() + ":\n" + str).getBytes());
                buf.flip();
                dc.send(buf, new InetSocketAddress("127.0.0.1", 9898));
                buf.clear();
            }
            dc.close();
        }
    
        @Test
        public void testNonBlockingNIOReceive_2() throws IOException{ 
            DatagramChannel dc = DatagramChannel.open();
            dc.configureBlocking(false);
            dc.bind(new InetSocketAddress(9898));
            Selector selector = Selector.open();
            dc.register(selector, SelectionKey.OP_READ);
            while(selector.select() > 0){ 
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while(it.hasNext()){ 
                    SelectionKey sk = it.next();
                    if(sk.isReadable()){ 
                        ByteBuffer buf = ByteBuffer.allocate(1024);
                        dc.receive(buf);
                        buf.flip();
                        System.out.println(new String(buf.array(), 0, buf.limit()));
                        buf.clear();
                    }
                }
                it.remove();
            }
        }
    
    
        @Test
        public void testPipe() throws IOException{ 
            //1. 获取管道
            Pipe pipe = Pipe.open();
            //2. 将缓冲区中的数据写入管道
            ByteBuffer buf = ByteBuffer.allocate(1024);
            Pipe.SinkChannel sinkChannel = pipe.sink();
            buf.put("通过单向管道发送数据".getBytes());
            buf.flip();
            sinkChannel.write(buf);
            //3. 读取缓冲区中的数据
            Pipe.SourceChannel sourceChannel = pipe.source();
            buf.flip();
            int len = sourceChannel.read(buf);
            System.out.println(new String(buf.array(), 0, len));
            sourceChannel.close();
            sinkChannel.close();
        }
    
    
    }