网络I/O模型--05多路复用I/O

Posted 2020-11-04 ParamousGIS

 

        多路复用I/O模型在应用层工作效率比我们俗称的 BIO 模型快的本质原因是，前者不再使用操作系统级别的“同步 I／O”模型 。 在 Linux 操作系统环境下， 多路复用 I/O 模型就是技术人员通常简称的 NIO 技术。多路复用I/O 目前具体的实现主要包括四种： select、 poll , epoll 、kqueue 。

        多路复用 I/O技术最适用的是“高并发”场景，所谓高并发是指 l 毫秒内至少同时有成百上千个连接请求准备就绪，其他情况下多路复用I/O技术发挥不出它的明显优势 。

 

 

 

重要概念

    无论使用哪种实现方式，它们都会有“选择器”、 “通道”、 “缓存”这几个操作要素，那么可以为不同的多路复用盯0 技术创建一个统一的抽象组，井且为不同的操作系 统进行具体的实现。

 

Channel 通道: 是一个用来完成应用程序和操作系统交互事件、传递内容的渠道，注意是连接到操作系统。一个通道会有一个专属的文件状态描述符。既然是和操作系统进行内容的传递，那就说明应用程序可以通过通道从操作系统读取数据，也可以通过通道向操作系统写数据 。

 

Buffer数据缓存区： 在 Java 原生 NIO 框架中，为了保证每个通道的数据读／写速度， Java NIO 框架为每一种需要支持数据读／写 的通道集成了 Buffer 的支持。Buffer 有两种工作模式： 写模式和读模式。在读模式下，应用程序只能从 Buffer 中 读取数据，不能进行写操作 。 但是在写模式下，应用程序是可以进行读操作的，这就表示可能会出现脏读的情况 。 所以一旦决定要从 Buffer 中读取数据，就一定要将 Buffer 的状态改为读模式 。
• position ： 缓存区目前正在操作的数据块位置。
• limit：缓存区最大可以进行操作的位置。缓存区的读／写状态正是由这个属性控制的 。
• capacity：缓存区的最大容量 。这个容量是在缓存区创建时进行指定的。由于高井发时通道数量往往会很庞大，所以每一个缓存区的容量最好不要过大。

 

Selector选择器： 可以把它称为“轮询代理器”、 “事件订阅器”、 “ Channel 容器管理机”等等。
• 事件订阅和 Channel 管理：应用程序将向 Selector 对象注册需要它关注的 Channel ，以及具体的某一个 Channel 会对哪些 I/O事件感兴趣。 Selector 中也会维护一个 “己经注册的 Channel“的容器 。 
• 轮询代理：应用层不再通过阻塞模式或者非阻塞模式直接询问操作系统“事件有没有发生”，而是由 Selector 代其询问。
• 实现不同操作系统的支持： 之前己经提到过， 多路复用I/O 技术是需要操作系统进行支持的，其特点就是操作系统可以同时扫描同一个端口上的多个网络连接 。 所以作为上层 的 JVM，必须要为不同操作系统的多路复用 I/O 实现编写不同 的代码 。

 

package testBlockSocket;

