Reactor网络编程模型解析
Posted 踩踩踩从踩
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Reactor网络编程模型解析相关的知识,希望对你有一定的参考价值。
前言
Reactor设计模式是一种处理并发I/O比较常见的一种模式,将客户端请求分别发送到不同处理器上,来提升事件处理的效率。最常见的应用场景java NIO当中用户处理网络请求,使用的是异步非阻塞IO。包括netty框架等都是使用该模型。至于什么异步 和 非阻塞的定义这个可以看看我下面的文章
网络编程基础之七层协议及TCP、UDP、Http、Nio解析
Reactor模型概述
为什么要使用Reactor模型,主要来自于只使用NIO,只能解决多连接问题,非阻塞。但是建立连接过后每个连接的处理也只有单线程,同步的处理;而怎么达到异步处理,这里引入Reactor模型使用NIO与线程相结合去快速处理任务。
在 doug lea 写的Scalable IO in Java中,这本书分析与构建可伸缩的高性能IO服务,介绍事件驱动模型,并阐述nio如何与线程相结合提升程序的处理能力。
下面截取的一部分图形并分析Reactor模型
包括多个客户端端发送给服务端,并使用Reactor 中进行分配 Dispatch 给处理
单线程模型下的Reactor
在单线程模型中:
reactor-thread客户端进来将客户端请求发送给acceptor处理,Acceptor将新进来的客户端连接分配到handler;每个handler进行处理。
这个就是在jdk中的selector 选择器达到的一个设计模式通过每个channel通道去处理业务逻辑;这里的逻辑实现
public class SelectorDemo {
public static void main(String[] args) throws IOException {
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
// 创建selector
Selector selector = Selector.open();
// 注册selector
ssc.register(selector, SelectionKey.OP_ACCEPT);// 这里需要设置默认的事件
ssc.socket().bind(new InetSocketAddress(8080));// 绑定端口
while (true) {
int readyChannels = selector.select();// 会阻塞,直到有事件触发
if (readyChannels == 0)
continue;
Set<SelectionKey> selectedKeys = selector.selectedKeys();// 获取被触发的事件集合
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while (keyIterator.hasNext()) {
SelectionKey key = keyIterator.next();
if (key.isAcceptable()) {
SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
// serverSocketChannel 收到一个新连接,只能作用于ServerSocketChannel
} else if (key.isConnectable()) {
// 连接到远程服务器,只在客户端异步连接时生效
} else if (key.isReadable()) {
// SocketChannel 中有数据可以读
ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
SocketChannel socketChannel = (SocketChannel) key.channel();
while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1) {
// 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
if (requestBuffer.position() > 0)
break;
}
requestBuffer.flip();
byte[] content = new byte[requestBuffer.limit()];
requestBuffer.get(content);
System.out.println(new String(content));
socketChannel.register(selector, SelectionKey.OP_WRITE);
} else if (key.isWritable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
// SocketChannel 可以开始写入数据
ByteBuffer buffer = ByteBuffer.wrap("sucess".getBytes());
socketChannel.write(buffer);
socketChannel.close();
}
// 将已处理的事件移除
keyIterator.remove();
}
}
}
}
这里会出现一个问题,也就是read 和wite状态有可能比较慢,因此多个客户端连接进来,也就是服务端处理逻辑太慢。解决办法在Scalable IO in Java 中提出的,使用工作线程池处理逻辑。
业务工作线程池模式的Reactor
这里新增了一个thread pool进行处理
此模式是在单Reactor的基础上,将数据处理交给线程出处理,业务操作慢导致堵车的。但还是会有问题
对于海量连接 所有io都只能在一个线程处理。读数据和写数据海量操作时但线程是处理不过来。
IO线程池和业务工作线程池模式的Reactor
这里将IO线程进行拆分开 分配多个连接的IO。也是为了提升性能。
实现一个NIOReactor多线程模型
- 建立NIOReactorDemo数据 创建 MainReactor 主线程 和 SubReactor Acceptor 对象
- MainReactor 中持有Selector 对象,运行 阻塞等待事件的驱动。 初始化主线程,这里直接线程去启动,而不采用线程池去管理,因为我觉得主线程Reactor单个线程够了,也不用多线程,不涉及数据读取和写入,只是做一个透传。
public class NIOReactorDemo {
private ServerSocketChannel serverSocketChannel;
private static ExecutorService threadPool = Executors.newFixedThreadPool(10);
private MainReactor mainReactor = new MainReactor();// 负责接收客户端的连接
public static void main(String[] args) throws Exception {
NIOReactorDemo demo = new NIOReactorDemo();
// 初始化
demo.init();
}
/**
* 初始化服务端channel并注册到mainReactor中,并且启动mainReactor
*
* @throws IOException
*/
public void init() throws IOException {
// 主线程初始化数据并启动
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
mainReactor.registor(serverSocketChannel);
mainReactor.start();
}
/**
* MainReactor 负责接收客户端连接 注册
*/
class MainReactor extends Thread {
private Selector selector;
@Override
public void run() {
while (true) {
try {
// 5. 启动selector(管家)
selector.select();// 阻塞,直到事件通知才会返回
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isAcceptable()) {
SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();
// 将客户端连接给到acceptor
new Acceptor(socketChannel);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
public void registor(ServerSocketChannel serverSocketChannel) throws IOException {
selector = Selector.open();
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
}
/**
* 将客户端的连接分配到一个subReactor线程中,并启动subReactor线程
*
* @author
*
*/
class Acceptor {
public Acceptor(SocketChannel socketChannel) {
}
}
}
/**
* 负责从客户端读数据后交给工作线程去处理,和给客户端写数据
*/
class SubReactor implements Runnable {
@Override
public void run() {
}
}
}
- 接下来继续写 Acceptor 类做一个数据的分配
/**
* 将客户端的连接分配到一个subReactor线程中,并启动subReactor线程
*
* @author
*
*/
class Acceptor {
public Acceptor(SocketChannel socketChannel) throws IOException {
socketChannel.configureBlocking(false);
SubReactor subReactor = new SubReactor();
subReactor.register(socketChannel);
subThreadPool.execute(subReactor);
}
}
- 并在次io次任务中创建一个 select并进行注册。
/**
* 负责从客户端读数据后交给工作线程去处理,和给客户端写数据
*/
class SubReactor implements Runnable {
private Selector selector;
public SubReactor() throws IOException {
selector = Selector.open();
}
@Override
public void run() {
}
/**
* 将客户端channel注册到selector中,注册OP_READ事件
*
* @param socketChannel
* @throws ClosedChannelException
*/
public void register(SocketChannel socketChannel) throws ClosedChannelException {
socketChannel.register(selector, SelectionKey.OP_READ);
}
}
- 在 SubReactor 读取数据并交给handler,事件进行传递并处理。
/**
* 负责从客户端读数据后交给工作线程去处理,和给客户端写数据
*/
class SubReactor implements Runnable {
private Selector selector;
public SubReactor() throws IOException {
selector = Selector.open();
}
@Override
public void run() {
try {
// 5. 启动selector(管家)
selector.select();// 阻塞,直到事件通知才会返回
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isReadable()) {// 客户端连接有数据可以读时触发
try {
SocketChannel socketChannel = (SocketChannel) key.channel();
new Handler(socketChannel);
} catch (Exception e) {
e.printStackTrace();
key.cancel();
}
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
- 最后来个handler处理并交给业务处理线程池 这里把申请的bytebuffer的大小是自己定的, 并没有做到那么完整,只是可以当作例子使用。如果需要优化,具体优化把
class Handler {
public Handler(SocketChannel socketChannel) throws IOException {
ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1) {
// 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
if (requestBuffer.position() > 0)
break;
}
if (requestBuffer.position() == 0)
return; // 如果没数据了, 则不继续后面的处理
requestBuffer.flip();
byte[] content = new byte[requestBuffer.remaining()];
requestBuffer.get(content);
System.out.println(new String(content));
System.out.println("收到数据,来自:" + socketChannel.getRemoteAddress());
// TODO 业务操作 数据库 接口调用等等
workPool.execute(new Runnable() {
@Override
public void run() {
// 处理业务
}
});
// 响应结果 200
String response = "HTTP/1.1 200 OK\\r\\n" + "Content-Length: 11\\r\\n\\r\\n" + "Hello World";
ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
while (buffer.hasRemaining()) {
socketChannel.write(buffer);
}
}
}
- 最后在mian方法中做个绑定和启动服务器
public static void main(String[] args) throws Exception {
NIOReactorDemo demo = new NIOReactorDemo();
// 初始化
demo.init();
demo.bind();
}
/**
* 绑定端口,启动服务
*
* @throws IOException
*/
private void bind() throws IOException {
serverSocketChannel.bind(new InetSocketAddress(8080));
}
采用的Reactor模型采用事件驱动,将IO数据流进行拆分开 分流处理,解决批量连接和业务处理不阻塞问题
以上是关于Reactor网络编程模型解析的主要内容,如果未能解决你的问题,请参考以下文章