NIO 代码总结

小咪咪 2022-12-23 04:59 145阅读 0赞

import cn.hutool.core.util.StrUtil;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.OutputStream;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    @Slf4j
    public class ServerDemo {
    
        public static void main(String[] args) throws Exception {
            ServerSocket serverSocket = new ServerSocket(6666);
            while(true){
                Socket accept = serverSocket.accept();
                handler(accept);
            }
        }
    
        private static void handler(Socket socket) {
            byte[] bytes = new byte[1024];
            InputStream inputStream = null;
            OutputStream outputStream = null;
            try {
                inputStream = socket.getInputStream();
                outputStream = socket.getOutputStream();
                while(true){
                    int read = inputStream.read(bytes);
                    if(read != -1){
                        String msg = new String(bytes, 0, read);
                        log.info("read msg:{}", msg);
                        if(StrUtil.isNotEmpty(msg)){
                            if(msg.equals("bye")){
                                break;
                            }
                        }
                    } else {
                        outputStream.write(bytes, 0, read);
                        outputStream.flush();
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    inputStream.close();
                    outputStream.close();
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    import java.io.*;
    import java.net.Socket;
    
    @Slf4j
    public class ClientDemo {
    
        public static void main(String[] args) {
            sendMsg();
    //        sendReceivedMsg();
        }
        private static void sendMsg() {
            String info = "你好啊！";
            Socket socket = null;
            OutputStream outputStream = null;
            InputStream inputStream = null;
            BufferedReader br = null;
            try {
    //            socket = new Socket("localhost",6666);
                socket = new Socket("localhost",7000);
                outputStream = socket.getOutputStream();
                outputStream.write(info.getBytes());
                socket.shutdownOutput();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //刷新缓冲区
                try {
                    outputStream.flush();
                    outputStream.close();
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private static void sendReceivedMsg() {
            String info = "你好啊！";
            Socket socket = null;
            OutputStream outputStream = null;
            InputStream inputStream = null;
            BufferedReader br = null;
            try {
    //            socket = new Socket("localhost",6666);
                socket = new Socket("localhost",7000);
                outputStream = socket.getOutputStream();
                outputStream.write(info.getBytes());
                socket.shutdownOutput();
                //接收服务器端的响应
                inputStream = socket.getInputStream();
                br = new BufferedReader(new InputStreamReader(inputStream));
                while ((info = br.readLine()) != null){
                    log.info("接收到了服务端的响应！" + info);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //刷新缓冲区
                try {
                    outputStream.flush();
                    outputStream.close();
                    inputStream.close();
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    /*========================================================*/
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.File;
    import java.io.FileInputStream;
    import java.io.FileOutputStream;
    import java.io.IOException;
    import java.nio.Buffer;
    import java.nio.ByteBuffer;
    import java.nio.IntBuffer;
    import java.nio.channels.FileChannel;
    
    @Slf4j
    public class MyBuffer {
    
        public static void main(String[] args) {
            IntBuffer intBuffer = IntBuffer.allocate(20);
            doLog(intBuffer, "after allocate");
            for (int i = 0; i< 5; i++)
            {
                //写入一个整数到缓冲区
                intBuffer.put(i);
            }
            doLog(intBuffer, "after put");
            intBuffer.flip();
            doLog(intBuffer, "after flip");
            for (int i = 0; i< intBuffer.limit(); i++)
            {
                int j = intBuffer.get();
                log.info("j = " + j);
            }
            intBuffer.clear();
            doLog(intBuffer, "after clear");
        }
    
        public static void fileRead(){
            try {
                File file = new File("f://github.txt");
                FileInputStream fileInputStream = new FileInputStream(file);
                FileChannel channel = fileInputStream.getChannel();
                // 创建缓冲区
                ByteBuffer byteBuffer = ByteBuffer.allocate((int) file.length());
                // channel将文件数据读取带入byteBuffer中
                channel.read(byteBuffer);
                System.out.println(new String(byteBuffer.array()));
                fileInputStream.close();
            } catch (Exception e){
    
            } finally {
    
            }
    
        }
    
        public static void fileWrite(){
            FileOutputStream fileOutputStream = null;
            FileChannel channel = null;
            String str = "hello, my name is lai";
            try {
                fileOutputStream = new FileOutputStream("f://github.txt");
                channel = fileOutputStream.getChannel();
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                byteBuffer.put(str.getBytes());
                byteBuffer.flip();
                // channel将文件数据读取写入byteBuffer中
                channel.write(byteBuffer);
            } catch (Exception e){
    
            } finally {
                try {
                    fileOutputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void doLog(Buffer buffer,String desc){
            log.info("------------{}------------------", desc);
            log.info("position=" + buffer.position());
            log.info("limit=" + buffer.limit());
            log.info("capacity=" + buffer.capacity());
        }
    }
    
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.*;
    import java.util.Arrays;
    import java.util.Iterator;
    import java.util.Set;
    
    @Slf4j
    public class NetNioServerDemo {
    
        public static void main(String[] args) {
            myServer();
        }
    
        /**
         *
         */
        public static void myServer(){
            try {
                Selector selector = Selector.open();
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                InetSocketAddress inetSocketAddress = new InetSocketAddress(6666);
                serverSocketChannel.socket().bind(inetSocketAddress);
                serverSocketChannel.configureBlocking(false);
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                while(true) {
                    if(selector.select(5 * 1000) == 0){
                        log.info("5s has no accept info");
                        continue;
                    }
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
                    while(keyIterator.hasNext()) {
                        SelectionKey selectionKey = keyIterator.next();
                        // OP_ACCEPT
                        if(selectionKey.isAcceptable()) {
                            SocketChannel acceptChannel = serverSocketChannel.accept();
                            acceptChannel.configureBlocking(false);
                            acceptChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                        }
                        if(selectionKey.isReadable()) {
                            SocketChannel channel = (SocketChannel)selectionKey.channel();
                            // get key's buffer
                            ByteBuffer byteBuffer = (ByteBuffer)selectionKey.attachment();
                            channel.read(byteBuffer);
                            log.info("read from client:{}", new String(byteBuffer.array()));
                        }
                    }
                    keyIterator.remove();
                }
            } catch (Exception e){
    
            } finally {
    
            }
        }
    
        public static void myServerBuffer(){
            try {
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                InetSocketAddress inetSocketAddress = new InetSocketAddress(7000);
                serverSocketChannel.socket().bind(inetSocketAddress);
                // bytebuffer
                ByteBuffer[] byteBuffers = new ByteBuffer[2];
                byteBuffers[0] = ByteBuffer.allocate(5);
                byteBuffers[1] = ByteBuffer.allocate(3);
                // 拿到连接
                SocketChannel socketChannel = serverSocketChannel.accept();
                int messageLength = 8;
                while(true) {
                    int byteRead = 0;
                    while(byteRead < messageLength) {
                        long readBytes = socketChannel.read(byteBuffers);
                        byteRead += readBytes;
                        Arrays.asList(byteBuffers).stream()
                                .map(buffer -> "position=" + buffer.position() + ", limit=" + buffer.limit())
                                .forEach(s -> log.info("buffer:{}", s));
                    }
                    for (ByteBuffer byteBuffer : Arrays.asList(byteBuffers)) {
                        byteBuffer.flip();
                    }
                    long byteWrite = 0;
                    while(byteWrite < messageLength) {
                        long write = socketChannel.write(byteBuffers);
                        byteWrite += write;
                    }
                    for (ByteBuffer byteBuffer : Arrays.asList(byteBuffers)) {
                        byteBuffer.clear();
                    }
                }
            } catch (Exception e){
    
            }
        }
    }
    
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SocketChannel;
    
    @Slf4j
    public class NetNioClientDemo {
    
        public static void main(String[] args) {
            myClient();
        }
    
        private static void myClient() {
            try {
                SocketChannel socketChannel = SocketChannel.open();
                socketChannel.configureBlocking(false);
                InetSocketAddress inetSocketAddress = new InetSocketAddress("localhost", 6666);
                if(!socketChannel.connect(inetSocketAddress)){
                    while(!socketChannel.finishConnect()) {
                        log.info("connecting ...");
                    }
                }
                String str = "hello, master";
                ByteBuffer buffer = ByteBuffer.wrap(str.getBytes());
                MyBuffer.doLog(buffer,"wrap after");
                socketChannel.write(buffer);
            } catch (Exception e){
            }
        }
    }
    
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.io.IOUtils;
    
    import java.io.File;
    import java.io.FileInputStream;
    import java.io.FileOutputStream;
    import java.nio.ByteBuffer;
    import java.nio.channels.FileChannel;
    
    @Slf4j
    public class FileNIOCopyDemo {
        public static final String srcPath = "f://github.txt";
        public static final String destPath = "f://2.txt";
        public static void main(String[] args) {
            transfor();
        }
    
        public static void baseFileCopy(){
            File srcFile = new File(srcPath);
            File destFile = new File(destPath);
            try {
                //如果目标文件不存在，则新建
                if (!destFile.exists()) {
                    destFile.createNewFile();
                }
                long startTime = System.currentTimeMillis();
                FileInputStream fis = null;
                FileOutputStream fos = null;
                FileChannel inChannel = null;
                FileChannel outchannel = null;
                try {
                    fis = new FileInputStream(srcFile);
                    fos = new FileOutputStream(destFile);
                    inChannel = fis.getChannel();
                    outchannel = fos.getChannel();
                    int length = -1;
                    ByteBuffer buf = ByteBuffer.allocate(1024);
                    //从输入通道读取到buf
                    while ((length = inChannel.read(buf)) != -1) {
                        //第一次切换：翻转buf，变成读取模式
                        buf.flip();
                        int outlength = 0;
                        //将buf写入到输出的通道
                        while ((outlength = outchannel.write(buf)) != 0) {
                            System.out.println("写入的字节数：" + outlength);
                        }
                        //第二次切换：清除buf，变成写入模式
                        buf.clear();
                    }
                    //强制刷新到磁盘
                    outchannel.force(true);
                } finally {
                    //关闭所有的可关闭对象
                    IOUtils.closeQuietly(outchannel);
                    IOUtils.closeQuietly(fos);
                    IOUtils.closeQuietly(inChannel);
                    IOUtils.closeQuietly(fis);
                }
                long endTime = System.currentTimeMillis();
                log.info("base复制毫秒数：" + (endTime - startTime));
            } catch (Exception e){
            }
        }
    
        public static void transfor(){
            try {
                FileInputStream fileInputStream = new FileInputStream(srcPath);
                FileOutputStream fileOutputStream = new FileOutputStream(destPath);
                FileChannel inputStreamChannel = fileInputStream.getChannel();
                FileChannel outputStreamChannel = fileOutputStream.getChannel();
                outputStreamChannel.transferFrom(inputStreamChannel, 0 ,inputStreamChannel.size());
                outputStreamChannel.close();
                inputStreamChannel.close();
                fileInputStream.close();
                outputStreamChannel.close();
            } catch (Exception e){
    
            }
        }
    }

本文参考了：

(1)韩顺平老师Netty教程

(2)《Netty、redis、Zookeeper高并发实战》部分代码