- 模式简介
- Connection Per Thread（一个线程处理一个连接）模式
- 单线程单Reactor模式
- 回显服务器单线程单Reactor版本
- 多线程Reactor模式
- 回显服务器优化
Reactor模式由Reactor线程、Handlers处理器两大角色组成，两
大角色的职责分别如下：
- Reactor线程的职责：负责响应IO事件，并且分发到Handlers处理器
- Handlers处理器的职责：非阻塞的执行业务处理逻辑
单线程单Reactor模式对于每一个新的网络连接都分配给一个线程。每个线程都独自处理自己负责的socket连接的输入和输出。当然，服务器的监听线程也是独立的，任何socket连接的输入和输出处理都不会阻塞到后面新socket连接的监听和建立，这样服务器的吞吐量就得到了提升。早期版本的Tomcat服务器就是这样实现的。Connection Per Thread模式的缺点是对应于大量的连接，需要耗费大量的线程资源，对线程资源要求太高。而且，线程的反复创建、销毁、切换也需要代价。因此，在高并发的应用场景下，多线程OIO的缺陷是致命的。
Reactor模式有点类似事件驱动模式。在事件驱动模式中，当有事件触发时，事件源会将事件分发到Handler（处理器），由
Handler负责事件处理，单线程Reactor就是Reactor和Handlers处在一个线程中执行。在Reactor模式中有Reactor和Handler两个重要的组件：
- Reactor：负责查询IO事件，当检测到一个IO事件时将其发送给相应的Handler处理器去处理。这里的IO事件就是NIO中选择器查询出来的通道IO事件。
- Handler：与IO事件（或者选择键）绑定，负责IO事件的处理，完成真正的连接建立、通道的读取、处理业务逻辑、负责将结果写到通道等。
package nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class EchoServerReactor implements Runnable{
Selector selector;
ServerSocketChannel serverSocketChannel;
public EchoServerReactor() throws IOException {
// 1.创建选择器
selector = Selector.open();
// 2.创建通道
serverSocketChannel = ServerSocketChannel.open();
// 3.将通道设置为非阻塞
serverSocketChannel.configureBlocking(false);
// 4.绑定链接
serverSocketChannel.bind(new InetSocketAddress(18899));
SelectionKey sk = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
sk.attach(new AcceptorHandler());
}
@Override
public void run() {
try {
while(!Thread.interrupted()) {
selector.select();
Set selected = selector.selectedKeys();
Iterator it = selected.iterator();
while (it.hasNext()) {
SelectionKey sk = it.next();
dispatch(sk);
}
selected.clear();
}
} catch (IOException ex) {
ex.printStackTrace();
}
}
void dispatch(SelectionKey sk) {
Runnable handler = (Runnable) sk.attachment();
// 调用之前，附加绑定到选择键的handle对象
if (handler != null) {
handler.run();
}
}
class AcceptorHandler implements Runnable {
public void run() {
try {
SocketChannel channel = serverSocketChannel.accept();
if (channel != null)
new EchoHandler(selector, channel);
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws IOException {
new Thread(new EchoServerReactor()).start();
}
}
package nio;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
public class EchoHandler implements Runnable{
final SocketChannel channel;
final SelectionKey sk;
final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
// 处理器实例的状态：发送和接收，一个连接对应一个处理器实例
static final int RECIEVING = 0, SENDING = 1;
int state = RECIEVING;
EchoHandler(Selector selector, SocketChannel c) throws IOException {
channel = c;
c.configureBlocking(false);
// 取得选择键，再设置感兴趣的IO事件
sk = channel.register(selector, 0);
// 将Handler自身作为选择键的附件，一个连接（通道实例）对应一个处理器实例
sk.attach(this);
// 注册Read就绪事件
sk.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
public void run() {
try {
if (state == SENDING) {
// 发送状态，把数据写入连接通道
channel.write(byteBuffer);
// byteBuffer切换成写模式，写完后，就准备开始从通道读
byteBuffer.clear();
// 注册read就绪事件，开始接收客户端数据
sk.interestOps(SelectionKey.OP_READ);
// 修改状态，进入接收状态
state = RECIEVING;
} else if (state == RECIEVING) {
// 接收状态，从通道读取数据
int length = 0;
while ((length = channel.read(byteBuffer)) > 0) {
System.out.println(new String(byteBuffer.array(), 0,
length));
}
// 读完后，翻转byteBuffer的读写模式
byteBuffer.flip();
// 准备写数据到通道，注册write就绪事件
sk.interestOps(SelectionKey.OP_WRITE);
// 注册完成后，进入发送状态
state = SENDING;
}
// 处理结束了，这里不能关闭select key，需要重复使用
// sk.cancel();
} catch (IOException ex) {
ex.printStackTrace();
}
}
}
在单线程Reactor模式中，Reactor和Handler都在同一条线程中执行。这样当其中某个Handler阻塞时，会导致其他所有的Handler都得不到执行。在这种场景下，被阻塞的Handler不仅仅负责输入和输出处理的传输处理器，还包括负责新连接监听的AcceptorHandler处理器，可能导致服务器无响应。这是一个非常严重的缺陷，导致单线程反应器模型在生产场景中使用得比较少。
多线程Reactor模式
多线程Reactor的演进的两个方面：
- 升级Handler。既要使用多线程，又要尽可能高效率，则可以考虑使用线程池。
将负责数据传输处理的IOHandler处理器的执行放入独立的线程池中。这样，业务处理线程与负责新连接监听的反应器线程就能相互隔离，避免服务器的连接监听受到阻塞。
- 升级Reactor。可以考虑引入多个Selector（选择器），提升选择大量通道的能力。
回显服务器优化如果服务器为多核的CPU，可以将反应器线程拆分为多个子反应器（SubReactor）线程；同时，引入多个选择器，并且为每一个SubReactor引入一个线程，一个线程负责一个选择器的事件轮询。这样充分释放了系统资源的能力，也大大提升了反应器管理大量连接或者监听大量传输通道的能力。
package nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
public class MultiThreadEchoServerReactor {
ServerSocketChannel serverSocketChannel;
AtomicInteger next = new AtomicInteger(0);
// selectors集合,引入多个selector选择器
Selector[] selectors = new Selector[2];
SubReactor[] subReactors = null;
MultiThreadEchoServerReactor() throws IOException {
// 初始化多个selector选择器
selectors[0] = Selector.open();
selectors[1] = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
InetSocketAddress address = new InetSocketAddress("127.0.0.1", 18899);
serverSocketChannel.socket().bind(address);
// 非阻塞
serverSocketChannel.configureBlocking(false);
// 第一个selector, 负责监控新连接事件
SelectionKey sk = serverSocketChannel.register(selectors[0], SelectionKey.OP_ACCEPT);
// 将新连接处理器AcceptorHandler附加到SelectionKey（选择键）
sk.attach(new AcceptorHandler());
// 第一个子反应器，一子反应器负责一个选择器
SubReactor subReactor1 = new SubReactor(selectors[0]);
// 第二个子反应器，一子反应器负责一个选择器
SubReactor subReactor2 = new SubReactor(selectors[1]);
subReactors = new SubReactor[]{subReactor1, subReactor2};
}
private void startService() {
// 一子反应器对应一条线程
new Thread(subReactors[0]).start();
new Thread(subReactors[1]).start();
}
class SubReactor implements Runnable {
// 每条线程负责一个选择器的查询
final Selector selector;
public SubReactor(Selector selector) {
this.selector = selector;
}
public void run() {
try {
while (!Thread.interrupted()) {
selector.select();
Set keySet = selector.selectedKeys();
Iterator it = keySet.iterator();
while (it.hasNext()) {
//Reactor负责dispatch收到的事件
SelectionKey sk = it.next();
dispatch(sk);
}
keySet.clear();
}
} catch (IOException ex) {
ex.printStackTrace();
}
}
void dispatch(SelectionKey sk) {
Runnable handler = (Runnable) sk.attachment();
// 调用之前attach绑定到选择键的handler处理器对象
if (handler != null) {
// 执行了run方法，由于引进了线程池和多线程，业务处理线程之间会隔离，不会受到阻塞，而新连接处理线程和业务处理线程由不同的SubReactor负责轮询，不会互相干扰
handler.run();
}
}
}
class AcceptorHandler implements Runnable {
public void run() {
try {
SocketChannel channel = serverSocketChannel.accept();
if (channel != null)
new MultiThreadEchoHandler(selectors[next.get()], channel);
} catch (IOException e) {
e.printStackTrace();
}
if (next.incrementAndGet() == selectors.length) {
next.set(0);
}
}
}
public static void main(String[] args) throws IOException {
MultiThreadEchoServerReactor server = new MultiThreadEchoServerReactor();
server.startService();
}
}
package nio;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class MultiThreadEchoHandler {
final SocketChannel channel;
final SelectionKey sk;
final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
static final int RECIEVING = 0, SENDING = 1;
int state = RECIEVING;
// 引入线程池
static ExecutorService pool = Executors.newFixedThreadPool(4);
MultiThreadEchoHandler(Selector selector, SocketChannel c) throws IOException {
channel = c;
c.configureBlocking(false);
// 仅仅取得选择键，后设置感兴趣的IO事件
sk = channel.register(selector, 0);
// 将本Handler作为sk选择键的附件，方便事件dispatch
sk.attach(this);
// 向sk选择键注册Read就绪事件
sk.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
public void run() {
// 异步任务，在独立的线程池中执行
pool.execute(new AsyncTask());
}
// 异步任务，不在Reactor线程中执行
public synchronized void asyncRun() {
try {
if (state == SENDING) {
// 写入通道
channel.write(byteBuffer);
// 写完后,准备开始从通道读,byteBuffer切换成写模式
byteBuffer.clear();
// 写完后,注册read就绪事件
sk.interestOps(SelectionKey.OP_READ);
// 写完后,进入接收的状态
state = RECIEVING;
} else if (state == RECIEVING) {
// 从通道读
int length = 0;
while ((length = channel.read(byteBuffer)) > 0) {
System.out.println(new String(byteBuffer.array(), 0, length));
}
// 读完后，准备开始写入通道,byteBuffer切换成读模式
byteBuffer.flip();
// 读完后，注册write就绪事件
sk.interestOps(SelectionKey.OP_WRITE);
// 读完后,进入发送的状态
state = SENDING;
}
// 处理结束了, 这里不能关闭select key，需要重复使用
// sk.cancel();
} catch (IOException ex) {
ex.printStackTrace();
}
}
// 异步任务的内部类
class AsyncTask implements Runnable {
public void run() {
MultiThreadEchoHandler.this.asyncRun();
}
}
}
