【网络编程】（五）NIO特点、实现客户端和服务端的单/双向通信

深藏阁楼爱情的钟 2022-06-06 01:16 205阅读 0赞

NIO的本质就是避免原始的TCP建立连接使用的3次握手的操作，减少网络开销。

# NIO #

new IO 或 non-block IO

jdk1.4出现

linux 多路复用技术(select模式)  实现IO事件的轮询方式

同步非阻塞的模式

# 重要的几个概念 #

## Buffer 缓冲区 ##

在NIO库中，所有数据都是用缓冲区处理的

## Channel管道 ##

网络数据通过Channel读取和写入。

分为两类：

SelectableChannel网络读写。子类：

SocketChannel

ServerSocketChannel

FileChannel文件读写。

通道和流的对比。

通道：双向的。既能读，又能写。

流：只能在一个方向上移动。

## Selector选择器、多路复用器 ##

当IO事件（Channel）注册到Selector后，Selector会给每个Channel分配一个key。

不断的轮询注册在其上的通道Channel。如果某个Channel发生了读写操作，则这个Channel就处于就绪状态，被Selector轮询出来以后，然后通过SelectionKey取得就绪的Channel集合，并进行后续操作。

一个Selector可以负责成千上万的Channel，没有上限［JDK使用epoll代替传统的select实现的］，只需要一个线程负责Selector的轮询即可。

![Center][]

图示说明：

1. Client端的SockerChannel注册到Server端的Selector上。

2. Selector轮询所有的注册通道，根据通道状态［Connect连接状态、Accept阻塞状态、Read可读状态、Write可写状态］，执行对应的操作。

# 举例一：实现客户端和服务端的单向通信。 #

Server.java