import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SocketServerReuseAddres {
    private final static Logger LOGGER = LoggerFactory.getLogger(SocketServerReuseAddres.class);
    private static final Map<Integer, StringBuffer> MESSAGEHASHCONTEXT = new ConcurrentHashMap<>();

    public static void main(String[] args) throws Exception {
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        // 一定要这样设置
        serverChannel.configureBlocking(false);
        ServerSocket serverSocket = serverChannel.socket();
        serverSocket.setReuseAddress(true);
        serverSocket.bind(new InetSocketAddress(8888));
        Selector selector = Selector.open();
        // 注意：服务器通道能且只能注册 SelectionKey . OP ACCEPT 事件
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        try {
            while (true) {
                // 如果条件成立，则说明本次询问 selector 并没有获取任何准备好的、感兴趣的事件
                if (selector.select(100) == 0) {
                    // =============
                    // 这里祝业务情况，可以做一些其他业务动作，但实际意义不大
                    // =============
                    continue;
                }
                // 这就是本次询问操作系统所获取的“所关心的事件"的事件类型(每一个通道是独立的）
                Iterator<SelectionKey> selecionKeys = selector.selectedKeys().iterator();
                while (selecionKeys.hasNext()) {
                    SelectionKey readyKey = selecionKeys.next();
                    // 一定要移除这个已经处理的 readyKey 如果不移除，就会一直存在在 selector.selectedKeys 集合中 待到下一次 selector
                    // . select() > 0 时，这个 readyKey 又会被处理一次
                    selecionKeys.remove();
                    SelectableChannel selectableChannel = readyKey.channel();
                    if (readyKey.isValid() && readyKey.isAcceptable()) {
                        SocketServerReuseAddres.LOGGER.info("＝＝channel 通道已经准备好＝＝＝");
                        // 当 serverSocketChannel 通道已经准备好 ， 就可以从 serverSocketChannel中获取socketChannel 了
                        // 拿到socketChannel后，要做的事情就是马上到
                        // selector注册这个socketChannel感兴趣的事情。否则无法监听到这个socketChannel到达的数据
                        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectableChannel;
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        registerSocketChannel(socketChannel, selector);
                    } else if (readyKey.isValid() && readyKey.isReadable()) {
                        SocketServerReuseAddres.LOGGER.info("======socket channel数据准备完成，可以去读＝＝读取＝＝＝＝＝＝＝ ");
                        readSocketChannel(readyKey);
                    }
                }
            }
        } catch (Exception e) {
            SocketServerReuseAddres.LOGGER.error(e.getMessage(), e);
        } finally {
            serverSocket.close();
        }

    }

    private static void registerSocketChannel(SocketChannel socketChannel, Selector selector) throws Exception {
        socketChannel.configureBlocking(false);
        // socket 通道可以且只可以注册三种事件！ SelectionKey.OP_READ | SelectionKey.OP_WRITE |
        // SelectionKey.OP_CONNECT
        // 最后一个参数视为这个 socketchanne 分配的缓存区
        socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(2048));
    }

    private static void readSocketChannel(SelectionKey readyKey) throws Exception {
        SocketChannel clientSocketChannel = (SocketChannel) readyKey.channel();
        // 获取客户端使用的端口
        InetSocketAddress sourceSocketAddress = (InetSocketAddress) clientSocketChannel.getRemoteAddress();
        Integer resoucePort = sourceSocketAddress.getPort();
        // 拿到这个 socket 通道使用的缓存区，准备读取数据
        // 解缓存区的用法实际上重要的就是三个元素： capacity 、 position 和 limit
        ByteBuffer contextBytes = (ByteBuffer) readyKey.attachment();
        // 将通道的数据写入缓存区
        // 由于之前设置了 ByteBuffer 的大小为 2048 byte ，所以可能存在写不完的情况
        int realLen = -1;
        StringBuffer message = new StringBuffer();
        while ((realLen = clientSocketChannel.read(contextBytes)) != 0) {
            // 一定要把 Buffer 切换成“读”模式 ， 否则 由于 limit = capacity; 在 read 没有写满的情况下 ， 就会导致多读
            contextBytes.flip();
            int position = contextBytes.position();
            int capacity = contextBytes.capacity();
            byte[] messageBytes = new byte[capacity];
            contextBytes.get(messageBytes, position, realLen);
            // 注意中文乱码的问题，使用 URLDecoder/URLEncoder 进行解编码,当然 Java NIO 框架本身也提供编解码方式，看个人使用习惯
            String messageEncode = new String(messageBytes, 0, realLen, "UTF-8");
            message.append(messageEncode);
            // 再切换成“写”模式，直接存入缓存的方式最快捷
            contextBytes.clear();
        }

        // 如果发现本次接收的信息中有“ over ” 关键字 ， 说明信息接收完成
        if (URLDecoder.decode(message.toString(), "UTF-8").indexOf("over") != -1) {
            // 则从 messageHashContext 中 ， 取出之前已经收到的信息 ， 组合成完整的信息
            Integer channelUUID = clientSocketChannel.hashCode();
            SocketServerReuseAddres.LOGGER.info("端口 ：" + resoucePort + "客户端发来的｛言息 ＝＝ ＝＝＝＝message : " + message);
            StringBuffer completeMessage;
            // 清空 MESSAGEHASHCONTEXT 中的历史记录
            StringBuffer historyMessage = MESSAGEHASHCONTEXT.remove(channelUUID);
            if (historyMessage == null) {
                completeMessage = message;
            } else {
                completeMessage = historyMessage.append(message);
            }
            SocketServerReuseAddres.LOGGER.info("端口 ：" + resoucePort + "客户端发来的完整 信息 ＝＝＝＝＝＝completeMessage ："
                    + URLDecoder.decode(completeMessage.toString(), "UTF-8"));

            // ======================
            // 接收完成后，可以在这里正式处理业务了
            // ＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝
            // 回发数据，并关闭 Channel
            ByteBuffer sendBuffer = ByteBuffer.wrap(URLEncoder.encode("回发处理结果", "UTF-8").getBytes());
            clientSocketChannel.write(sendBuffer);
            clientSocketChannel.close();
        } else {
            // 如果没有发现“ over ”关键字，说明还没有接收究，则将本次接收到的信息存入 MESSAGEHASHCONTEXT
            SocketServerReuseAddres.LOGGER.info("端口：" + resoucePort + "客户端信息还未接收完，继续接收＝＝＝＝＝＝message ："
                    + URLDecoder.decode(message.toString(), "UTF-8"));

            // 每一个 Channel 对象都是独立的，所以可以使用对象的 Hash 值，作为唯一标识
            Integer channelUUID = clientSocketChannel.hashCode();
            // 然后获取这个 Channel 下以前已经到达的信息
            StringBuffer historyMessage = MESSAGEHASHCONTEXT.get(channelUUID);
            if (historyMessage == null) {
                historyMessage = new StringBuffer();
                MESSAGEHASHCONTEXT.put(channelUUID, historyMessage.append(message));
            }
            historyMessage.append(message);
        }
    }
}

 

 

      多路复用 I/O 技术由操作系统提供支持，并提供给各种高级语言进行使用。它针对阻塞式同步 I/O 和非阻塞式同步 I/O 而言有很多优势，最直接的效果就是它绕过了I/O在操作系统层面的 accept()方法的阻塞问题 。

      • 使用多路复用I/O技术后 ，应用程序就可以不用再单纯使用多线程技术来解决并发I/O处理的性能问题了（针对操作系统内核I/O 管理模块和应用程序进程而言都是这样的）。在实际业务的处理中 ， 应用程序进程还是需要引 入 （由线程池支持的）多线程技术的。
       • 同一个端口可以处理多种网络协议。 例 如，使用 ServerSocketChannel 类的服务器端口监昕，既可以接收到 TCP 协议又可以接收到 UDP 协议内容。 也就是说端口的数据接收规则只和 Selector 注册的需要关心的事件有关。
       • 操作系统级别的优化 ： 多路复用 I/O 技术可以使操作系统级别在一个端口上能够同时接受多个客户端的 I/O 事件，同时具有之前我们讲到的阻塞式同步 I/O 和非阻塞式同步I/O的所有特点。 Selector 的一部分作用更相当于 “轮询代理器”。
       • 都是同步 I/O 模型：目前我们介绍的阻塞式 I/O 、非阻塞式I/O ， 甚至包括多路复用I/O ,这些都是基于操作系统级别对“同步I/O”的实现。我们一直在说“同步I/O ”， 一直都没有详细说什么叫作“同步I/O”。实际上一句话就可以解释清楚： 只有上层 （包括上层的某种代理机制）系统询问“我”是否有某个事件发生了，否则“我”不会主动告诉上层系统事件发生了。