【Netty】Reactor 模型（十）

前言

回顾Netty系列文章：

• Netty 概述（一）
• Netty 架构设计（二）
• Netty Channel 概述（三）
• Netty ChannelHandler（四）
• ChannelPipeline源码分析（五）
• 字节缓冲区 ByteBuf （六）（上）
• 字节缓冲区 ByteBuf（七）（下）
• Netty 如何实现零拷贝（八）
• Netty 程序引导类（九）

说到 NIO、Netty，Reactor模型一定是绕不开的，因为这种模式架构太经典了，接下来我们就静下心来好好看看Netty的基石——Reactor模型。

本文就带着大家看看 Reactor 模型，让大家对 Reactor 模型有个浅显而又感性的认识。

一、传统服务的设计模型

这是最为传统的Socket服务设计，有多个客户端连接服务端，服务端会开启很多线程，一个线程为一个客户端服务。
在绝大多数场景下，处理一个网络请求有如下几个步骤：

1. read：从socket读取数据。
2. decode：解码，网络上的数据都是以byte的形式进行传输的，要想获取真正的请求，必定需要解码。
3. compute：计算，也就是业务处理，你想干啥就干啥。
4. encode：编码，同理，因为网络上的数据都是以byte的形式进行传输的，也就是socket只接收byte，所以必定需要编码。

二、NIO 分发模型

NIO就很好的解决了传统Socket问题：

1. 一个线程可以监听多个Socket，不再是一夫当关，万夫莫开；
2. 基于事件驱动：等发生了各种事件，系统可以通知我，我再去处理。

三、Reactor 模型

Reactor 也可以称作反应器模型，它有以下几个特点：

• Reactor 模型中会通过分配适当的处理器来响应 I/O 事件。
• 每个处理器执行非阻塞的操作。
• 通过将处理器绑定到事件进行管理。

Reactor 模型整合了分发模型和事件驱动这两大优势，特别适合处理海量的 I/O 事件及高并发的场景。

3.1、Reactor 处理请求的流程

Reactor 处理请求的流程主要分为读取和写入两种操作。
对于读取操作而言，流程如下：

• 应用程序注册读就绪事件和相关联的事件处理器。
• 事件分发器等待事件的发生。
• 当发生读就绪事件时，事件分离器调用第一步注册的事件处理器。

写入操作类似于读取操作，只不过第一步注册的是写就绪事件。

3.2、Reactor 三种角色

Reactor 模型中定义了 3 种角色。

• Reactor ：负责监听和分配事件，将 I/O 事件分派给对应的 Handler。新的事件包含连接建立就绪、读就绪、写就绪等。
• Acceptor：处理客户端新连接，并分派请求到处理器链中。
• Handler：将自身与事件绑定，执行非阻塞读/写任务，完成 channel 的读入，完成处理业务逻辑后，负责将结果写出 Channel。可用资源池来管理。

根据不同的应用场景，Reactor 模型又可以细分为：单Reactor 单线程模型、单Reactor 多线程模型及主从Reactor 多线程模型。

四、单Reactor 单线程模型

下图展示的就是单线程下的 Reactor 设计模型。Reactor 线程负责多路分离套接字，Accept 负责接收新连接，并分派请求到 Handler。

4.1、消息处理流程

单Reactor 单线程模型的消息处理流程如下：

• Reactor 对象通过 select 监控连接事件，收到事件后通过 dispatch 进行转发。
• 如果是连接建立的事件，则由 Acceptor 接收连接，并创建 Handler 处理后续的事件。
• 如果不是建立连接事件，则 Reactor 会分发调用 Handler 来响应。
• Handler 会完成 read、decode、compute、encode、send等一整套流程。

4.2、缺点

单Reactor 单线程模型只是在代码上进行了组件的区分，但是整体操作还是单线程，不能充分利用硬件资源。Handler 业务处理部分没有异步。

对于一些小容量应用场景，可以使用单Reactor 单线程模型。但是对于高负载、高并发的应用场景却不合适。主要原因如下：

