《Reactor详解》一步一步带你构建后端框架
目录
总结 送学习资料（包含视频、面试、面试题、技术学习路线图谱、文档等）
1.框架的构建：多个技术点的平衡再选择；
2. 框架的实现：合理的拆分以及理解本质后的抽象；
3. 网络模块实现：reactor 与 协程融合形成同步非阻塞；
4. 连接池实现：异步连接 与 同步连接池的实现方式；
一、Reactor 是什么
详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。
关于re。actor 是什么，我们先从wiki上看下:
The reactor design pattern is an event handling pattern for handling service requests delivered concurrently to a service handler by one or more inputs. The service handler then demultiplexes the incoming requests and dispatches them synchronously to the associated request handlers.
从上述文字中我们可以看出以下关键点 ：
事件驱动（event handling）
可以处理一个或多个输入源（one or more inputs）
通过Service Handler同步的将输入事件（Event）采用多路复用分发给相应的Request Handler（多个）处理
自POSA2 中的关于Reactor Pattern 介绍中，我们了解了Reactor 的处理方式:
同步的等待多个事件源到达（采用select()实现）
将事件多路分解以及分配相应的事件服务进行处理，这个分派采用server集中处理（dispatch）
分解的事件以及对应的事件服务应用从分派服务中分离出去（handler）
关于Reactor Pattern 的OMT 类图设计:
详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。
二、为何要用Reactor
常见的网络服务中，如果每一个客户端都维持一个与登陆服务器的连接。那么服务器将维护多个和客户端的连接以出来和客户端的contnect 、read、write ，特别是对于长链接的服务，有多少个c端，就需要在s端维护同等的IO连接。这对服务器来说是一个很大的开销。
1. 举个例子吧
??假设我们需要建立一个提供分布式日志服务的事件驱动服务器。客户们使用客户端会向该服务器发送请求记录自己的状态信息，信息包括错误通知、debug信息、表现诊断等。该日志服务器对于收到的信息进行分类并分发操作，包括：显示屏显示、打印机打印、数据库存储等，如下图所示。
详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。
客户端和服务器之间通信协议通常选用TCP等面向连接的协议（为了保证可靠性），通过IP和端口的四元组来认定客户端和服务器。日志服务器被多个客户端同时使用，为此该服务器需要保证多用户连接请求和日志记录的并发性。为了保证这个基本要求，不难想到使用多线程去实现该服务器，每一个线程专门针对一个连接。
??然而，该设计存在以下的问题：
（1） 效率问题。
??多线程导致的上下文切换、同步、数据移动等可能带来效率的下降。
（2） 编程的简单性。
??多线程需要考虑复杂的并发设计，包括线程安全等。
（3）可移植性
??多线程在不同OS下的不同（或者没有）影响了可移植性。
??由于以上问题，多线程设计往往既不是最高效也不是最易实现的方案。由此我们需要别的方案来实现可以处理并发请求的服务器（推销自己的reactor）。
三、Reactor
在应用Java NIO构建Reactor Pattern中，大神 Doug Lea（让人无限景仰的java 大神）在“Scalable IO in Java”中给了很好的阐述。我们采用大神介绍的3种Reactor 来分别介绍。
首先我们基于Reactor Pattern 处理模式中，定义以下三种角色:
Reactor 将I/O事件分派给对应的Handler
Acceptor 处理客户端新连接，并分派请求到处理器链中
Handlers 执行非阻塞读/写 任务
1、单Reactor单线程模型
详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。
我们看代码的实现方式:
/**
* 等待事件到来，分发事件处理
*/
class Reactor implements Runnable {
private Reactor() throws Exception {
?
SelectionKey sk =
serverSocket.register(selector,
SelectionKey.OP_ACCEPT);
// attach Acceptor 处理新连接
sk.attach(new Acceptor());
}
?
public void run() {
try {
while (!Thread.interrupted()) {
selector.select();
Set selected = selector.selectedKeys();
Iterator it = selected.iterator();
while (it.hasNext()) {
it.remove();
//分发事件处理
dispatch((SelectionKey) (it.next()));
}
}
} catch (IOException ex) {
//do something
}
}
?
void dispatch(SelectionKey k) {
// 若是连接事件获取是acceptor
// 若是IO读写事件获取是handler
Runnable runnable = (Runnable) (k.attachment());
if (runnable != null) {
runnable.run();
}
}
?
}
/**
* 连接事件就绪,处理连接事件
*/
class Acceptor implements Runnable {
@Override
public void run() {
try {
SocketChannel c = serverSocket.accept();
if (c != null) {// 注册读写
new Handler(c, selector);
}
} catch (Exception e) {
?
}
}
}
这是最基本的单Reactor单线程模型。其中Reactor线程，负责多路分离套接字，有新连接到来触发connect 事件之后，交由Acceptor进行处理，有IO读写事件之后交给hanlder 处理。
Acceptor主要任务就是构建handler ，在获取到和client相关的SocketChannel之后 ，绑定到相应的hanlder上，对应的SocketChannel有读写事件之后，基于racotor 分发,hanlder就可以处理了（所有的IO事件都绑定到selector上，有Reactor分发）。
2、单Reactor多线程模型
详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。
相对于第一种单线程的模式来说，在处理业务逻辑，也就是获取到IO的读写事件之后，交由线程池来处理，这样可以减小主reactor的性能开销，从而更专注的做事件分发工作了，从而提升整个应用的吞吐。
我们看下实现方式：
/**
* 多线程处理读写业务逻辑
*/
class MultiThreadHandler implements Runnable {
public static final int READING = 0, WRITING = 1;
int state;
final SocketChannel socket;
final SelectionKey sk;
?
//多线程处理业务逻辑
ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
?
?
public MultiThreadHandler(SocketChannel socket, Selector sl) throws Exception {
this.state = READING;
this.socket = socket;
sk = socket.register(selector, SelectionKey.OP_READ);
sk.attach(this);
socket.configureBlocking(false);
}
?
@Override
public void run() {
if (state == READING) {
read();
} else if (state == WRITING) {
write();
}
}
?
private void read() {
//任务异步处理
executorService.submit(() -> process());
?
//下一步处理写事件
sk.interestOps(SelectionKey.OP_WRITE);
this.state = WRITING;
}
?
private void write() {
//任务异步处理
executorService.submit(() -> process());
?
//下一步处理读事件
sk.interestOps(SelectionKey.OP_READ);
this.state = READING;
}
?
/**
* task 业务处理
*/
public void process() {
//do IO ,task,queue something
}
}
3、多Reactor多线程模型
详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。
第三种模型比起第二种模型，是将Reactor分成两部分，
mainReactor负责监听server socket，用来处理新连接的建立，将建立的socketChannel指定注册给subReactor。
subReactor维护自己的selector, 基于mainReactor 注册的socketChannel多路分离IO读写事件，读写网 络数据，对业务处理的功能，另其扔给worker线程池来完成。
我们看下实现方式:
/**
* 多work 连接事件Acceptor,处理连接事件
*/
class MultiWorkThreadAcceptor implements Runnable {
?
// cpu线程数相同多work线程
int workCount =Runtime.getRuntime().availableProcessors();
SubReactor[] workThreadHandlers = new SubReactor[workCount];
volatile int nextHandler = 0;
?
public MultiWorkThreadAcceptor() {
this.init();
}
?
public void init() {
nextHandler = 0;
for (int i = 0; i < workThreadHandlers.length; i++) {
try {
workThreadHandlers[i] = new SubReactor();
} catch (Exception e) {
}
?
}
}
?
@Override
public void run() {
try {
SocketChannel c = serverSocket.accept();
if (c != null) {// 注册读写
synchronized (c) {
// 顺序获取SubReactor，然后注册channel
SubReactor work = workThreadHandlers[nextHandler];
work.registerChannel(c);
nextHandler++;
if (nextHandler >= workThreadHandlers.length) {
nextHandler = 0;
}
}
}
} catch (Exception e) {
}
}
}
/**
* 多work线程处理读写业务逻辑
*/
class SubReactor implements Runnable {
final Selector mySelector;
?
//多线程处理业务逻辑
int workCount =Runtime.getRuntime().availableProcessors();
ExecutorService executorService = Executors.newFixedThreadPool(workCount);
?
?
public SubReactor() throws Exception {
// 每个SubReactor 一个selector
this.mySelector = SelectorProvider.provider().openSelector();
}
?
/**
* 注册chanel
*
* @param sc
* @throws Exception
*/
public void registerChannel(SocketChannel sc) throws Exception {
sc.register(mySelector, SelectionKey.OP_READ | SelectionKey.OP_CONNECT);
}
?
@Override
public void run() {
while (true) {
try {
//每个SubReactor 自己做事件分派处理读写事件
selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isReadable()) {
read();
} else if (key.isWritable()) {
write();
}
}
?
} catch (Exception e) {
?
}
}
}
?
private void read() {
//任务异步处理
executorService.submit(() -> process());
}
?
private void write() {
//任务异步处理
executorService.submit(() -> process());
}
?
/**
* task 业务处理
*/
public void process() {
//do IO ,task,queue something
}
}
四、事件处理模式
在 Douglas Schmidt 的大作《POSA2》中有关于事件处理模式的介绍，其中有四种事件处理模式:
Reactor
Proactor
Asynchronous Completion Token
Acceptor-Connector
1.Proactor
本文介绍的Reactor就是其中一种，而Proactor的整体结构和reacotor的处理方式大同小异，不同的是Proactor采用的是异步非阻塞IO的方式实现，对数据的读写由异步处理，无需用户线程来处理，服务程序更专注于业务事件的处理，而非IO阻塞。
2.Asynchronous Completion Token
简单来说，ACT就是应对应用程序异步调用服务操作，并处理相应的服务完成事件。从token这个字面意思，我们大概就能了解到，它是一种状态的保持和传递。
比如，通常应用程序会有调用第三方服务的需求，一般是业务线程请求都到，需要第三方资源的时候，去同步的发起第三方请求，而为了提升应用性能，需要异步的方式发起请求，但异步请求的话，等数据到达之后，此时的我方应用程序的语境以及上下文信息已经发生了变化，你没办法去处理。
ACT 解决的就是这个问题，采用了一个token的方式记录异步发送前的信息，发送给接受方，接受方回复的时候再带上这个token，此时就能恢复业务的调用场景。
总结；
上图中我们可以看到在client processing 这个阶段，客户端是可以继续处理其他业务逻辑的，不是阻塞状态。service 返回期间会带上token信息。
详细教程资料+课件 关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题 资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式 等。