[toc]

---

• Posted by 微博@Yangsc_o
• 原创文章，版权声明：自由转载-非商用-非衍生-保持署名 | Creative Commons BY-NC-ND 3.0

---

# 摘要

在前两篇《快速理解Linux网络I_O》、《java的I_O模型-BIO&NIO&AIO》两边中介绍了Linux下的I/O模型和java中的I/O模型，今天我们介绍Reactor模型，并探究Netty的实现

高性能服务器

在互联网时代，我们使用的软件基本上全是C/S架构，C/S架构的软件一个明显的好处就是：只要有网络，你可以在任何地方干同一件事。C/S架构可以抽象为如下模型：

image-20200608122739089

• C就是Client(客户端),上面的B是Browser(浏览器)
• S就是Server(服务器)：服务器管理某种资源，并且通过操作这种资源来为它的客户端提供某种服务

那服务器如何能快速的处理用户的请求呢？在我看来高性能服务器至少要满足如下几个需求：

• 效率高：既然是高性能，那处理客户端请求的效率当然要很高了
• 高可用：不能随便就挂掉了
• 编程简单：基于此服务器进行业务开发需要足够简单
• 可扩展：可方便的扩展功能
• 可伸缩：可简单的通过部署的方式进行容量的伸缩，也就是服务需要无状态

而满足如上需求的一个基础就是高性能的IO!

Reactor模式

什么是Reactor模式？

两种I/O多路复用模式：Reactor和Proactor，两个与事件分离器有关的模式是Reactor和Proactor。Reactor模式采用同步IO，而Proactor采用异步IO。

在Reactor中，事件分离器负责等待文件描述符或socket为读写操作准备就绪，然后将就绪事件传递给对应的处理器，最后由处理器负责完成实际的读写工作。

在Proactor模式中，处理器--或者兼任处理器的事件分离器，只负责发起异步读写操作。IO操作本身由操作系统来完成。传递给操作系统的参数需要包括用户定义的数据缓冲区地址和数据大小，操作系统才能从中得到写出操作所需数据，或写入从socket读到的数据。事件分离器捕获IO操作完成事件，然后将事件传递给对应处理器。

说人话的方式理解：

• reactor：能收了你跟俺说一声。
• proactor: 你给我收十个字节，收好了跟俺说一声。

Doug Lea是这样类比的

• Reactor通过调度适当的处理程序来响应IO事件；
• 处理程序执行非阻塞操作
• 通过将处理程序绑定到事件来管理；

image-20200608173303651

Reactor单线程模型设计

image-20200608175038780

单线程版本Java NIO的支持：

• Channels：与支持非阻塞读取的文件，套接字等的连接

• Buffers：类似于数组的对象，可由Channels直接读取或写入

• Selectors：通知一组通道中哪一个有IO事件

• SelectionKeys：维护IO事件状态和绑定

