Reactor 网络模型、Java代码实例

文章目录

1. 概述
2. Reactor 单线程模型
- 2.1 ByteBufferUtil
- 2.2 服务端代码
- 2.3 客户端
- 2.4 运行截图
3. Reactor多线程模型
- 3.1 服务端代码
- 3.2 运行截图
4. 主从 Reactor多线程模型
- 4.1 服务端代码
- 4.2 运行截图
参考文献

1. 概述

在 I/O 多路复用的场景下，当有数据处于就绪状态后，需要一个事件分发器（Event Dispather），它负责将读写事件分发给对应的读写事件处理器（Event Handler）。

Reactor 模型主要分为三种

Reactor 单线程模型
Reactor 多线程模型
主从 Reactor 多线程模型

Doug Lea 教授的课件 : https://gee.cs.oswego.edu/dl/cpjslides/nio.pdf

Java Socket 网络编程实例（阻塞IO、非阻塞IO、多路复用Selector、AIO）

2. Reactor 单线程模型

Reactor 单线程模型，是指所有I/O操作（监听服务端，接受客户端连接请求；消息的读取、解码、编码、发送）都在同一个NIO线程上面完成
在这里插入图片描述

2.1 ByteBufferUtil

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.StandardCharsets;

/**
 * ByteBufferUtil类提供了ByteBuffer和String之间转换的便捷方法。
 * 这些方法使用UTF-8编码进行转换，确保了数据的正确性和一致性。
 */
public class ByteBufferUtil {

    /**
     * 从ByteBuffer中读取字符串。
     *
     * @param byteBuffer 待读取的ByteBuffer，应确保其为读模式。
     * @return 从ByteBuffer解码得到的字符串。
     * @throws CharacterCodingException 如果解码过程中发生错误。
     */
    public static String read(ByteBuffer byteBuffer) throws CharacterCodingException {
        // 使用UTF-8解码器将ByteBuffer中的字节解码为CharBuffer。
        CharBuffer charBuffer = StandardCharsets.UTF_8.decode(byteBuffer);
        // 将CharBuffer转换为字符串并返回。
        return charBuffer.toString();
    }

    /**
     * 将字符串写入ByteBuffer。
     *
     * @param string 待写入的字符串。
     * @return 编码后的ByteBuffer。
     * @throws CharacterCodingException 如果编码过程中发生错误。
     */
    public static ByteBuffer read(String string) throws CharacterCodingException {
        // 使用UTF-8编码器将字符串编码为ByteBuffer。
        return StandardCharsets.UTF_8.encode(string);
    }

    /**
     * 主函数用于演示ByteBuffer和字符串之间的相互转换。
     *
     * @param args 命令行参数。
     * @throws CharacterCodingException 如果编码或解码过程中发生错误。
     */
    public static void main(String[] args) throws CharacterCodingException {
        // 将字符串"test"编码为ByteBuffer，然后从ByteBuffer解码回字符串并打印。
        System.out.println(ByteBufferUtil.read(ByteBufferUtil.read("test")));
    }

}

2.2 服务端代码

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.nio.charset.CharacterCodingException;
import java.util.Set;

public class Reactor implements Runnable {
    final Selector selector;
    final ServerSocketChannel serverSocket;

    /**
     * 初始化Reactor，打开选择器和服务器套接字通道，并注册接受操作。
     *
     * @param port 服务器监听的端口号。
     * @throws IOException 如果打开选择器或服务器套接字通道失败。
     */
    public Reactor(int port) throws IOException {
        selector = Selector.open();
        serverSocket = ServerSocketChannel.open();
        serverSocket.socket().bind(new InetSocketAddress(port));
        serverSocket.configureBlocking(false);
        SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
        sk.attach(new Acceptor());
    }

