Reactor模式
许多高性能的服务器软件离不开Reactor模式.像高性能缓存Redis,高性能web服务器Nginx,高性能的网络组件Netty,高性能的消息中间件Kafka,RocketMQ等.
那什么是Reactor模式呢?借用Doug Lea大师的话来说,就是:
Reactor模式由Reactor线程,Handles处理器两大角色组成,它们的职责分别是:
1.Reactor线程负责响应IO事件,并且将IO事件分发到Handles处理器
2.Handles线程:IO的读取,业务逻辑处理,写入
那为什么会产生Reactor模式呢?这就不得不说起OIO了.OIO又叫BIO,阻塞IO,像ServerScoket,Socket的read和write都会阻塞线程,直到完成IO操作. 系统的吞吐量特别低,每个IO操作都会阻塞其他的操作.为了解决这个问题,后面引入了每个连接一个线程.由每个线程处理一个连接的IO读,业务处理,IO写.这样每个连接在IO操作阻塞时不会影响其他的线程.
在系统的连接数少时没有问题,当连接数越来越多时,线程的数量就越来越多.对系统的资源消耗太高.
为了解决以上的问题,需要控制线程的数量.假设一个线程处理大量的连接,就可以控制系统资源的使用,同时提高系统的吞吐量.
单线程的Reactor模式
如果Reactor线程和Handles线程是同一个线程,就是最简单的Reactor模式了.
来看个例子,实现个Echo服务,简单返回客户端发送的数据.
服务端:
public interface Handler {
void handle() throws IOException;
}
public class AcceptorHandler implements Handler{
ServerSocketChannel serverSocketChannel;
Selector selector;
public AcceptorHandler(ServerSocketChannel serverSocketChannel,
Selector selector) {
this.serverSocketChannel = serverSocketChannel;
this.selector = selector;
}
@Override
public void handle() {
try {
SocketChannel channel = serverSocketChannel.accept();
channel.configureBlocking(false);
SelectionKey selectionKey = channel.register(selector, 0);
selectionKey.attach(new IOHandler(channel, selector,selectionKey));
selectionKey.interestOps(SelectionKey.OP_READ);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
public class IOHandler implements Handler {
Selector selector;
SocketChannel socketChannel;
SelectionKey selectionKey;
ByteBuffer byteBuffer = ByteBuffer.allocate(2048);
public IOHandler(SocketChannel socketChannel,Selector selector,
SelectionKey selectionKey) {
this.selector = selector;
this.socketChannel = socketChannel;
this.selectionKey = selectionKey;
}
@Override
public void handle() {
if (selectionKey.isReadable()) {
try {
while (socketChannel.read(byteBuffer) > 0) {
byteBuffer.flip();
System.out.println("读的内容是"+ new String(byteBuffer.array(),byteBuffer.position(),byteBuffer.limit()));
}
SelectionKey selectionKey1 = selectionKey.interestOps(SelectionKey.OP_WRITE);
selectionKey1.attach(this);
} catch (IOException e) {
try {
socketChannel.close();
} catch (IOException ex) {
}
}
}else if (selectionKey.isWritable()) {
try {
while (socketChannel.write(byteBuffer) > 0) {
}
byteBuffer.clear();
SelectionKey selectionKey1 = selectionKey.interestOps(SelectionKey.OP_READ);
selectionKey1.attach(this);
} catch (Exception e) {
try {
socketChannel.close();
} catch (IOException ex) {
}
}
}
}
}
public class EchoServer {
public static void main(String[] args) {
try {
startServer();
}catch (Exception e) {
e.printStackTrace();
}
}
public static void startServer() throws Exception {
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress("localhost",10700));
Selector selector = Selector.open();
SelectionKey selectionKey = serverSocketChannel.register(selector, 0);
selectionKey.attach(new AcceptorHandler(serverSocketChannel,selector));
selectionKey.interestOps(SelectionKey.OP_ACCEPT);
while (!Thread.interrupted()) {
int select = selector.select();
if (select > 0) {
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey selectionKey1 = iterator.next();
Handler handler = (Handler) selectionKey1.attachment();
handler.handle();
}
selectionKeys.clear();
}
}
serverSocketChannel.close();
selector.close();
}
}
客户端:
public class EchoClient {
private static final ByteBuffer byteBuffer = ByteBuffer.allocate(2048);
public static void main(String[] args) {
try {
startClient();
} catch (Exception e) {
e.printStackTrace();
}
}
public static void startClient() throws Exception {
SocketChannel socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
socketChannel.connect(new InetSocketAddress("localhost",10700));
while (!socketChannel.finishConnect()) {
}
System.out.println("成功与服务器建立连接");
Scanner scanner = new Scanner(System.in);
System.out.print("请输入一行:");
while (scanner.hasNext()) {
String s = scanner.nextLine();
byteBuffer.clear();
byteBuffer.put(s.getBytes(StandardCharsets.UTF_8));
byteBuffer.flip();
while(socketChannel.write(byteBuffer)>0) {
}
byteBuffer.clear();
while (socketChannel.read(byteBuffer) == 0) {
}
byteBuffer.flip();
System.out.println("从服务器接收数据:"+new String(byteBuffer.array(), byteBuffer.position(), byteBuffer.limit()));
System.out.print("请输入一行:");
}
socketChannel.close();
}
}
可以看到一个线程管理了很多连接,解决了每个连接一个线程的系统资源消耗的问题.可以看出,缺点就是进行IO操作时还会阻塞其他的线程.
多线程Reactor模式
在单线程的Reactor模式加上多线程来改进阻塞问题:
1.Handle加多线程,考虑使用线程池
2.Reactor加多线程,引入多个Selector
现在看下多线程版本的Reactor实现的Echo服务:
服务器
public class MultiAcceptorHandler implements Handler {
ServerSocketChannel serverSocketChannel;
Selector selector;
ExecutorService executorService;
public MultiAcceptorHandler(ServerSocketChannel serverSocketChannel,
Selector selector,ExecutorService executorService) {
this.serverSocketChannel = serverSocketChannel;
this.selector = selector;
this.executorService = executorService;
}
@Override
public void handle() {
try {
SocketChannel channel = serverSocketChannel.accept();
channel.configureBlocking(false);
new MultiIOHandler(channel, selector,executorService);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
public class MultiIOHandler implements Handler {
final Selector selector;
final SocketChannel socketChannel;
final SelectionKey selectionKey;
final ExecutorService executorService;
final ByteBuffer byteBuffer = ByteBuffer.allocate(2048);
int read = 0;
public MultiIOHandler(SocketChannel socketChannel, Selector selector,
ExecutorService executorService) {
this.selector = selector;
this.socketChannel = socketChannel;
this.executorService = executorService;
try {
this.selectionKey = socketChannel.register(selector, 0);
this.socketChannel.configureBlocking(false);
this.selectionKey.attach(this);
this.selectionKey.interestOps(SelectionKey.OP_READ);
selector.wakeup();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
@Override
public void handle() {
executorService.submit(this::syncRun);
}
public synchronized void syncRun() {
if (read==0) {
try {
while (socketChannel.read(byteBuffer) > 0) {
System.out.println("读的内容是"+ new String(byteBuffer.array(),0,byteBuffer.position()));
}
byteBuffer.flip();
System.out.println("读取"+byteBuffer.position()+","+byteBuffer.limit());
selectionKey.interestOps(SelectionKey.OP_WRITE);
read=1;
selector.wakeup();
} catch (IOException e) {
try {
socketChannel.close();
} catch (IOException ex) {
}
}
}else {
try {
while (socketChannel.write(byteBuffer) > 0) {
}
byteBuffer.clear();
selectionKey.interestOps(SelectionKey.OP_READ);
read=0;
selector.wakeup();
} catch (Exception e) {
try {
socketChannel.close();
} catch (IOException ex) {
}
}
}
}
}
public class SelectorThread implements Runnable {
Selector selector;
int index;
public SelectorThread(Selector selector,int index) {
this.selector = selector;
this.index = index;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
// System.out.println(index+"正在运行");
selector.select();
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey selectionKey = iterator.next();
Handler handler = (Handler) selectionKey.attachment();
if (handler != null) {
handler.handle();
}
}
selectionKeys.clear();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
public class MultiEchoServer {
public static void main(String[] args) {
try {
startServer();
}catch (Exception e) {
e.printStackTrace();
}
}
public static void startServer() throws Exception {
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress("localhost",10700));
Selector[] selectors = new Selector[2];
SelectorThread[] selectorThreads = new SelectorThread[2];
for (int i = 0; i < 2; i++) {
selectors[i] = Selector.open();
}
ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
SelectionKey selectionKey = serverSocketChannel.register(selectors[0], 0);
selectionKey.attach(new MultiAcceptorHandler(serverSocketChannel,selectors[1],executorService));
selectionKey.interestOps(SelectionKey.OP_ACCEPT);
for (int i = 0; i < 2; i++) {
selectorThreads[i] = new SelectorThread(selectors[i],i);
new Thread(selectorThreads[i]).start();
}
// Thread closeThread = new Thread(() -> {
// try {
// executorService.awaitTermination(5, TimeUnit.SECONDS);
//
// executorService.shutdown();
// serverSocketChannel.close();
//
// for (int i= 0; i < 2; i++) {
// selectors[i].close();
// }
// } catch (Exception e) {
// e.printStackTrace();
// }
// });
//
// Runtime.getRuntime().addShutdownHook(closeThread);
}
}
Selector[] selectors = new Selector[2]使用了两个Selector,第一个用于接收客户端的连接,注册的Handle是MultiAcceptorHandler;接收连接后将处理IO注册到第二个Selector, 第二个用于处理IO读取和写入,注册的Handle是MultiIOHandler.每个连接的IO处理也是用线程池处理的.
注意:在MultiIOHandler不能用selectionKey.isReadable()来判断是否是可读,增加了read标志来判断.
优缺点
优点
1.响应快,虽然Reactor线程是同步的,但是不会被IO操作所阻塞
2.编程简单
3.可扩展,可以加多线程充分利用CPU资源
缺点
1.有一定复杂性
2.依赖操作系统支持
3.同一个Handle中出现长时间读写,会造成Reactor线程的其他通道的IO处理