Java NIO 反应堆模式简单模型
一般nio里反应堆模式都是这样：一个acceptor（当然多个也行，不过一般场景一个够了）负责accept事件，把接收到socket channel注册到按某种算法从reactor池中取出的一个reactor上，注册的事件为读，写等，之后这个socket channel的所有io事件都和acceptor没关系，都由被注册到的那个reactor来负责。
每个Acceptor和每个Reactor都各自持有一个Selector
当然每个Acceptor和Reactor都得是一个线程（起码在逻辑上得是线程）
简单实现，三个类NioAcceptor、NioReactor和ReactorPool：
中国工商网电子商务购物中心系统EMall
下载
完全公开源代码，并无任何许可限制 特别基于大型电子商务网站的系统开发 Microsoft SQL Server 2000后台数据库，充分应用了存储过程的巨大功效 基于类模块的扩展数据访问能力支持任何类型的大型数据库 加密用户登录信息(cookie) 易于安装的系统和应用功能 100%的asp.net的代码，没有COM，java或者其他的格式 完全基于MS建议的系统安全设计 最佳的应用程序,数据库
package cc.lixiaohui.demo.dp.reator;
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.Objects;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Acceptor负责处理SelectionKey.OP_ACCEPT事件, 将接收到的SocketChannel注册到Reactor上去
*/
public class NioAcceptor {
private int port;
private String host;
private Selector selector; // Java NIO Selector
private final ServerSocketChannel serverChannel; // Java NIO ServerSocketChannel
private ReactorPool reactorPool; // NioReactor池
private Thread thread; // 工作线程
private volatile boolean stop = false;
private static final Logger logger = LoggerFactory.getLogger(NioAcceptor.class);
public NioAcceptor(int port, String host, int reactorPoolSize) throws IOException {
this.port = port;
this.host = Objects.requireNonNull(host);
this.reactorPool = new ReactorPool(reactorPoolSize);
selector = Selector.open(); // 创建selector
serverChannel = ServerSocketChannel.open(); // new server socket channel
serverChannel.configureBlocking(false); // in non-blocking mode
serverChannel.bind(new InetSocketAddress(host, port)); // bind
serverChannel.register(selector, SelectionKey.OP_ACCEPT); //
}
public void stop() throws InterruptedException {
stop = true;
thread.join();
}
public void start() {
thread = new Thread(new AcceptTask(this));
thread.start();
}
private static class AcceptTask implements Runnable {
NioAcceptor acceptor;
AcceptTask(NioAcceptor acceptor) {
this.acceptor = acceptor;
}
public void run() {
final Selector selector = acceptor.selector;
Set keys = null;
while (!acceptor.stop) { // 运行中
try {
selector.select(1000L); // select, 最多等1秒
keys = selector.selectedKeys();
try {
for (SelectionKey key : keys) {
if (key.isValid() && key.isAcceptable()) { // 可accept
SocketChannel channel = acceptor.serverChannel.accept();
channel.configureBlocking(false);
// 取下一个Reactor并把SocketChannel加入到Reactor的注册队列
acceptor.reactorPool.nextReactor().postRegistry(channel);
} else {
key.cancel();
}
}
} finally {
keys.clear();
}
} catch (IOException e) {
logger.error("", e);
}
}
}
}
} /**
* Reactor负责SelectionKey.OP_READ | SelectionKey.OP_WRITE等事件
*/
public class NioReactor {
/** 待注册的{@link SocketChannel} 队列 */
private Queue registerQueue = new ConcurrentLinkedQueue();
private Selector selector;
private volatile boolean stop = false;
private Thread thread;
private static final Logger logger = LoggerFactory.getLogger(NioReactor.class);
public NioReactor() throws IOException {
selector = Selector.open();
}
public void postRegistry(SocketChannel channel) {
registerQueue.add(channel);
selector.wakeup(); // 唤醒selector, 以便让其即时处理注册
}
public NioReactor start() {
thread = new Thread(new ReactTask(this));
thread.start();
return this;
}
public void stop() throws InterruptedException {
stop = true;
thread.join();
}
/**
* 处理队列里面的待注册的SocketChannel
*/
private void doRegister(Selector selector) {
while (!registerQueue.isEmpty()) {
SocketChannel channel = registerQueue.poll();
try {
// 注册读事件, 写事件无需注册, 写事件是业务驱动的, 当往channel写入 数据未写完时再注册写事件
channel.register(selector, SelectionKey.OP_READ);
} catch (ClosedChannelException e) {
logger.error("", e);
}
}
}
private void handleWrite(SelectionKey key) {
// TODO 业务写
}
private void handleRead(SelectionKey key) {
// TODO 业务读
}
private static class ReactTask implements Runnable {
NioReactor reactor;
ReactTask(NioReactor reactor) {
this.reactor = reactor;
}
public void run() {
Set keys = null;
while (!reactor.stop) {
final Selector selector = reactor.selector;
try {
selector.select(500L);
reactor.doRegister(selector); // 处理注册
keys = selector.selectedKeys();
for (SelectionKey key : keys) {
try {
if (!key.isValid()) { // not valid
key.cancel();
continue;
}
if (key.isReadable()) { // 可读
reactor.handleRead(key);
}
if (key.isWritable()) { // 可写
reactor.handleWrite(key);
}
} catch (Throwable t) {
logger.error("", t);
continue;
}
}
} catch (IOException e) {
logger.error("", e);
}
}
}
}
} ReactorPool用来管理Reactor：
立即学习“Java免费学习笔记（深入）”；
public class ReactorPool extends LinkedList{ private static final long serialVersionUID = 6525233920805533099L; private final int capacity; public ReactorPool(int size) { this.capacity = size; } // 轮询算法取下一个Reactor public NioReactor nextReactor() throws IOException { // 新建或从头部拿一个Reactor NioReactor reactor = size() < capacity ? new NioReactor().start() : poll(); add(reactor);// 加到尾部 return reactor; } }
