社区微信群开通啦，扫一扫抢先加入社区官方微信群
社区微信群
原文地址：
http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf
《Scalable IO in Java》是java.util.concurrent包的作者，大师Doug Lea关于分析与构建可伸缩的高性能IO服务的一篇经典文章，在文章中Doug Lea通过各个角度，循序渐进的梳理了服务开发中的相关问题，以及在解决问题的过程中服务模型的演变与进化，文章中基于Reactor反应器模式的几种服务模型架构，也被Netty、Mina等大多数高性能IO服务框架所采用，因此阅读这篇文章有助于你更深入了解Netty、Mina等服务框架的编程思想与设计模式。
基本上所有的网络处理程序都有以下基本的处理过程:
Read request
Decode request
Process service
Encode reply
Send reply
1.传统的服务设计模式
每一个连接的处理都会对应分配一个新的线程，下面我们看一段经典的Server端Socket服务代码：
class Server implements Runnable {
public void run() {
try {
ServerSocket ss = new ServerSocket(PORT);
while (!Thread.interrupted())
new Thread(new Handler(ss.accept())).start(); //创建新线程来handle
// or, single-threaded, or a thread pool
} catch (IOException ex) { /* ... */ }
}
static class Handler implements Runnable {
final Socket socket;
Handler(Socket s) { socket = s; }
public void run() {
try {
byte[] input = new byte[MAX_INPUT];
socket.getInputStream().read(input);
byte[] output = process(input);
socket.getOutputStream().write(output);
} catch (IOException ex) { /* ... */ }
}
private byte[] process(byte[] cmd) { /* ... */ }
}
}
对于每一个请求都分发给一个线程，每个线程中都独自处理上面的流程。
这种模型由于IO在阻塞时会一直等待，因此在用户负载增加时，性能下降的非常快。
server导致阻塞的原因：
1、serversocket的accept方法，阻塞等待client连接，直到client连接成功。
2、线程从socket inputstream读入数据，会进入阻塞状态，直到全部数据读完。
3、线程向socket outputstream写入数据，会阻塞直到全部数据写完。
client导致阻塞的原因：
1、client建立连接时会阻塞，直到连接成功。
2、线程从socket输入流读入数据，如果没有足够数据读完会进入阻塞状态，直到有数据或者读到输入流末尾。
3、线程从socket输出流写入数据，直到输出所有数据。
4、socket.setsolinger()设置socket的延迟时间，当socket关闭时，会进入阻塞状态，直到全部数据都发送完或者超时。
改进：采用基于事件驱动的设计，当有事件触发时，才会调用处理器进行数据处理。
2. 构建高性能可伸缩的IO服务
在构建高性能可伸缩IO服务的过程中，我们希望达到以下的目标：
① 能够在海量负载连接情况下优雅降级；
② 能够随着硬件资源的增加，性能持续改进；
③ 具备低延迟、高吞吐量、可调节的服务质量等特点；
而分发处理就是实现上述目标的一个最佳方式。
3、分发模式
分发模式具有以下几个机制：
① 将一个完整处理过程分解为一个个细小的任务；
② 每个任务执行相关的动作且不产生阻塞；
③ 在任务执行状态被触发时才会去执行，例如只在有数据时才会触发读操作；
在一般的服务开发当中，IO事件通常被当做任务执行状态的触发器使用，在hander处理过程中主要针对的也就是IO事件；
java.nio包就很好的实现了上述的机制：
① 非阻塞的读和写
② 通过感知IO事件分发任务的执行
所以结合一系列基于事件驱动模式的设计，给高性能IO服务的架构与设计带来丰富的可扩展性；
基于事件驱动的架构设计通常比其他架构模型更加有效，因为可以节省一定的性能资源，事件驱动模式下通常不需要为每一个客户端建立一个线程，这意味这更少的线程开销，更少的上下文切换和更少的锁互斥，但任务的调度可能会慢一些，而且通常实现的复杂度也会增加，相关功能必须分解成简单的非阻塞操作，类似与GUI的事件驱动机制，当然也不可能把所有阻塞都消除掉，特别是GC， page faults(内存缺页中断)等。由于是基于事件驱动的，所以需要跟踪服务的相关状态（因为你需要知道什么时候事件会发生）;
下图是AWT中事件驱动设计的一个简单示意图，可以看到，在不同的架构设计中的基于事件驱动的IO操作使用的基本思路是一致的；
Reactor也可以称作反应器模式，它有以下几个特点：
① Reactor模式中会通过分配适当的handler(处理程序)来响应IO事件，类似与AWT 事件处理线程；
② 每个handler执行非阻塞的操作，类似于AWT ActionListeners 事件监听
③ 通过将handler绑定到事件进行管理，类似与AWT addActionListener 添加事件监听；
1、单线程模式
下图展示的就是单线程下基本的Reactor设计模式
Reactor单线程模式示例
class Reactor implements Runnable {
final Selector selector;
final ServerSocketChannel serverSocket;
Reactor(int port) throws IOException { //Reactor初始化
selector = Selector.open();
serverSocket = ServerSocketChannel.open();
serverSocket.socket().bind(new InetSocketAddress(port));
serverSocket.configureBlocking(false); //非阻塞
SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT); //分步处理,第一步,接收accept事件
sk.attach(new Acceptor()); //attach callback object, Acceptor
}
public void run() {
try {
while (!Thread.interrupted()) {
selector.select();
Set selected = selector.selectedKeys();
Iterator it = selected.iterator();
while (it.hasNext())
dispatch((SelectionKey)(it.next()); //Reactor负责dispatch收到的事件
selected.clear();
}
} catch (IOException ex) { /* ... */ }
}
void dispatch(SelectionKey k) {
Runnable r = (Runnable)(k.attachment()); //调用之前注册的callback对象
if (r != null)
r.run();
}
class Acceptor implements Runnable { // inner
public void run() {
try {
SocketChannel c = serverSocket.accept();
if (c != null)
new Handler(selector, c);
}
catch(IOException ex) { /* ... */ }
}
}
}
final class Handler implements Runnable {
final SocketChannel socket;
final SelectionKey sk;
ByteBuffer input = ByteBuffer.allocate(MAXIN);
ByteBuffer output = ByteBuffer.allocate(MAXOUT);
static final int READING = 0, SENDING = 1;
int state = READING;
Handler(Selector sel, SocketChannel c) throws IOException {
socket = c; c.configureBlocking(false);
// Optionally try first read now
sk = socket.register(sel, 0);
sk.attach(this); //将Handler作为callback对象
sk.interestOps(SelectionKey.OP_READ); //第二步,接收Read事件
sel.wakeup();
}
boolean inputIsComplete() { /* ... */ }
boolean outputIsComplete() { /* ... */ }
void process() { /* ... */ }
public void run() {
try {
if (state == READING) read();
else if (state == SENDING) send();
} catch (IOException ex) { /* ... */ }
}
void read() throws IOException {
socket.read(input);
if (inputIsComplete()) {
process();
state = SENDING;
// Normally also do first write now
sk.interestOps(SelectionKey.OP_WRITE); //第三步,接收write事件
}
}
void send() throws IOException {
socket.write(output);
if (outputIsComplete()) sk.cancel(); //write完就结束了, 关闭select key
}
}
//上面 的实现用Handler来同时处理Read和Write事件, 所以里面出现状态判断
//我们可以用State-Object pattern来更优雅的实现
class Handler { // ...
public void run() { // initial state is reader
socket.read(input);
if (inputIsComplete()) {
process();
sk.attach(new Sender()); //状态迁移, Read后变成write, 用Sender作为新的callback对象
sk.interest(SelectionKey.OP_WRITE);
sk.selector().wakeup();
}
}
class Sender implements Runnable {
public void run(){ // ...
socket.write(output);
if (outputIsComplete()) sk.cancel();
}
}
}
关于Reactor模式的一些概念：
Reactor：负责响应IO事件，当检测到一个新的事件，将其发送给相应的Handler去处理。
Handler：负责处理非阻塞的行为，标识系统管理的资源；同时将handler与事件绑定。
Reactor为单个线程，需要处理accept连接，同时发送请求到处理器中。
由于只有单个线程，所以处理器中的业务需要能够快速处理完。
2、多线程设计模式
在多处理器场景下，为实现服务的高性能我们可以有目的的采用多线程模式：
1、增加Worker线程，专门用于处理非IO操作，因为通过上面的程序我们可以看到，反应器线程需要迅速触发处理流程，而如果处理过程也就是process()方法产生阻塞会拖慢反应器线程的性能，所以我们需要把一些非IO操作交给Woker线程来做；
2、拆分并增加反应器Reactor线程，一方面在压力较大时可以饱和处理IO操作，提高处理能力；另一方面维持多个Reactor线程也可以做负载均衡使用；线程的数量可以根据程序本身是CPU密集型还是IO密集型操作来进行合理的分配；
2.1 多线程模式
Reactor多线程设计模式具备以下几个特点：
① 通过卸载非IO操作来提升Reactor 线程的处理性能，这类似与POSA2 中Proactor的设计；
② 比将非IO操作重新设计为事件驱动的方式更简单；
③ 但是很难与IO重叠处理，最好能在第一时间将所有输入读入缓冲区；（这里我理解的是最好一次性读取缓冲区数据，方便异步非IO操作处理数据）
④ 可以通过线程池的方式对线程进行调优与控制，一般情况下需要的线程数量比客户端数量少很多；
下面是Reactor多线程设计模式的一个示意图与示例代码（我们可以看到在这种模式中在Reactor线程的基础上把非IO操作放在了Worker线程中执行）：
多线程模式示例：
class Handler implements Runnable {
// uses util.concurrent thread pool
static PooledExecutor pool = new PooledExecutor(...);
static final int PROCESSING = 3;
// ...
synchronized void read() { // ...
socket.read(input);
if (inputIsComplete()) {
state = PROCESSING;
pool.execute(new Processer()); //使用线程pool异步执行
}
}
synchronized void processAndHandOff() {
process();
state = SENDING; // or rebind attachment
sk.interest(SelectionKey.OP_WRITE); //process完,开始等待write事件
}
class Processer implements Runnable {
public void run() {
processAndHandOff();
}
}
}
当你把非IO操作放到线程池中运行时，你需要注意以下几点问题：
① 任务之间的协调与控制，每个任务的启动、执行、传递的速度是很快的，不容易协调与控制；
② 每个hander中dispatch的回调与状态控制；
③ 不同线程之间缓冲区的线程安全问题；
④ 需要任务返回结果时，任务线程等待和唤醒状态间的切换；
为解决上述问题可以使用PooledExecutor线程池框架，这是一个可控的任务线程池，主函数采用execute(Runnable r)，它具备以下功能，可以很好的对池中的线程与任务进行控制与管理：
① 可设置线程池中最大与最小线程数；
② 按需要判断线程的活动状态，及时处理空闲线程；
③ 当执行任务数量超过线程池中线程数量时，有一系列的阻塞、限流的策略；
2.2 基于多个反应器的多线程模式
这是对上面模式的进一步完善，使用反应器线程池，一方面根据实际情况用于匹配调节CPU处理与IO读写的效率，提高系统资源的利用率，另一方面在静态或动态构造中每个反应器线程都包含对应的Selector,Thread,dispatchloop,下面是一个简单的代码示例与示意图（Netty就是基于这个模式设计的，一个处理Accpet连接的mainReactor线程，多个处理IO事件的subReactor线程）：
参考代码：
Selector[] selectors; //subReactors集合, 一个selector代表一个subReactor
int next = 0;
class Acceptor { // ...
public synchronized void run() { ...
Socket connection = serverSocket.accept(); //主selector负责accept
if (connection != null)
new Handler(selectors[next], connection); //选个subReactor去负责接收到的connection
if (++next == selectors.length) next = 0;
}
}
感谢:
https://yq.aliyun.com/articles/50466
http://www.mamicode.com/info-detail-2736833.html
https://www.cnblogs.com/luxiaoxun/archive/2015/03/11/4331110.html
如果觉得我的文章对您有用，请随意打赏。你的支持将鼓励我继续创作！