• Reactor 代码如下

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

    public Reactor(int port) throws IOException {
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);
        SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        key.attach(new Acceptor());
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                for (SelectionKey selectionKey : selectionKeys) {
                    dispatch(selectionKey);
                }
                selectionKeys.clear();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void dispatch(SelectionKey selectionKey) {
        Runnable run = (Runnable) selectionKey.attachment();
        if (run != null) {
            run.run();
        }
    }

    class Acceptor implements Runnable {
        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocketChannel.accept();
                if (channel != null) {
                    new Handler(selector, channel);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException {
        new Thread(
                new Reactor(1234)
        ).start();
    }

}

• Handler代码如下：

public class Handler implements Runnable{
    private final static int DEFAULT_SIZE = 1024;
    private final SocketChannel socketChannel;
    private final SelectionKey seletionKey;
    private static final int READING = 0;
    private static final int SENDING = 1;
    private int state = READING;

    ByteBuffer inputBuffer = ByteBuffer.allocate(DEFAULT_SIZE);
    ByteBuffer outputBuffer = ByteBuffer.allocate(DEFAULT_SIZE);

    public Handler(Selector selector, SocketChannel channel) throws IOException {
        this.socketChannel = channel;
        socketChannel.configureBlocking(false);
        this.seletionKey = socketChannel.register(selector, 0);
        seletionKey.attach(this);
        seletionKey.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }

    @Override
    public void run() {
        if (state == READING) {
            read();
        } else if (state == SENDING) {
            write();
        }
    }


    private void write() {
        try {
            socketChannel.write(outputBuffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
        while (outIsComplete()) {
            seletionKey.cancel();
        }
    }

    private void read() {
        try {
            socketChannel.read(inputBuffer);
            if (inputIsComplete()) {
                process();
                System.out.println("接收到来自客户端（" + socketChannel.socket().getInetAddress().getHostAddress()
                        + "）的消息：" + new String(inputBuffer.array()));
                seletionKey.attach(new Sender());
                seletionKey.interestOps(SelectionKey.OP_WRITE);
                seletionKey.selector().wakeup();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean inputIsComplete() {
        return true;
    }
    public boolean outIsComplete() {
        return true;
    }


    public void process() {
        // do something...
    }

    class Sender implements Runnable {
        @Override
        public void run() {
            try {
                socketChannel.write(outputBuffer);
            } catch (IOException e) {
                e.printStackTrace();
            }
            if (outIsComplete()) {
                seletionKey.cancel();
            }
        }
    }

}

这个模型和上面的NIO流程很类似，只是将消息相关处理独立到了Handler中去了！虽然说到NIO一个线程就可以支持所有的IO处理。但是瓶颈也是显而易见的！如果这个客户端多次进行请求，如果在Handler中的处理速度较慢，那么后续的客户端请求都会被积压，导致响应变慢！所以引入了Reactor多线程模型!

Reactor多线程模型设计

Reactor多线程模型就是将Handler中的IO操作和非IO操作分开，操作IO的线程称为IO线程，非IO操作的线程称为工作线程!这样的话，客户端的请求会直接被丢到线程池中，客户端发送请求就不会堵塞！

image-20200608190105063

Reactor保持不变，仅需要改动Handler代码：

public class Handler implements Runnable{
    private final static int DEFAULT_SIZE = 1024;
    private final SocketChannel socketChannel;
    private final SelectionKey seletionKey;
    private static final int READING = 0;
    private static final int SENDING = 1;
    private int state = READING;

    ByteBuffer inputBuffer = ByteBuffer.allocate(DEFAULT_SIZE);
    ByteBuffer outputBuffer = ByteBuffer.allocate(DEFAULT_SIZE);

    private static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime()
            .availableProcessors());
    private static final int PROCESSING = 3;
    private Selector selector;


    public Handler(Selector selector, SocketChannel channel) throws IOException {
        this.selector = selector;
        this.socketChannel = channel;
        socketChannel.configureBlocking(false);
        this.seletionKey = socketChannel.register(selector, 0);
        seletionKey.attach(this);
        seletionKey.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }

    @Override
    public void run() {
        if (state == READING) {
            read();
        } else if (state == SENDING) {
            write();
        }
    }

    private void write() {
        try {
            socketChannel.write(outputBuffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
        while (outIsComplete()) {
            seletionKey.cancel();
        }
    }

    private void read() {
        try {
            socketChannel.read(inputBuffer);
            if (inputIsComplete()) {
                process();
                executorService.execute(new Processer());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean inputIsComplete() {
        return true;
    }
    public boolean outIsComplete() {
        return true;
    }


    public void process() {
        // do something...
    }

    class Sender implements Runnable {
        @Override
        public void run() {
            try {
                socketChannel.write(outputBuffer);
            } catch (IOException e) {
                e.printStackTrace();
            }
            if (outIsComplete()) {
                seletionKey.cancel();
            }
        }
    }

    synchronized void processAndHandOff() {
        process();
        // or rebind attachment
        state = SENDING;
        seletionKey.interestOps(SelectionKey.OP_WRITE);
        selector.wakeup();
    }

    class Processer implements Runnable {
        @Override
        public void run() {
            processAndHandOff();
        }
    }

}

主从Reactor多线程模型设计

主从Reactor多线程模型是将Reactor分成两部分，mainReactor负责监听server socket，accept新连接，并将建立的socket分派给subReactor。subReactor负责多路分离已连接的socket，读写网络数据，对业务处理功能，其扔给worker线程池完成。通常，subReactor个数上可与CPU个数等同：

image-20200608222654773

Handler保持不变，仅需要改动Reactor代码：

public class Reactor {
    // also create threads
    Selector[] selectors;
    AtomicInteger next = new AtomicInteger(0);
    final ServerSocketChannel serverSocketChannel;

    private static ExecutorService sunReactors = Executors.newFixedThreadPool(Runtime.getRuntime()
            .availableProcessors());
    private static final int PROCESSING = 3;

    public Reactor(int port) throws IOException {
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);
        selectors = new Selector[4];
        for (int i = 0; i < selectors.length; i++) {
            Selector selector = selectors[i];
            serverSocketChannel.socket().bind(new InetSocketAddress(port));
            serverSocketChannel.configureBlocking(false);
            SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            key.attach(new Acceptor());
            new Thread(()->{
                while (!Thread.interrupted()) {
                    try {
                        selector.select();
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        for (SelectionKey selectionKey : selectionKeys) {
                            dispatch(selectionKey);
                        }
                        selectionKeys.clear();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }



    private void dispatch(SelectionKey selectionKey) {
        Runnable run = (Runnable) selectionKey.attachment();
        if (run != null) {
            run.run();
        }
    }

    class Acceptor implements Runnable {
        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocketChannel.accept();
                if (channel != null) {
                    sunReactors.execute(new Handler(selectors[next.getAndIncrement() % selectors.length], channel));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException {
        new Reactor(1234);
    }

}

以上是三种不同的设计思路，接下来看一下Netty这个一个高性能NIO框架，其是如何实现Reactor模型的！

Netty Reactor模型设计

• 看一个最简单的Netty服务端代码

public final class EchoServer {
    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
    public static void main(String[] args) throws Exception {
        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(serverHandler);
                 }
             });
            ChannelFuture f = b.bind(PORT).sync();
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

• Netty Server Handler

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ctx.write(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

我们从Netty服务器代码来看，与Reactor模型进行对应！

• EventLoopGroup就相当于是Reactor，bossGroup对应主Reactor,workerGroup对应从Reactor
• TimeServerHandler就是Handler
• child开头的方法配置的是客户端channel，非child开头的方法配置的是服务端channel

当然Netty的线程模型并不是固定的，它支持Reactor单线程模型、Reactor多线程模型、主从模型，上面的例子是一个主从模型的，下面进行详细的分析，如图所示：

image-20200610100723181

服务启动时，创建了两个EventLoopGroup，它们实际上是两个Reactor线程池，一个用于接收TCP连接、一个用于处理I/O相关的读写操作、或者执行系统task、定时task等；

• Netty用于接收客户端请求连接池职责如下：
  • 接收客户端请求并初始化channel参数；
  • 讲链路变更事件通知给ChannelPipiline；
• Netty用于处理I/O连接池职责如下：
  • 异步读取通信对端的数据报，发送读事件到ChannelPipiline；
  • 异步发送消息对端的数据报，调用ChannelPipiline的消息发送接口；
  • 执行系统调用task；
  • 执行系统定时任务task，例如链路空闲状态检测定时任务；

参考

Scalable IO in Java

高性能Server---Reactor模型

NIO技术概览

《Netty 权威指南》第二版 -- 李林峰

你的鼓励也是我创作的动力

打赏地址