• 即便 Reactor 线程的 CPU 负荷达到 100%，也无法满足海量消息的 read、decode、compute、encode和send。
• 单 Reactor 线程负载过重后，处理速度将变慢，这会导致大量客户端连接超时，超时之后往往会进行重发，这更加重了 Reactor 线程的负荷，最终会导致大量消息积压和处理超时，成为系统的性能瓶颈。
• 一旦 Reactor 线程意外中断或者进入死循环，会导致整个系统通信模块不可用，不能接收和处理外部消息，造成节点故障。

为了解决上述的问题，单Reactor 多线程模型便出现了。

五、单Reactor 多线程模型

下图展示的就是单 Reactor 多线程的设计模型。该模型在事件处理器（Handler）部分采用了多线程（线程池）。

5.1、消息处理流程

单Reactor 多线程模型的消息处理流程如下：

• Reactor 对象通过 select 监控客户端请求事件，收到事件后通过 dispatch 进行分发。
• 如果是建立连接请求事件，则由 Acceptor 通过 accept 处理连接请求，然后创建一个 Handler 对象处理连接完成后续的各种事件。
• 如果不是建立连接事件，则 Reactor 会分发调用 Handler 来响应。
• Handler 只负责响应事件，不做具体业务处理，通过 read 读取数据后，会分发给后面的 Worker 线程池进行业务处理。
• Worker 线程池会分配独立的线程完成真正的业务处理，将响应的结果发送给 Handler 进行处理。
• Handler 收到响应结果后会通过 send 将响应结果返回给 Client。

相对于第一种模型来说，该业务逻辑是交由线程池来处理的，Handler 收到响应后通过 send 将响应结果返回给客户端。这样可以降低 Reactor 的性能开销，从而更专注地做事件分发工作，提升了整个应用的吞吐性能。

5.2、缺点

单Reactor 多线程模型存在以下问题。

• 多线程数据共享和访问比较复杂。如果子线程完成业务处理后，把结果传递给主线程 Reactor 进行发送，就会涉及共享数据的互斥和保护机制。
• Reactor 承担所有事件的监听和响应，只在主线程中运行，可能会存在性能问题。例如，并发百万客户端连接，或者服务端需要对客户端握手进行安全认证，但是认证本身非常损耗性能。

为了解决上述的性能问题，产生了第三种 主从 Reactor 多线程模型。

六、主从Reactor 多线程模型

相较于单Reactor 多线程模型，主从Reactor 多线程模型是将 Reactor 分成两部分。

• mainReactor（主 Reactor）负责监听 Server Socket，用来处理网络 I/O 连接事件操作，将建立的 - - SocketChannel 指定注册给 SubReactor。
• SubReactor（从 Reactor）主要和建立连接起来的 socket 做数据交互和事件业务处理操作。通常，SubReactor 个数可与 CPU 个数等同。

Nginx、Swoole、Memcached和 Netty 都采用了这种实现。
主从Reactor 多线程模型的消息处理流程如下：

• 从主线程池随机选择一个 Reactor 线程作为 Acceptor 线程，用于绑定监听端口，接收客户端连接。
• Acceptor 线程接收客户端连接请求之后创建新的 SocketChannel ，将其注册到主线程池的其他 -Reactor 线程上，由其负责接入认证、IP黑白名单过滤、握手等操作。
• 上述步骤完成之后，业务层的链路正式建立，将 SocketChannel 从主线程池的 Reactor 线程的多路复用器上摘除，重新注册到子线程池的线程上，并创建一个 Handler 用于处理各种连接事件。
• 当有新的事件发生时，SubReactor 会调用连接对应的 Handler 进行响应。
• Handler 通过 Read 读取数据后，会分发给后面的 Worker 线程池进行业务处理。
• Worker 线程池会分配独立的线程完成真正的业务处理，将响应的结果发送给 Handler 进行处理。
• Handler 收到响应结果后会通过 send 将响应结果返回给 Client。

