1、Reactor 与 Procator
- Reactor 模式反应器,同时接收多个服务器请求，并且依次同步的处理它们事件驱动程序
- Procator 模式，异步接收和同时处理多个服务请求的事件驱动的程序
- Netty整体框架是Reactor模式是一个完整体现
2、 Doug Lea 大神
- Doug Lea 是Java并发主导者大神(1995的论文)，可以说没有他的思想就没有java的NIO并发编程
为什么Netty使用两个NioEventLoopGroup（重点面试可能问题）
- 一个是mainReactor（bossGroup/parentGroup）用于参数接收请求,不做任何事情处理
- 一个是subReactor（workGroup/childrenGroup）用于参数对要实现对象处理
- ServerBootstrap类group方法
/**
*
设置EventLoopGroup的父（acceptor）和子（client）。 这些EventLoopGroup用于处理ServerChannel和Channel的所有事件和IO。
*/
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
super.group(parentGroup);
if (childGroup == null) {
throw new NullPointerException("childGroup");
}
if (this.childGroup != null) {
throw new IllegalStateException("childGroup set already");
}
this.childGroup = childGroup;
return this;
}
3、网络结构
- 读请求 => 解码请求 => 进程服务 => 编码 => 发送响应
4、传统服务设计
-
图解关系
- 1、弊端： 多线程的模型设计，线程多，CPU开销大
- 2、建议改良:
- 优雅降级,可伸缩，将不同的任务分解最小的任务执行，事件驱动设计改良。在改良过程中也有它的缺点,
- 分发较慢（需要手动绑定事件）跟踪状态，如：AWT Event Queue 图形事件队列
- 加大带宽资源，调节资源服务质量
5、传统Reactor设计（单线程）
图解关系
Reactor本身就是一个线程，Netty的EventLoopGroup的对象，第一个EventLoopGroup不做任何事情，交给第二个EventLoopGroup进行处理
采用了java.nio的支持，在NiO的有Channel，Buffer，Selectors（SelectionKeys）三个重要概念、请看
Channels(通道)
- 连接文件，Sockets，支持非阻塞读等操作
Buffers(缓存区)
- 类似于数组的对象，可以通过管道Channel直接读或写操作
Selectors(选择器)
- 告诉Channel通道有IO事件
-
SelectionKeys(事件选择处理)
- 保持IO事件状态(SelectionKey.OP_ACCEPT连接事件，OP_READ读事件等)和绑定
6、Worker Thread Pools 工作线程池
图解关系
-
Worker Thread Pools 工作线程池
- 足够的处理来超过开销
- 更难以与IO重叠处理，最好什么时候可以先将所有输入读入缓冲区
- 使用线程池，因此可以协调和控制，通常需要比客户端少得多的线程
Coordinating Tasks 协调任务
- 切换：每个任务都启用，触发或调用下一个任务，通常最快但可能很脆弱
- 回调：每个处理程序调度程序，设置状态，附件等，GoF Mediator模式的变体
- 队列：例如，跨阶段传递缓冲区
- 期待: 当每个任务产生结果时,协调分层在加入或等待/通知之上
Multiple Reactor Threads 复合Reactor线程
-
使用Reactor池
- 1、用于匹配CPU和IO速率
- 2、静态或动态结构:每个都有自己的Selector，Thread，dispatch循环,主要接收器发布分配给其他Reactor
7、 Multiple Reactors 多个Reactor线程池（非常重要）
- 图解关系
java.nio features 的使用（重点）
- 每个Reactor都有自己多个的Selectors选择器，将不同的处理程序绑定到不同的IO事件，需要关注同步更要协调
- File transfer文件传输 : 文件在网络的传输拷贝过程
- Memory-mapped files 内存文件映射： 通过缓冲区访问文件
- Direct buffers 直接缓冲：有时可以实现零拷贝传输，但是有设置和最终化开销，最适合具有长期连接的应用程序
基于连接的扩展
- 1、客户端的连接
- 2、客户端发送一系列消息/请求
- 3、客户端断开连接
- 4、例如：数据库和事务监视器，多人游戏，聊天等
可以扩展基本的网络服务模式
- 1、处理更多相对长连接的客户端
- 2、跟踪客户端和会话状态（包括丢弃）
- 3、跨多个主机分发服务
API 场景使用
- Buffer
- ByteBuffer (CharBuffer, LongBuffer, etc not shown.)
- Channel
- SelectableChannel
- SocketChannel
- ServerSocketChannel
- FileChannel
- Selector
- SelectionKey
- 代码展示
Buffer
abstract class Buffer {
int capacity();
int position();
Buffer position(int newPosition);
int limit();
Buffer limit(int newLimit);
Buffer mark();
Buffer reset();
Buffer clear();
Buffer flip();
Buffer rewind();
int remaining();
boolean hasRemaining();
boolean isReadOnly();
}
ByteBuffer
abstract class ByteBuffer extends Buffer {
static ByteBuffer allocateDirect(int capacity);
static ByteBuffer allocate(int capacity);
static ByteBuffer wrap(byte[] src, int offset, int len);
static ByteBuffer wrap(byte[] src);
boolean isDirect();
ByteOrder order();
ByteBuffer order(ByteOrder bo);
ByteBuffer slice();
ByteBuffer duplicate();
ByteBuffer compact();
ByteBuffer asReadOnlyBuffer();
byte get();
byte get(int index);
ByteBuffer get(byte[] dst, int offset, int length);
ByteBuffer get(byte[] dst);
ByteBuffer put(byte b);
ByteBuffer put(int index, byte b);
ByteBuffer put(byte[] src, int offset, int length);
ByteBuffer put(ByteBuffer src);
ByteBuffer put(byte[] src);
char getChar();
char getChar(int index);
ByteBuffer putChar(char value);
ByteBuffer putChar(int index, char value);
CharBuffer asCharBuffer();
short getShort();
short getShort(int index);
ByteBuffer putShort(short value);
ByteBuffer putShort(int index, short value);
ShortBuffer asShortBuffer();
int getInt();
int getInt(int index);
ByteBuffer putInt(int value);
ByteBuffer putInt(int index, int value);
IntBuffer asIntBuffer();
long getLong();
long getLong(int index);
ByteBuffer putLong(long value);
ByteBuffer putLong(int index, long value);
LongBuffer asLongBuffer();
float getFloat();
float getFloat(int index);
ByteBuffer putFloat(float value);
ByteBuffer putFloat(int index, float value);
FloatBuffer asFloatBuffer();
double getDouble();
double getDouble(int index);
ByteBuffer putDouble(double value);
ByteBuffer putDouble(int index, double value);
DoubleBuffer asDoubleBuffer();
}
Channel
interface Channel {
boolean isOpen();
void close() throws IOException;
}
interface ReadableByteChannel extends Channel {
int read(ByteBuffer dst) throws IOException;
}
interface WritableByteChannel extends Channel {
int write(ByteBuffer src) throws IOException;
}
interface ScatteringByteChannel extends ReadableByteChannel {
int read(ByteBuffer[] dsts, int offset, int length)throws IOException;
int read(ByteBuffer[] dsts) throws IOException;
}
interface GatheringByteChannel extends WritableByteChannel {
int write(ByteBuffer[] srcs, int offset, int length)throws IOException;
int write(ByteBuffer[] srcs) throws IOException;
}
SelectableChannel
abstract class SelectableChannel implements Channel {
int validOps();
boolean isRegistered();
SelectionKey keyFor(Selector sel);
SelectionKey register(Selector sel, int ops) throws ClosedChannelException;
void configureBlocking(boolean block) throws IOException;
boolean isBlocking();
Object blockingLock();
}
8、Reactor Pattern设计模型
- 根据IO事件进行分发与响应
- Handlers非阻塞行为
- 通过Handlers绑定到事件上
事件驱动模型
- 选择更优的替代物，减少资源
- 通常不需要每个客户端一个线程，减少开销
- 减少上下文切换，通常减少锁定，但调度可能会更慢
- 必须手动将操作绑定到事件
- 通常难以编程,必须分解为简单的非阻塞操作与GUI事件驱动的操作类似
- 无法消除所有阻塞：GC，页面错误等 必须跟踪逻辑服务状态
状态Handler与多复用线程模式
- 处理器会减慢Reactor方法的速度，将非IO的线程交给其他线程处理
Multithreaded Designs(多线程设计)
- 策略性地添加线程：实现可伸缩性 ,主要适用于多处理器
- 线程工作：Reactor应该快速触发handlers处理程序，但处理器会减慢Reactor方法的速度
- Reactor线程可以使IO完全饱和将负载分配给其他反应器负载平衡以匹配CPU和IO速率
9、Reactor模型的角色构成（Reactor模型一共有5种角色构成、非常重点）
-
图解关系
*