package com.asiainno.utils;
    
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    
    public class Server implements Runnable {
        //1 多路复用器（管理所有的通道）
        private Selector selector;
        //2 建立读缓冲区
        private ByteBuffer readBuf = ByteBuffer.allocate(1024);
        //3 建立写缓冲区
        private ByteBuffer writeBuf = ByteBuffer.allocate(1024);
    
        public Server(int port) {
            try {
                //1 打开多路复用器
                this.selector = Selector.open();
                //2 打开服务器通道
                ServerSocketChannel server = ServerSocketChannel.open();
                //3 设置服务器通道为非阻塞模式
                server.configureBlocking(false);
                //4 绑定地址
                server.bind(new InetSocketAddress(port));
                //5 把服务器通道注册到多路复用器上，并且监听阻塞事件
                server.register(this.selector, SelectionKey.OP_ACCEPT);
    
                System.out.println("Server start, port :" + port);
    
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    //1 必须要让多路复用器开始监听
                    this.selector.select();
                    //2 返回多路复用器已经选择的结果集
                    Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
                    //3 进行遍历
                    while (keys.hasNext()) {
                        //4 获取一个选择的元素
                        SelectionKey key = keys.next();
                        //5 直接从容器中移除就可以了
                        keys.remove();
                        //6 如果是有效的
                        if (key.isValid()) {
                            //7 如果为阻塞状态
                            if (key.isAcceptable()) {
                                this.accept(key);
                            }
                            //8 如果为可读状态
                            if (key.isReadable()) {
                                this.read(key);
                            }
                        }
    
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private void read(SelectionKey key) {
            try {
                //1 清空缓冲区旧的数据
                this.readBuf.clear();
                //2 获取之前注册的socket通道对象
                SocketChannel client = (SocketChannel) key.channel();
                //3 读取数据
                int count = client.read(this.readBuf);
                //4 如果没有数据
                if (count == -1) {
                    key.channel().close();
                    key.cancel();
                    return;
                }
                //5 有数据则进行读取 读取之前需要进行复位方法(把position 和limit进行复位)
                this.readBuf.flip();
                //6 根据缓冲区的数据长度创建相应大小的byte数组，接收缓冲区的数据
                byte[] bytes = new byte[this.readBuf.remaining()];
                //7 接收缓冲区数据
                this.readBuf.get(bytes);
                //8 打印结果
                String body = new String(bytes).trim();
                System.out.println("Server : " + body);
    
                // 9..可以写回给客户端数据
    
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
    
        private void accept(SelectionKey key) {
            try {
                //1 获取服务通道
                ServerSocketChannel server = (ServerSocketChannel) key.channel();
                //2 执行阻塞方法
                SocketChannel client = server.accept();
                //3 设置阻塞模式
                client.configureBlocking(false);
                //4 注册到多路复用器上，并设置读取标识
                client.register(this.selector, SelectionKey.OP_READ);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) {
    
            new Thread(new Server(8765)).start();
        }
    }

Client.java

package com.asiainno.utils;
    
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SocketChannel;
    
    public class Client {
    
        //需要一个Selector
        public static void main(String[] args) {
    
            //创建连接的地址
            InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8765);
    
            //声明连接通道
            SocketChannel client = null;
    
            //建立缓冲区
            ByteBuffer writeBuf = ByteBuffer.allocate(1024);
    
            try {
                //打开通道
                client = SocketChannel.open();
                //进行连接
                client.connect(address);
    
                while (true) {
                    //定义一个字节数组，然后使用系统录入功能：
                    byte[] bytes = new byte[1024];
                    System.in.read(bytes);
    
                    //把数据放到缓冲区中
                    writeBuf.put(bytes);
                    //对缓冲区进行复位
                    writeBuf.flip();
                    //写出数据
                    client.write(writeBuf);
                    //清空缓冲区数据
                    writeBuf.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (client != null) {
                    try {
                        client.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }
    
    }

# 举例二：实现客户端和服务端的双向通信。 #

1.Client端和Server端都需要自己的缓冲区byteBuffer

2.Client端和Server端都需要选择器，即都可以进行读写操作

Server2.java

package com.asiainno.utils;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    
    public class Server2 implements Runnable {
    
        private int flag = 1;
        //1 多路复用器（管理所有的通道）
        private Selector selector;
        //2 建立读缓冲区
        private ByteBuffer readBuf = ByteBuffer.allocate(1024);
        //3 建立写缓冲区
        private ByteBuffer writeBuf = ByteBuffer.allocate(1024);
    
        public Server2(int port) {
            try {
                //1 打开多路复用器
                this.selector = Selector.open();
                //2 打开服务器通道
                ServerSocketChannel server = ServerSocketChannel.open();
                //3 设置服务器通道为非阻塞模式
                server.configureBlocking(false);
                //4 绑定地址
                server.bind(new InetSocketAddress(port));
                //5 把服务器通道注册到多路复用器上，并且监听阻塞事件
                server.register(this.selector, SelectionKey.OP_ACCEPT);
    
                System.out.println("Server start, port :" + port);
    
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    //1 必须要让多路复用器开始监听
                    this.selector.select();
                    //2 返回多路复用器已经选择的结果集
                    Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
                    //3 进行遍历
                    while (keys.hasNext()) {
                        //4 获取一个选择的元素
                        SelectionKey key = keys.next();
                        //5 直接从容器中移除就可以了
                        keys.remove();
                        //6 如果是有效的
                        if (key.isValid()) {
                            //7 如果为阻塞状态
                            if (key.isAcceptable()) {
                                this.accept(key);
                            }
                            //8 如果为可读状态
                            if (key.isReadable()) {
                                this.read(key);
                            }
                            //9 写数据
                            if (key.isWritable()) {
                                this.write(key);
                            }
                        }
    
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private void write(SelectionKey key) throws IOException {
            //server.register(this.selector, SelectionKey.OP_WRITE);
            //服务端发送数据给客户端
            this.writeBuf.clear();
            SocketChannel client = (SocketChannel) key.channel();
            String sendText = "msg send to client:" + flag++;
            writeBuf.put(sendText.getBytes());
            writeBuf.flip();
            client.write(writeBuf);
            System.out.println("服务端发送数据给客户端：" + sendText);
        }
    
        private void read(SelectionKey key) throws IOException {
    
            //1 清空缓冲区旧的数据
            this.readBuf.clear();
            //2 获取之前注册的socket通道对象
            SocketChannel client = (SocketChannel) key.channel();
            //3 读取数据
            int count = client.read(this.readBuf);
            //4 如果没有数据
            if (count == -1) {
                key.channel().close();
                key.cancel();
                return;
            }
            //5 有数据则进行读取 读取之前需要进行复位方法(把position 和limit进行复位)
            this.readBuf.flip();
            String body =new String(readBuf.array(),0,count);
            System.out.println("Server : " + body);
    
            // 9..可以写回给客户端数据
            client.register(this.selector, SelectionKey.OP_WRITE);
    
    
        }
    
        private void accept(SelectionKey key) {
            try {
                //1 获取服务通道
                ServerSocketChannel server = (ServerSocketChannel) key.channel();
                //2 执行阻塞方法
                SocketChannel client = server.accept();
                //3 设置阻塞模式
                client.configureBlocking(false);
                //4 注册到多路复用器上，并设置读取标识
                client.register(this.selector, SelectionKey.OP_READ);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) {
    
            new Thread(new Server(8765)).start();
        }
    }

Client2.java

package com.asiainno.utils;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    
    public class Client2 implements Runnable {
    
        private int flag = 1;
        //1 多路复用器（管理所有的通道）
        private Selector selector;
        //2 建立读缓冲区
        private ByteBuffer readBuf = ByteBuffer.allocate(1024);
        //3 建立写缓冲区
        private ByteBuffer writeBuf = ByteBuffer.allocate(1024);
    
        //创建连接的地址：ip+端口
        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8765);
    
    
        public Client2() {
            try {
                //1 打开多路复用器
                this.selector = Selector.open();
                //2 打开服务器通道
                SocketChannel client = SocketChannel.open();
                //3 设置服务器通道为非阻塞模式
                client.configureBlocking(false);
                //5 把服务器通道注册到多路复用器上，并且监听阻塞事件
                client.register(this.selector, SelectionKey.OP_CONNECT);
                client.connect(address);
    
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    //1 必须要让多路复用器开始监听
                    this.selector.select();
                    //2 返回多路复用器已经选择的结果集
                    Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
                    //3 进行遍历
                    while (keys.hasNext()) {
                        //4 获取一个选择的元素
                        SelectionKey key = keys.next();
                        //5 直接从容器中移除就可以了
                        keys.remove();
                        //6 如果是有效的
                        if (key.isValid()) {
                            //7 如果为阻塞状态
                            if (key.isConnectable()) {
                                System.out.println("client connect");
                                this.connect(key);
                            }
                            //8 如果为可读状态
                            if (key.isReadable()) {
                                this.read(key);
                            }
                            //9 写数据
                            if (key.isWritable()) {
                                this.write(key);
                            }
                        }
    
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private void write(SelectionKey key) throws IOException {
            //客户端发送数据给服务端
            this.writeBuf.clear();
            SocketChannel client = (SocketChannel) key.channel();
            String sendText = "msg from to client:" + flag++;
            writeBuf.put(sendText.getBytes());
            writeBuf.flip();
            client.write(writeBuf);
            System.out.println("客户端发送数据给服务端：" + sendText);
            client.register(selector, SelectionKey.OP_READ);
        }
    
        private void read(SelectionKey key) throws IOException {
            //1 清空缓冲区旧的数据
            this.readBuf.clear();
    
            SocketChannel client = (SocketChannel) key.channel();
    
            int count = client.read(readBuf);
            if (count > 0) {
                String receiveTest = new String(readBuf.array(), 0, count);
                System.out.println("客户端接收到服务端的数据:" + receiveTest);
                client.register(selector, SelectionKey.OP_WRITE);
    
            }
    
        }
    
        private void connect(SelectionKey key) {
            try {
                //1 获取服务通道
                SocketChannel client = (SocketChannel) key.channel();
                if (client.isConnectionPending()) {
                    client.finishConnect();
                    System.out.println("客户端完成连接操作！");
                    writeBuf.clear();
                    writeBuf.put("Hello,Server".getBytes());
                    writeBuf.flip();
                    client.write(writeBuf);
                }
                //4 注册到多路复用器上，并设置读取标识
                client.register(selector, SelectionKey.OP_READ);
    
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) {
            new Thread(new Client2()).start();
        }
    }

总结：

1.本文主要介绍NIO的特点，其中几个重要的概念：Buffer、Channel、Selector。

2.并代码举例说明：客户端和服务端的单向通信，双向通信。

3.NIO本质就是避免原始的TCP建立连接使用的3次握手的操作，减少网络开销。

[Center]: /images/20220606/a7132513685b48369e813c37b28bd0a4.png