主从Reactor 多线程模型示例

6.1、Reactor

public class Reactor implements Runnable {

	private final Selector selector;
	private final ServerSocketChannel serverSocketChannel;

	public Reactor(int port) throws IOException {
		selector = Selector.open(); // 打开一个Selector
		serverSocketChannel = ServerSocketChannel.open(); // 建立一个Server端通道
		serverSocketChannel.socket().bind(new InetSocketAddress(port)); // 绑定服务端口
		serverSocketChannel.configureBlocking(false); // selector模式下，所有通道必须是非阻塞的

		// Reactor是入口，最初给一个channel注册上去的事件都是accept
		SelectionKey sk = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

		// 绑定Acceptor处理类
		sk.attach(new Acceptor(serverSocketChannel));
	}

	@Override
	public void run() {
		try {
			while (!Thread.interrupted()) {
				int count = selector.select(); // 就绪事件到达之前，阻塞
				if (count == 0) {
					continue;
				}
				Set<SelectionKey> selected = selector.selectedKeys(); // 拿到本次select获取的就绪事件
				Iterator<SelectionKey> it = selected.iterator();
				while (it.hasNext()) {
					// 这里进行任务分发
					dispatch((SelectionKey) (it.next()));
				}
				selected.clear();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	void dispatch(SelectionKey k) {
		// 附带对象为Acceptor
		Runnable r = (Runnable) (k.attachment());

		// 调用之前注册的回调对象
		if (r != null) {
			r.run();
		}
	}
}

该模块内容包含两个核心方法，即select和dispatch，该模块负责监听就绪事件和对事件的分发处理。分发附带对象为Acceptor处理类。

6.2、Acceptor

public class Acceptor implements Runnable {

	private final ServerSocketChannel serverSocketChannel;

	private final int coreNum = Runtime.getRuntime().availableProcessors(); // CPU核心数

	private final Selector[] selectors = new Selector[coreNum]; // 创建selector给SubReactor使用

	private int next = 0; // 轮询使用subReactor的下标索引

	private SubReactor[] reactors = new SubReactor[coreNum]; // subReactor

	private Thread[] threads = new Thread[coreNum]; // subReactor的处理线程

	Acceptor(ServerSocketChannel serverSocketChannel) throws IOException {
		this.serverSocketChannel = serverSocketChannel;
		// 初始化
		for (int i = 0; i < coreNum; i++) {
			selectors[i] = Selector.open();
			reactors[i] = new SubReactor(selectors[i], i); // 初始化sub reactor
			threads[i] = new Thread(reactors[i]); // 初始化运行sub reactor的线程
			threads[i].start(); // 启动（启动后的执行参考SubReactor里的run方法）
		}
	}

	@Override
	public void run() {
		SocketChannel socketChannel;
		try {
			socketChannel = serverSocketChannel.accept(); // 连接
			if (socketChannel != null) {
				System.out.println(String.format("accpet %s", socketChannel.getRemoteAddress()));
				socketChannel.configureBlocking(false);

				// 注意一个selector在select时是无法注册新事件的，因此这里要先暂停下select方法触发的程序段，
				// 下面的weakup和这里的setRestart都是做这个事情的，具体参考SubReactor里的run方法
				reactors[next].registering(true);
				selectors[next].wakeup(); // 使一个阻塞住的selector操作立即返回
				SelectionKey selectionKey = 
						socketChannel.register(selectors[next], SelectionKey.OP_READ); // 注册一个读事件
				selectors[next].wakeup(); // 使一个阻塞住的selector操作立即返回

				// 本次事件注册完成后，需要再次触发select的执行，
				// 因此这里Restart要在设置回false（具体参考SubReactor里的run方法）
				reactors[next].registering(false);

				// 绑定Handler
				selectionKey.attach(new AsyncHandler(socketChannel, selectors[next], next));
				if (++next == selectors.length) {
					next = 0; // 越界后重新分配
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}

该模块负责处理连接就绪的事件，并初始化一批subReactor进行分发处理，拿到客户端的socketChannel，绑定Handler，这样就可以继续完成接下来的读写任务了。

6.3、subReactor

public class SubReactor implements Runnable {
	private final Selector selector;
	private boolean register = false; // 注册开关表示
	private int num; // 序号，也就是Acceptor初始化SubReactor时的下标

	SubReactor(Selector selector, int num) {
		this.selector = selector;
		this.num = num;
	}

	@Override
	public void run() {
		while (!Thread.interrupted()) {
			System.out.println(String.format("NO %d SubReactor waitting for register...", num));
			while (!Thread.interrupted() && !register) {
				try {
					if (selector.select() == 0) {
						continue;
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
				Set<SelectionKey> selectedKeys = selector.selectedKeys();
				Iterator<SelectionKey> it = selectedKeys.iterator();
				while (it.hasNext()) {
					dispatch(it.next());
					it.remove();
				}
			}
		}
	}

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

	void registering(boolean register) {
		this.register = register;
	}

}

这个类负责Acceptor交给自己的事件select，在上述例子中实际就是read和send操作。

6.4、AsyncHandler

public class AsyncHandler implements Runnable {

	private final Selector selector;

	private final SelectionKey selectionKey;
	private final SocketChannel socketChannel;

	private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
	private ByteBuffer sendBuffer = ByteBuffer.allocate(2048);

	private final static int READ = 0; // 读取就绪
	private final static int SEND = 1; // 响应就绪
	private final static int PROCESSING = 2; // 处理中

	private int status = READ; // 所有连接完成后都是从一个读取动作开始的

	private int num; // 从反应堆序号

	// 开启线程数为4的异步处理线程池
	private static final ExecutorService workers = Executors.newFixedThreadPool(5);

	AsyncHandler(SocketChannel socketChannel, Selector selector, int num) throws IOException {
		this.num = num; // 为了区分Handler被哪个从反应堆触发执行做的标记
		this.socketChannel = socketChannel; // 接收客户端连接
		this.socketChannel.configureBlocking(false); // 置为非阻塞模式
		selectionKey = socketChannel.register(selector, 0); // 将该客户端注册到selector
		selectionKey.attach(this); // 附加处理对象，当前是Handler对象
		selectionKey.interestOps(SelectionKey.OP_READ); // 连接已完成，那么接下来就是读取动作
		this.selector = selector;
		this.selector.wakeup();
	}

	@Override
	public void run() {
		// 如果一个任务正在异步处理，那么这个run是直接不触发任何处理的，
		// read和send只负责简单的数据读取和响应，业务处理完全不阻塞这里的处理
		switch (status) {
		case READ:
			read();
			break;
		case SEND:
			send();
			break;
		default:
		}
	}

	private void read() {
		if (selectionKey.isValid()) {
			try {
				readBuffer.clear();

				// read方法结束，意味着本次"读就绪"变为"读完毕"，标记着一次就绪事件的结束
				int count = socketChannel.read(readBuffer);
				if (count > 0) {
					status = PROCESSING; // 置为处理中
					workers.execute(this::readWorker); // 异步处理
				} else {
					selectionKey.cancel();
					socketChannel.close();
					System.out.println(String.format("NO %d SubReactor read closed", num));
				}
			} catch (IOException e) {
				System.err.println("处理read业务时发生异常！异常信息：" + e.getMessage());
				selectionKey.cancel();
				try {
					socketChannel.close();
				} catch (IOException e1) {
					System.err.println("处理read业务关闭通道时发生异常！异常信息：" + e.getMessage());
				}
			}
		}
	}

	void send() {
		if (selectionKey.isValid()) {
			status = PROCESSING; // 置为执行中
			workers.execute(this::sendWorker); // 异步处理
			selectionKey.interestOps(SelectionKey.OP_READ); // 重新设置为读
		}
	}

	// 读入信息后的业务处理
	private void readWorker() {
		try {

			// 模拟一段耗时操作
			Thread.sleep(5000L);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		try {
			System.out.println(String.format("NO %d %s -> Server： %s", 
					num, socketChannel.getRemoteAddress(),
					new String(readBuffer.array())));
		} catch (IOException e) {
			System.err.println("异步处理read业务时发生异常！异常信息：" + e.getMessage());
		}
		status = SEND;
		selectionKey.interestOps(SelectionKey.OP_WRITE); // 注册写事件
		this.selector.wakeup(); // 唤醒阻塞在select的线程
	}

	private void sendWorker() {
		try {
			sendBuffer.clear();
			sendBuffer.put(String.format("NO %d SubReactor recived %s from %s", num,
					new String(readBuffer.array()), 
					socketChannel.getRemoteAddress()).getBytes());
			sendBuffer.flip();

			// write方法结束，意味着本次写就绪变为写完毕，标记着一次事件的结束
			int count = socketChannel.write(sendBuffer);

			if (count < 0) {
				// 同上，write场景下，取到-1，也意味着客户端断开连接
				selectionKey.cancel();
				socketChannel.close();
				System.out.println(String.format("%d SubReactor send closed", num));
			}

			// 没断开连接，则再次切换到读
			status = READ;
		} catch (IOException e) {
			System.err.println("异步处理send业务时发生异常！异常信息：" + e.getMessage());
			selectionKey.cancel();
			try {
				socketChannel.close();
			} catch (IOException e1) {
				System.err.println("异步处理send业务关闭通道时发生异常！异常信息：" + e.getMessage());
			}
		}
	}
}

AsyncHandler负责接下来的读写操作。

6.5、MainSubReactorDemo

public class MainSubReactorDemo {

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

}

七、客户端

7.1、Connector

public class Connector implements Runnable {

	private final Selector selector;

	private final SocketChannel socketChannel;

	Connector(SocketChannel socketChannel, Selector selector) {
		this.socketChannel = socketChannel;
		this.selector = selector;
	}

	@Override
	public void run() {
		try {
			if (socketChannel.finishConnect()) {
				// 这里连接完成（与服务端的三次握手完成）
				System.out.println(String.format("connected to %s", socketChannel.getRemoteAddress()));

				// 连接建立完成后，接下来的动作交给Handler去处理（读写等）
				new Handler(socketChannel, selector);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}

7.2、Handler

public class Handler implements Runnable {

	private final SelectionKey selectionKey;
	private final SocketChannel socketChannel;

	private ByteBuffer readBuffer = ByteBuffer.allocate(2048);
	private ByteBuffer sendBuffer = ByteBuffer.allocate(1024);

	private final static int READ = 0;
	private final static int SEND = 1;

	private int status = SEND; // 与服务端不同，默认最开始是发送数据

	private AtomicInteger counter = new AtomicInteger();

	Handler(SocketChannel socketChannel, Selector selector) throws IOException {
		this.socketChannel = socketChannel; // 接收客户端连接
		this.socketChannel.configureBlocking(false); // 置为非阻塞模式
		selectionKey = socketChannel.register(selector, 0); // 将该客户端注册到selector
		selectionKey.attach(this); // 附加处理对象，当前是Handler对象
		selectionKey.interestOps(SelectionKey.OP_WRITE); // 建连已完成，那么接下来就是读取动作
		selector.wakeup(); // 唤起select阻塞
	}

	@Override
	public void run() {
		try {
			switch (status) {
			case SEND:
				send();
				break;
			case READ:
				read();
				break;
			default:
			}
		} catch (IOException e) {
			// 这里的异常处理是做了汇总，同样的，客户端也面临着正在与服务端进行写/读数据时，
			// 突然因为网络等原因，服务端直接断掉连接，这个时候客户端需要关闭自己并退出程序
			System.err.println("send或read时发生异常！异常信息：" + e.getMessage());
			selectionKey.cancel();
			try {
				socketChannel.close();
			} catch (IOException e2) {
				System.err.println("关闭通道时发生异常！异常信息：" + e2.getMessage());
				e2.printStackTrace();
			}
		}
	}

	void send() throws IOException {
		if (selectionKey.isValid()) {
			sendBuffer.clear();
			int count = counter.incrementAndGet();
			if (count <= 10) {
				sendBuffer.put(String.format("msg is %s", count).getBytes());
				sendBuffer.flip(); // 切换到读模式，用于让通道读到buffer里的数据
				socketChannel.write(sendBuffer);

				// 则再次切换到读，用以接收服务端的响应
				status = READ;
				selectionKey.interestOps(SelectionKey.OP_READ);
			} else {
				selectionKey.cancel();
				socketChannel.close();
			}
		}
	}

	private void read() throws IOException {
		if (selectionKey.isValid()) {
			readBuffer.clear(); // 切换成buffer的写模式，用于让通道将自己的内容写入到buffer里
			socketChannel.read(readBuffer);
			System.out.println(String.format("Server -> Client: %s", new String(readBuffer.array())));

			// 收到服务端的响应后，再继续往服务端发送数据
			status = SEND;
			selectionKey.interestOps(SelectionKey.OP_WRITE); // 注册写事件
		}
	}
}

7.3、NIOClient

public class NIOClient implements Runnable {

    private Selector selector;

    private SocketChannel socketChannel;

    NIOClient(String ip, int port) {
        try {
            selector = Selector.open(); //打开一个Selector
            socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false); //设置为非阻塞模式
            socketChannel.connect(new InetSocketAddress(ip, port)); //连接服务
            
            //入口，最初给一个客户端channel注册上去的事件都是连接事件
            SelectionKey sk = socketChannel.register(selector, SelectionKey.OP_CONNECT);
            
            //附加处理类，第一次初始化放的是连接就绪处理类
            sk.attach(new Connector(socketChannel, selector));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
            	 //就绪事件到达之前，阻塞
                selector.select();
                
                //拿到本次select获取的就绪事件
                Set<SelectionKey> selected = selector.selectedKeys();
                Iterator<SelectionKey> it = selected.iterator();
                while (it.hasNext()) {
                    //这里进行任务分发
                    dispatch((SelectionKey) (it.next()));
                }
                selected.clear();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    void dispatch(SelectionKey k) {
    	// 附带对象为Connector（
        Runnable r = (Runnable) (k.attachment()); 
        
        //调用之前注册的回调对象
        if (r != null) {
            r.run();
        }
    }
}

7.4、ClientDemo

public class ClientDemo {

    public static void main(String[] args) {
        new Thread(new NIOClient("127.0.0.1", 2333)).start();
        new Thread(new NIOClient("127.0.0.1", 2333)).start();
    }

}

7.5、测试

运行上述应用及客户端，在控制台输出如下内容：

NO 2 SubReactor waitting for register...
NO 1 SubReactor waitting for register...
NO 3 SubReactor waitting for register...
NO 0 SubReactor waitting for register...
accpet /127.0.0.1:63223
NO 0 SubReactor waitting for register...
accpet /127.0.0.1:63226
NO 1 SubReactor waitting for register...
NO 0 /127.0.0.1:63223 -> Server： msg is 1                                                                                       
NO 1 /127.0.0.1:63226 -> Server： msg is 1                                                                                       
NO 0 /127.0.0.1:63223 -> Server： msg is 2                                                                           
NO 1 /127.0.0.1:63226 -> Server： msg is 2                                                                                       
NO 0 /127.0.0.1:63223 -> Server： msg is 3                                                                              
NO 1 /127.0.0.1:63226 -> Server： msg is 3           

总结

以上就是关于 Reactor 模型的详细介绍，相信看完的小伙伴对于 Reactor 模型也有了一定的认识，对于 Netty 的架构也更加深层次了解。下节我们继续深入 Netty 的源码。文章来源地址https://www.toymoban.com/news/detail-460239.html

到了这里，关于【Netty】Reactor 模型（十）的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！