【后端开发】Reactor 模型详解

public class BIOServer {    public static void main(String[] args) {        try {            // 服务端监听端口8080            ServerSocket serverSocket = new ServerSocket(8080);            // 服务端接收客户端链接请求            Socket socket = serverSocket.accept();            new Thread(() -> {                try {                    byte[] bytes = new byte[1024];                    // 将信息从输入流读取到创建的byte数组中                    socket.getInputStream().read(bytes);                    String message = new String(bytes, CharsetUtil.UTF_8);                    System.out.println("客户端发送过来的信息是:" + message);                    byte[] byteWrite = "Hello Client".getBytes(CharsetUtil.UTF_8);                    // 返回信息给客户端                    socket.getOutputStream().write(byteWrite);                } catch (IOException e) {                    e.printStackTrace();                }            }).start();        } catch (IOException e) {            e.printStackTrace();        }    }}

public class ChatServer {    public static void main(String[] args) throws Exception {        // 1. 创建选择器        Selector selector = Selector.open();        // 2. 创建服务端 channel        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();        // 3. 创建服务端的监听端口        serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 9000));        // 4.设置serversocketchannel 是非阻塞的        serverSocketChannel.configureBlocking(false);        // 5. 将serversocketchannel注册到selector选择器上面,并将事件设置成连接事件        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);        // 监听就绪事件        while (true) {            System.out.println("等待......");            // 休眠1秒  无论是否有读写事件发生 selector每隔1秒被唤醒            int selected = selector.select(1000);            if (selected > 0) { // 证明有事件已经准备就绪                // 返回已经就绪的事件                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();                if (iterator.hasNext()) {                    SelectionKey key = iterator.next();                    // 获取socketChannel                    if (key.isAcceptable()) {                         // 连接事件就绪，将其感兴趣的事件设置成已读事件                        // 处理接入的新请求                        handleAccept(selector, key);                    }                    if (key.isReadable()) { // 已读事件就绪                        // 处理通道的读请求                        handleRead(key);                    }                    iterator.remove();// 将处理完的数据进行了移除                }            }        }
    }
    /**     * 处理客户端读操作请求     */    private static void handleRead(SelectionKey key) {        SocketChannel socketChannel = (SocketChannel) key.channel();        // 申请一个buffer        ByteBuffer buffer = ByteBuffer.allocate(1024);        // 将通道的数据读入到buffer中        try {            socketChannel.read(buffer);        } catch (IOException e) {            e.printStackTrace();        }        System.out.println("客户端发来消息: " + new String(buffer.array(),                                                     CharsetUtil.UTF_8));    }
    /**     * 处理连接操作     */    private static void handleAccept(Selector selector, SelectionKey key) {        // 通过 ServerSocketChannel 监听过来连接的客户端事件        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();        //通过调用 accept 方法，返回一个具体的客户端连接管道        try {            SocketChannel socketChannel = serverChannel.accept();            System.out.println("客户端 " +                                socketChannel.getRemoteAddress() + "已上线......");            // 将channel 注册到selector 上面，而且需要设置成是非阻塞的            socketChannel.configureBlocking(false);            socketChannel.register(selector, SelectionKey.OP_READ);        } catch (IOException e) {            e.printStackTrace();        }    }}