    /**
     * Reactor的主要运行方法，负责循环监听选择器上的事件，并分派处理。
     */
    public void run() {
        try {
            while (!Thread.interrupted()) {
                selector.select();
                Set<SelectionKey> selected = selector.selectedKeys();
                System.out.println("selected:" + selected.size());
                for (SelectionKey selectionKey : selected) {
                    dispatch(selectionKey);
                }
                selected.clear();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 根据选择键分派相应的处理逻辑。
     *
     * @param k 选择键。
     * @throws IOException 如果发生I/O错误。
     */
    void dispatch(SelectionKey k) throws IOException {
        Run r = (Run) (k.attachment());
        if (r != null)
            r.run();
    }

    /**
     * 接受者类，负责接受新的客户端连接。
     */
    class Acceptor implements Run { // inner
        public void run() {
            try {
                SocketChannel c = serverSocket.accept();
                if (c != null)
                    new Handler(selector, c);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    /**
     * 处理者类，负责处理客户端的读写操作。
     */
    final class Handler implements Run {
        final SocketChannel socket;
        final SelectionKey sk;
        ByteBuffer input = ByteBuffer.allocate(1024);
        ByteBuffer output = ByteBuffer.allocate(1024);

        /**
         * 初始化处理者，注册读操作兴趣。
         *
         * @param sel 选择器。
         * @param c   客户端套接字通道。
         * @throws IOException 如果注册操作失败。
         */
        Handler(Selector sel, SocketChannel c)
                throws IOException {
            socket = c;
            c.configureBlocking(false);
            sk = socket.register(sel, 0);
            sk.attach(this);
            sk.interestOps(SelectionKey.OP_READ);
            sel.wakeup();
        }

        /**
         * 检查输入缓冲区是否已完成读取。
         *
         * @return 如果输入缓冲区还有剩余数据，则返回true；否则返回false。
         */
        boolean inputIsComplete() {
            return input.hasRemaining();
        }

        /**
         * 检查输出缓冲区是否已完成写入。
         *
         * @return 如果输出缓冲区没有剩余空间，则返回true；否则返回false。
         */
        boolean outputIsComplete() {
            return !output.hasRemaining();
        }

        /**
         * 处理输入缓冲区的数据。
         *
         * @throws CharacterCodingException 如果字符编码转换失败。
         */
        void process() throws CharacterCodingException {
            // 否则，将缓冲区反转并打印读取的数据
            input.flip();
            String request = ByteBufferUtil.read(input);

            System.out.println(request);
            input.clear();

            output = ByteBufferUtil.read("你好: " + request);
        }

        /**
         * 执行处理逻辑，包括读取数据、处理数据和准备写操作。
         *
         * @throws IOException 如果发生I/O错误。
         */
        public void run() throws IOException {
            socket.read(input);
            if (inputIsComplete()) {
                process();
                sk.attach(new Sender());
                sk.interestOps(SelectionKey.OP_WRITE);
                sk.selector().wakeup();
            }
        }

        /**
         * 发送者类，负责将处理后的数据写回客户端。
         */
        class Sender implements Run {
            public void run() throws IOException {
                socket.write(output);
                if (outputIsComplete())  {
                    new Handler(selector, socket);
                }
            }
        }
    }

    /**
     * 接口Run定义了所有处理逻辑的运行方法。
     */
    public interface Run {
        public abstract void run() throws IOException;
    }

    /**
     * 程序入口点，创建并启动Reactor线程。
     *
     * @param args 命令行参数。
     * @throws IOException 如果创建Reactor失败。
     */
    public static void main(String[] args) throws IOException {
        Reactor reactor = new Reactor(6666);
        new Thread(reactor).start();
        while (true) ;
    }
}

2.3 客户端

import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;

public class SelectorClient {

    public static void main(String[] args) throws IOException, InterruptedException {
        // 创建Socket通道并连接到服务器
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 6666));

        // 初始化输入和输出ByteBuffer
        ByteBuffer inputBuffer = ByteBuffer.allocate(512);
        ByteBuffer serverOutput = ByteBuffer.allocate(512);

        // 循环接收用户输入并发送给服务器
        while (true) {
            // 使用Scanner获取用户输入
            Scanner in = new Scanner(System.in);
            String input = in.nextLine();
            System.out.println("user input: " + input);

            if (StringUtils.isBlank(input)) {
                continue;
            }

            // 清空输入缓冲区，放入用户输入，然后反转准备写入
            inputBuffer.clear();
            inputBuffer.put(input.getBytes(StandardCharsets.UTF_8));
            inputBuffer.flip();

            // 将输入数据写入Socket通道
            sc.write(inputBuffer);
            System.out.println("send to server " + input);

            // 循环读取服务器响应
            int times = 1;
            while (true) {
                // 清空服务器响应缓冲区，准备读取数据
                serverOutput.clear();
                // 从Socket通道读取数据
                sc.read(serverOutput);

                // 如果没有读取到数据，继续尝试读取
                if (!serverOutput.hasRemaining()) {
                    TimeUnit.SECONDS.sleep(1);
                    times++;
                    System.out.println(times);
                    if (times > 10) {
                        break;
                    }
                    continue;
                }

                // 反转缓冲区，读取数据并打印
                serverOutput.flip();
                System.out.println("server response " + ByteBufferUtil.read(serverOutput));

                // 读取完成后退出内层循环
                break;
            }
        }
    }
}

2.4 运行截图

在这里插入图片描述

3. Reactor多线程模型

Reactor多线程模型和 Reactor单线程模型最大的区别就是有一组NIO线程来处理I/O操作：

有一个NIO线程 Acceptor线程，监听服务端，接受客户端连接请求
网络I/O操作，读写（消息的读取、解码、编码、发送）等由一个NIO线程池负责
一个NIO线程可以处理N条链路，一个链路之对应一个NIO线程，防止出现并发问题

在这里插入图片描述

3.1 服务端代码

服务端端代码如下，客户端同上：

import lombok.SneakyThrows;
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.nio.charset.CharacterCodingException;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 该类实现了使用线程池处理NIO服务器的逻辑。
 */
public class ReactorWithThreadPool implements Runnable {

    /**
     * 处理器线程池，用于执行具体的处理任务。
     */
    static ThreadPoolExecutor HANDLER_POOL = new ThreadPoolExecutor(2, 4,
            10, TimeUnit.MINUTES, new ArrayBlockingQueue<>(20), new ThreadPoolExecutor.CallerRunsPolicy());

    final Selector selector;
    final ServerSocketChannel serverSocket;

    /**
     * 创建一个NIO服务器，监听指定端口。
     *
     * @param port 服务器监听的端口号。
     * @throws IOException 如果打开选择器或服务器套接字失败。
     */
    public ReactorWithThreadPool(int port) throws IOException {
        selector = Selector.open();
        serverSocket = ServerSocketChannel.open();
        serverSocket.socket().bind(new InetSocketAddress(port));
        serverSocket.configureBlocking(false);
        SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
        sk.attach(new Acceptor());
    }

    /**
     * 主循环，负责监听选择器上的事件。
     */
    public void run() {
        try {
            while (!Thread.interrupted()) {
                selector.select();
                Set<SelectionKey> selected = selector.selectedKeys();
                System.out.println("selected:" + selected.size());
                for (SelectionKey selectionKey : selected) {
                    if (selectionKey.isReadable()) {
                        System.out.println("selectionKey read");
                    }
                    if (selectionKey.isWritable()) {
                        System.out.println("selectionKey write");
                    }
                    dispatch(selectionKey);
                }
                selected.clear();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 分派选择键对应的处理程序。
     *
     * @param k 需要处理的选择键。
     * @throws IOException 如果操作通道失败。
     */
    void dispatch(SelectionKey k) throws IOException {
        Run r = (Run) (k.attachment());
        if (r != null)
            r.run();
    }

    /**
     * 接受者处理程序，负责接受新的客户端连接。
     */
    class Acceptor implements Run { // inner
        public void run() {
            try {
                SocketChannel c = serverSocket.accept();
                if (c != null)
                    new Handler(selector, c);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    /**
     * 处理客户端请求的处理程序。
     */
    final class Handler implements Run {
        final SocketChannel socket;
        final SelectionKey sk;
        ByteBuffer input = ByteBuffer.allocate(1024);
        ByteBuffer output = ByteBuffer.allocate(1024);

        /**
         * 创建一个新的处理程序实例。
         *
         * @param sel 选择器。
         * @param c   客户端套接字通道。
         * @throws IOException 如果注册选择键或配置套接字失败。
         */
        Handler(Selector sel, SocketChannel c)
                throws IOException {
            socket = c;
            c.configureBlocking(false);
            sk = socket.register(sel, 0);
            sk.attach(this);
            sk.interestOps(SelectionKey.OP_READ);
            sel.wakeup();
        }

        /**
         * 检查输入缓冲区是否已完成读取。
         *
         * @return 如果输入缓冲区还有剩余，则为true；否则为false。
         */
        boolean inputIsComplete() {
            return input.hasRemaining();
        }

        /**
         * 检查输出缓冲区是否已完成发送。
         *
         * @return 如果输出缓冲区没有剩余，则为true；否则为false。
         */
        boolean outputIsComplete() {
            return !output.hasRemaining();
        }

        /**
         * 处理输入数据。
         *
         * @throws CharacterCodingException 如果字符编码失败。
         */
        void process() throws CharacterCodingException {
            // 否则，将缓冲区反转并打印读取的数据
            input.flip();
            String request = ByteBufferUtil.read(input);

            System.out.println(request);
            input.clear();

            output = ByteBufferUtil.read("你好: " + request);
        }

        /**
         * 读取客户端输入，并根据情况启动处理程序或发送器。
         *
         * @throws IOException 如果读取通道失败。
         */
        public void run() throws IOException {
            socket.read(input);
            if (inputIsComplete()) {
                HANDLER_POOL.execute(new Processor());
            }
        }

        /**
         * 发送器处理程序，负责向客户端发送数据。
         */
        class Sender implements Run {
            public void run() throws IOException {
                socket.write(output);
                if (outputIsComplete()) {
                    new Handler(selector, socket);
                }
            }
        }

        /**
         * 处理请求的处理程序，负责处理输入数据并准备输出。
         */
        class Processor implements Runnable {
            @Override
            @SneakyThrows
            public void run() {
                process();
                sk.attach(new Sender());
                sk.interestOps(SelectionKey.OP_WRITE);
                sk.selector().wakeup();
            }
        }
    }

    /**
     * 处理器接口，定义了处理程序应实现的运行方法。
     */
    public interface Run {
        public abstract void run() throws IOException;
    }

    /**
     * 程序入口点。
     *
     * @param args 命令行参数。
     * @throws IOException 如果启动服务器失败。
     */
    public static void main(String[] args) throws IOException {
        ReactorWithThreadPool reactor = new ReactorWithThreadPool(6666);
        new Thread(reactor).start();
        while (true) ;
    }
}

3.2 运行截图

在这里插入图片描述

4. 主从 Reactor多线程模型

主从 Reactor多线程模型的特点：服务端接受客户端连接，不再是一个单独的NIO线程，而是一个独立的NIO线程池。

Acceptor 接收到客户端TCP连接请求并处理完成后，将新创建的SocketChannel 注册到 I/O线程池（sub Reactor）。

Acceptor线程池仅负责客户端的登陆、握手、安全认证, 一旦链路建立成功, 就将链路注册到 I/O线程池（sub Reactor）, I/O线程池（sub Reactor）负责后续的 I/O操作
在这里插入图片描述

4.1 服务端代码

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.SneakyThrows;
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.nio.charset.CharacterCodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MultiReactor implements Runnable {

    Selector selector = null;
    ServerSocketChannel serverSocket;

    static ThreadPoolExecutor REACTOR_THREAD_POOL = new ThreadPoolExecutor(2, 16,
            10, TimeUnit.MINUTES, new ArrayBlockingQueue<>(20),
            new ThreadFactoryBuilder().setNameFormat("REACTOR_THREAD_POOL-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy());

    /**
     * 构造函数，初始化多线程反应器。
     *
     * @param port 服务器监听端口。
     * @throws IOException 如果打开selector或服务器SocketChannel时发生错误。
     */
    public MultiReactor(int port) throws IOException {
        selector = Selector.open();
        serverSocket = ServerSocketChannel.open();
        serverSocket.socket().bind(new InetSocketAddress(port));
        serverSocket.configureBlocking(false);
        SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
        sk.attach(new Acceptor(serverSocket));
    }

    /**
     * 构造函数，使用已有的selector。
     *
     * @param selector 已打开的selector。
     */
    public MultiReactor(Selector selector) throws IOException {
        this.selector = selector;
    }

    /**
     * 主运行方法，负责监听和分发事件。
     */
    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                System.out.println(Thread.currentThread() + " select start");
                selector.select();
                Set<SelectionKey> selected = selector.selectedKeys();
                System.out.println(Thread.currentThread() + " " + "selected:" + selected.size());
                for (SelectionKey selectionKey : selected) {
                    if (selectionKey.isReadable()) {
                        System.out.println("selectionKey read");
                    }
                    if (selectionKey.isWritable()) {
                        System.out.println("selectionKey write");
                    }
                    dispatch(selectionKey);
                }
                selected.clear();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 分发已选择的事件到相应的处理程序。
     *
     * @param k 选择的关键。
     * @throws IOException 如果操作通道时发生错误。
     */
    void dispatch(SelectionKey k) throws IOException {
        Run r = (Run) (k.attachment());
        if (r != null)
            r.run();
    }

    /**
     * Acceptor类负责接受新的客户端连接，并将它们分配给子反应器处理。
     */
    class Acceptor implements Run {

        private final ServerSocketChannel listenSocketChannel;
        private final List<MultiReactor> subReactors = new ArrayList<>(ACCEPTOR_POOL_NUM);

        private static final int ACCEPTOR_POOL_NUM = 4;

        private final ThreadPoolExecutor ACCEPTOR_POOL = new ThreadPoolExecutor(ACCEPTOR_POOL_NUM, ACCEPTOR_POOL_NUM,
                10, TimeUnit.MINUTES, new ArrayBlockingQueue<>(20),
                new ThreadFactoryBuilder().setNameFormat("ACCEPTOR_POOL-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy());

        Acceptor(ServerSocketChannel listenSocketChannel) throws IOException {
            this.listenSocketChannel = listenSocketChannel;
            for (int i = 0; i < ACCEPTOR_POOL_NUM; i++) {
                MultiReactor subReactor = new MultiReactor(Selector.open());
                subReactors.add(subReactor);
                ACCEPTOR_POOL.execute(subReactor);
            }
        }

        /**
         * 接受新的客户端连接，并分配给子反应器处理。
         */
        @Override
        public void run() {
            try {
                SocketChannel clientSocketChannel = listenSocketChannel.accept();
                // 设置为非阻塞
                // 任意选择一个从Reactor，让其监听连接的客户端的READ事件
                Optional<MultiReactor> anySubReactor = subReactors.stream().findAny();
                if (anySubReactor.isPresent() && clientSocketChannel != null) {
                    MultiReactor subReactor = anySubReactor.get();
                    System.out.println(Thread.currentThread() + ": "+ subReactor);
                    new Handler(subReactor.selector, clientSocketChannel);
                }

            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * Handler类负责处理与客户端的通信，包括读取请求和发送响应。
     */
    final class Handler implements Run {
        final SocketChannel socket;
        final SelectionKey sk;
        ByteBuffer input = ByteBuffer.allocate(1024);
        ByteBuffer output = ByteBuffer.allocate(1024);

        Handler(Selector sel, SocketChannel c)
                throws IOException {
            socket = c;
            c.configureBlocking(false);
            sel.wakeup();
            sk = socket.register(sel, SelectionKey.OP_READ);
            sk.attach(this);
            sk.interestOps(SelectionKey.OP_READ);
        }

        /**
         * 检查输入缓冲区是否已完成读取。
         *
         * @return 如果输入缓冲区还有剩余，则返回true；否则返回false。
         */
        boolean inputIsComplete() {
            return input.hasRemaining();
        }

        /**
         * 检查输出缓冲区是否已完成发送。
         *
         * @return 如果输出缓冲区没有剩余，则返回true；否则返回false。
         */
        boolean outputIsComplete() {
            return !output.hasRemaining();
        }

        /**
         * 处理输入数据，将其解码并准备生成响应。
         *
         * @throws CharacterCodingException 如果字符编码发生错误。
         */
        void process() throws CharacterCodingException {
            // 否则，将缓冲区反转并打印读取的数据
            input.flip();
            String request = ByteBufferUtil.read(input);

            System.out.println(Thread.currentThread() + ": " + request);
            input.clear();
            System.out.println(input.toString());

            System.out.println(output.toString());
            output = ByteBufferUtil.read("你好: " + request);
        }

        /**
         * 读取客户端请求，并根据需要启动处理过程。
         *
         * @throws IOException 如果读取通道时发生错误。
         */
        @Override
        public void run() throws IOException {
            socket.read(input);
            if (inputIsComplete()) {
                REACTOR_THREAD_POOL.execute(new Processor(sk.selector()));
            }
        }

        /**
         * Sender类负责发送响应给客户端。
         */
        class Sender implements Run {
            private Selector selector;

            public Sender(Selector selector) {
                this.selector = selector;
            }

            /**
             * 发送输出缓冲区中的数据到客户端。
             *
             * @throws IOException 如果写入通道时发生错误。
             */
            public void run() throws IOException {
                System.out.println("start write");
                socket.write(output);
                if (outputIsComplete()) {
                    new Handler(this.selector, socket);
                }
            }
        }

        /**
         * Processor类负责处理请求并准备响应。
         */
        class Processor implements Runnable {

            private Selector selector;

            public Processor(Selector selector) {
                this.selector = selector;
            }

            @Override
            @SneakyThrows
            public void run() {
                process();
                sk.attach(new Sender(this.selector));
                sk.interestOps(SelectionKey.OP_WRITE);
                sk.selector().wakeup();
            }
        }
    }

    /**
     * Run接口定义了处理事件的运行时行为。
     */
    public interface Run {
        void run() throws IOException;
    }

    /**
     * 程序入口点。
     *
     * @param args 命令行参数。
     * @throws IOException 如果初始化反应器时发生错误。
     */
    public static void main(String[] args) throws IOException {
        MultiReactor reactor = new MultiReactor(6666);
        new Thread(reactor).start();
        while (true) ;
    }
}

4.2 运行截图

在这里插入图片描述

参考文献

Doug Lea 教授的课件 : https://gee.cs.oswego.edu/dl/cpjslides/nio.pdf
Netty权威指南（第2版）李林锋 / 著
https://juejin.cn/post/7210375522512666679?searchId=20240612213218FE474007F2FADD0130AA

在这里插入图片描述