网络I/O模型--05多路复用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);
        }
    }
}