java 反应堆模式_Java NIO 反应堆模式
Java NIO 反应堆模式简单模型
一般NIO里反应堆模式都是这样:一个Acceptor(当然多个也行,不过一般场景一个够了)负责accept事件,把接收到Socket CHannel注册到按某种算法从Reactor池中取出的一个Reactor上,注册的事件为读,写等,之后这个Socket Channel的所有IO事件都和Acceptor没关系,都由被注册到的那个Reactor来负责。
每个Acceptor和每个Reactor都各自持有一个Selector
当然每个Acceptor和Reactor都得是一个线程(起码在逻辑上得是线程)
简单实现,三个类NioAcceptor、NioReactor和ReactorPool:
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;
Setkeys = 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 {
/** [email protected] SocketChannel} 队列 */
private QueueregisterQueue = 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() {
Setkeys = 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:
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;
}
}
参考:
Netty的NIO模型
Mycat的NIO实现
java 反应堆模式_Java NIO 反应堆模式相关推荐
- java简单工厂模式_Java 简单工厂模式
Java 简单工厂模式 2014-05-28·WeaponX 3717 次浏览 ## 介绍 简单工厂模式又称静态工厂模式. 简单工厂模式由工厂类角色.抽象产品角色和具体产品角色组成. 工厂类角色是本模 ...
- java抽象工厂模式_Java 抽象工厂模式
Java设计模式 - 抽象工厂模式 抽象工厂模式是另一个创建模式. 抽象工厂模式,也称为工厂的工厂,有一个工厂创建其他工厂. 当使用抽象工厂模式时,我们首先使用超级工厂创建工厂,然后使用创建的工厂创建 ...
- java改写模式_Java基于状态模式实现的文档编辑模式切换功能实例
本文实例讲述了Java基于状态模式实现的文档编辑模式切换功能.分享给大家供大家参考,具体如下: 一 模式定义 状态模式:当一个对象的内在状态改变时允许改变其行为,这个对象看起来就像是改变了其类. 二 ...
- java创建型_Java创建型模式
Java创建型模式 在软件工程中,创建型模式是处理对象创建的设计模式,试图根据实际情况使用合适的方式创建对象.基本的对象创建方式可能会导致设计上的问题,或增加设计的复杂度.创建型模式通过以某种方式控制 ...
- java设计模式建造_java设计模式-建造者模式
建造者模式:将一个复杂对象的构建与它的表示分离,使得同样的构建过程可以创造不同的表示. 建造者模式将一个产品的内部表象和产品的生成过程分割开来,建造对象的时候只需要指定需要建造的类型,而不必了解具体的 ...
- java 管理员命令模式_java中命令模式详解和使用方法
命令模式(Command Pattern)是一种数据驱动的设计模式,它属于行为型模式.请求以命令的形式包裹在对象中,并传给调用对象.调用对象寻找可以处理该命令的合适的对象,并把该命令传给相应的对象,该 ...
- java设计模式 组合_JAVA 设计模式 组合模式
用途组合模式 (Component) 将对象组合成树形结构以表示"部分-整体"的层次结构. 组合模式使得用户对单个对象和组合对象的使用具有唯一性. 组合模式是一种结构型模式. 结构 ...
- java 装饰者_Java装饰者模式
Hello,即将要写的这篇文章,是对刚刚学习完的装饰者模式的一个总结啦,也不是什么技术难点,就是巩固一下啦. 一.装饰者模式的定义 装饰者模式动态的将责任附加到对象上.若要扩展功能,装饰者提供了比继承 ...
- java netty 教程_Java NIO框架Netty教程(十六)
该图是OneCoder通过阅读Netty源码,逐渐记录下来的.基本可以说明Netty服务的启动流程.这里在具体讲解一下. 首先说明,我们这次顺利的流程是基于NioSocketServer的.也就是基于 ...
- 组合模式_Java设计模式-组合模式
在现实生活中,存在很多"部分-整体"的关系,例如,大学中的部门与学院.总公司中的部门与分公司.学习用品中的书与书包.生活用品中的衣月艮与衣柜以及厨房中的锅碗瓢盆等.在软件开发中也是 ...
最新文章
- 2021CCPC河北省省赛F题(河南省CCPC测试赛重现)
- 重构——解决过长参数列表(long parameter list)
- Module build failed (from ./node_modules/babel-loader/lib/index.js): Error: [BABEL] D:\moblie\web_co
- 你以为环幕屏就结束了?真正的小米MIX 4或下月发布:1亿像素相机加持
- Error:scalac: missing or invalid dependency detected while loading class file ‘RDDOperationScope.cla
- cli命令行配置路由器_2.3.3 使用CLI执行基本路由器配置
- the jacobian matrix and the gradient matrix
- [Bzoj2039][2009国家集训队]employ人员雇佣(最小割)
- 英文地址以及英文地址中的缩写都是什么意思
- 灵异:各国家千年诅咒悬迷——不可思议
- 微信小程序弹框滑动冒泡,外层页面也被滑动问题
- SQL-按日期不间断统计
- css-浮动布局与样式规则
- 15张超详细的Python学习路线图,纯良心分享,零基础学习宝典
- 基于Visual Studio2010开发office2010办公自动化应用(12)自定义VisioAddIn插件
- mysql造1000万数据_MySQL制作具有千万条测试数据的测试库
- 用Python对两个数据集中的图像进行水平拼接
- CPU被挖矿,Redis竟是内鬼,
- 架构师01-架构师的道
- java停车场管理系统,实现功能即可,要求图形界面。
热门文章
- 科技爱好者周刊(第 194 期):悲观者正确,乐观者成功
- 关于低版本IE浏览器无法显示图片的问题
- 滴滴章文嵩分享大数据在城市智慧交通领域探索实践
- osg显示CEGUI界面
- 介绍一种冷门但简单的双拼方案——紫光双拼
- 【自动驾驶】浅谈自动驾驶在业界的发展
- 中南大学计算机大一学什么时候发,中南大学计算机学院2019年夏令营
- WPS-Word:下一页分节符不能分页,插入下一页分节符时下一节没有在新的一页开始
- 【易通慧谷】供应链金融六大模式解析
- Nacos+openFeign 服务之间调用 出现错误:Load balancer does not contain an instance for the service 解决