高性能Server---Reactor模型

無處不在的C/S架構

在這個充斥著云的時代,我們使用的軟件可以說99%都是C/S架構的！

你發郵件用的Outlook,Foxmail等
你看視頻用的優酷，土豆等
你寫文檔用的Office365,googleDoc，Evernote等
你瀏覽網頁用的IE,Chrome等(B/S是特殊的C/S)
……

C/S架構的軟件帶來的一個明顯的好處就是：只要有網絡，你可以在任何地方干同一件事。

例如：你在家里使用Office365編寫了文檔。到了公司，只要打開編輯地址就可以看到在家里編寫的文檔，進行展示或者繼續編輯。甚至在手機上進行閱讀與編輯。不再需要U盤拷來拷去了。

C/S架構可以抽象為如下模型：

C就是Client(客戶端),上面的B是Browser(瀏覽器)
S就是Server(服務器)：服務器管理某種資源，并且通過操作這種資源來為它的客戶端提供某種服務

C/S架構之所以能夠流行的一個主要原因就是網速的提高以及費用的降低，特別是無線網絡速度的提高。試想在2G時代，大家最多就是看看文字網頁，小說什么的。看圖片，那簡直就是奢侈！更別說看視頻了！

網速的提高，使得越來越多的人使用網絡，例如：優酷，微信都是上億用戶量，更別說天貓雙11的瞬間訪問量了！這就對服務器有很高的要求！能夠快速處理海量的用戶請求！那服務器如何能快速的處理用戶的請求呢？

高性能服務器

高性能服務器至少要滿足如下幾個需求：

效率高：既然是高性能，那處理客戶端請求的效率當然要很高了
高可用：不能隨便就掛掉了
編程簡單：基于此服務器進行業務開發需要足夠簡單
可擴展：可方便的擴展功能
可伸縮：可簡單的通過部署的方式進行容量的伸縮，也就是服務需要無狀態

而滿足如上需求的一個基礎就是高性能的IO!

Socket

無論你是發郵件，瀏覽網頁，還是看視頻～實際底層都是使用的TCP/IP，而TCP/IP的編程抽象就是Socket!

我一直對Socket的中文翻譯很困惑，個人覺得是我所接觸的技術名詞翻譯里最莫名其妙的，沒有之一！

Socket中文翻譯為”套接字”！什么鬼？在很長的時間里我都無法將其和網絡編程關聯上！后來專門找了一些資料，最后在知乎上找到了一個還算滿意的答案(具體鏈接，請見文末的參考資料鏈接)！

Socket的原意是插口，想表達的意思是插口與插槽的關系！”send socket”插到”receive socket”里，建立了鏈接，然后就可以通信了！

套接字的翻譯，應該是參考了套接管(如下圖)！從這個層面上來看，是有那么點意思！

套接字這個翻譯已經是標準了，不糾結這個了！

我們看一下Socket之間建立鏈接及通信的過程！實際上就是對TCP/IP連接與通信過程的抽象:

服務端Socket會bind到指定的端口上，Listen客戶端的”插入”
客戶端Socket會Connect到服務端
當服務端Accept到客戶端連接后
就可以進行發送與接收消息了
通信完成后即可Close

對于IO來說，我們聽得比較多的是:

BIO:阻塞IO
NIO:非阻塞IO
同步IO
異步IO

以及其組合:

同步阻塞IO
同步非阻塞IO
異步阻塞IO
異步非阻塞IO

那么什么是阻塞IO、非阻塞IO、同步IO、異步IO呢？

一個IO操作其實分成了兩個步驟：發起IO請求和實際的IO操作
阻塞IO和非阻塞IO的區別在于第一步：發起IO請求是否會被阻塞，如果阻塞直到完成那么就是傳統的阻塞IO;如果不阻塞，那么就是非阻塞IO
同步IO和異步IO的區別就在于第二個步驟是否阻塞，如果實際的IO讀寫阻塞請求進程，那么就是同步IO，因此阻塞IO、非阻塞IO、IO復用、信號驅動IO都是同步IO;如果不阻塞，而是操作系統幫你做完IO操作再將結果返回給你，那么就是異步IO

舉個不太恰當的例子：比如你家網絡斷了，你打電話去中國電信報修！

你撥號—客戶端連接服務器
電話通了—連接建立
你說：“我家網斷了,幫我修下”—發送消息
說完你就在那里等，那么就是阻塞IO
如果正好你有事，你放下帶電話，然后處理其他事情了，過一會你來問下，修好了沒—那就是非阻塞IO
如果客服說：“馬上幫你處理，你稍等”—同步IO
如果客服說：“馬上幫你處理，好了通知你”，然后掛了電話—異步IO

本文只討論BIO和NIO,AIO使用度沒有前兩者普及，暫不討論！

下面從代碼層面看看BIO與NIO的流程!

BIO

客戶端代碼

//Bind,Connect
Socket client = new Socket("127.0.0.1",7777); 
//讀寫
PrintWriter pw = new PrintWriter(client.getOutputStream());
BufferedReader br=
new BufferedReader(new InputStreamReader(System.in)); 
pw.write(br.readLine());
//Close
pw.close();
br.close();

服務端代碼

Socket socket;
//Bind,Listen
ServerSocket ss = new ServerSocket(7777); 
while (true) { 
//Accept
socket = ss.accept();
//一般新建一個線程執行讀寫
BufferedReader br = new BufferedReader(
new InputStreamReader(socket .getInputStream())); 
System.out.println("you input is : " + br.readLine()); 
}

上面的代碼可以說是學習Java的Socket的入門級代碼了
代碼流程和前面的圖可以一一對上

模型圖如下所示：

BIO優缺點

優點
- 模型簡單
- 編碼簡單
缺點
- 性能瓶頸低

優缺點很明顯。這里主要說下缺點：主要瓶頸在線程上。每個連接都會建立一個線程。雖然線程消耗比進程小，但是一臺機器實際上能建立的有效線程有限，以Java來說，1.5以后，一個線程大致消耗1M內存！且隨著線程數量的增加，CPU切換線程上下文的消耗也隨之增加，在高過某個閥值后，繼續增加線程，性能不增反降！而同樣因為一個連接就新建一個線程，所以編碼模型很簡單！

就性能瓶頸這一點，就確定了BIO并不適合進行高性能服務器的開發！像Tomcat這樣的Web服務器，從7開始就從BIO改成了NIO，來提高服務器性能！

NIO

NIO客戶端代碼(連接)

//獲取socket通道
SocketChannel channel = SocketChannel.open();
channel.configureBlocking(false);
//獲得通道管理器
selector=Selector.open();
channel.connect(new InetSocketAddress(serverIp, port));
//為該通道注冊SelectionKey.OP_CONNECT事件
channel.register(selector, SelectionKey.OP_CONNECT);

NIO客戶端代碼(監聽)

while(true){
//選擇注冊過的io操作的事件(第一次為SelectionKey.OP_CONNECT)
selector.select();
while(SelectionKey key : selector.selectedKeys()){
if(key.isConnectable()){
SocketChannel channel=(SocketChannel)key.channel();
if(channel.isConnectionPending()){
channel.finishConnect();//如果正在連接，則完成連接
}
channel.register(selector, SelectionKey.OP_READ);
}else if(key.isReadable()){ //有可讀數據事件。
SocketChannel channel = (SocketChannel)key.channel();
ByteBuffer buffer = ByteBuffer.allocate(10);
channel.read(buffer);
byte[] data = buffer.array();
String message = new String(data);
System.out.println("recevie message from server:, size:"
+ buffer.position() + " msg: " + message);
}
}
}

NIO服務端代碼(連接)

//獲取一個ServerSocket通道
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port));
//獲取通道管理器
selector = Selector.open();
//將通道管理器與通道綁定，并為該通道注冊SelectionKey.OP_ACCEPT事件，
serverChannel.register(selector, SelectionKey.OP_ACCEPT);

NIO服務端代碼(監聽)

while(true){
//當有注冊的事件到達時，方法返回，否則阻塞。
selector.select();
for(SelectionKey key : selector.selectedKeys()){
if(key.isAcceptable()){
ServerSocketChannel server =
(ServerSocketChannel)key.channel();
SocketChannel channel = server.accept();
channel.write(ByteBuffer.wrap(
new String("send message to client").getBytes()));
//在與客戶端連接成功后，為客戶端通道注冊SelectionKey.OP_READ事件。
channel.register(selector, SelectionKey.OP_READ);
}else if(key.isReadable()){//有可讀數據事件
SocketChannel channel = (SocketChannel)key.channel();
ByteBuffer buffer = ByteBuffer.allocate(10);
int read = channel.read(buffer);
byte[] data = buffer.array();
String message = new String(data);
System.out.println("receive message from client, size:"
+ buffer.position() + " msg: " + message);
}
}
}

NIO模型示例如下：

Acceptor注冊Selector，監聽accept事件
當客戶端連接后，觸發accept事件
服務器構建對應的Channel，并在其上注冊Selector，監聽讀寫事件
當發生讀寫事件后，進行相應的讀寫處理

NIO優缺點

優點
- 性能瓶頸高
缺點
- 模型復雜
- 編碼復雜
- 需處理半包問題

NIO的優缺點和BIO就完全相反了!性能高，不用一個連接就建一個線程，可以一個線程處理所有的連接！相應的，編碼就復雜很多，從上面的代碼就可以明顯體會到了。還有一個問題，由于是非阻塞的，應用無法知道什么時候消息讀完了，就存在了半包問題！

半包問題

簡單看一下下面的圖就能理解半包問題了！

我們知道TCP/IP在發送消息的時候，可能會拆包(如上圖1)！這就導致接收端無法知道什么時候收到的數據是一個完整的數據。例如:發送端分別發送了ABC,DEF,GHI三條信息，發送時被拆成了AB,CDRFG,H,I這四個包進行發送，接受端如何將其進行還原呢？在BIO模型中，當讀不到數據后會阻塞，而NIO中不會!所以需要自行進行處理!例如，以換行符作為判斷依據，或者定長消息發生，或者自定義協議！

NIO雖然性能高，但是編碼復雜，且需要處理半包問題！為了方便的進行NIO開發，就有了Reactor模型!

Reactor模型

AWT Events

Reactor模型和AWT事件模型很像，就是將消息放到了一個隊列中，通過異步線程池對其進行消費！

Reactor中的組件

Reactor:Reactor是IO事件的派發者。
Acceptor:Acceptor接受client連接，建立對應client的Handler，并向Reactor注冊此Handler。
Handler:和一個client通訊的實體，按這樣的過程實現業務的處理。一般在基本的Handler基礎上還會有更進一步的層次劃分，用來抽象諸如decode，process和encoder這些過程。比如對Web Server而言，decode通常是HTTP請求的解析， process的過程會進一步涉及到Listener和Servlet的調用。業務邏輯的處理在Reactor模式里被分散的IO事件所打破，所以Handler需要有適當的機制在所需的信息還不全（讀到一半）的時候保存上下文，并在下一次IO事件到來的時候（另一半可讀了）能繼續中斷的處理。為了簡化設計，Handler通常被設計成狀態機，按GoF的state pattern來實現。

對應上面的NIO代碼來看:

Reactor：相當于有分發功能的Selector
Acceptor：NIO中建立連接的那個判斷分支
Handler：消息讀寫處理等操作類

Reactor從線程池和Reactor的選擇上可以細分為如下幾種：

Reactor單線程模型

這個模型和上面的NIO流程很類似，只是將消息相關處理獨立到了Handler中去了！

雖然上面說到NIO一個線程就可以支持所有的IO處理。但是瓶頸也是顯而易見的！我們看一個客戶端的情況，如果這個客戶端多次進行請求，如果在Handler中的處理速度較慢，那么后續的客戶端請求都會被積壓，導致響應變慢！所以引入了Reactor多線程模型!

Reactor多線程模型

Reactor多線程模型就是將Handler中的IO操作和非IO操作分開，操作IO的線程稱為IO線程，非IO操作的線程稱為工作線程!這樣的話，客戶端的請求會直接被丟到線程池中，客戶端發送請求就不會堵塞！

但是當用戶進一步增加的時候，Reactor會出現瓶頸！因為Reactor既要處理IO操作請求，又要響應連接請求！為了分擔Reactor的負擔，所以引入了主從Reactor模型!

主從Reactor模型

主Reactor用于響應連接請求，從Reactor用于處理IO操作請求！

Netty

Netty是一個高性能NIO框架，其是對Reactor模型的一個實現！

Netty客戶端代碼

EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true);
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeClientHandler());
}
});
ChannelFuture f = b.connect(host, port).sync();
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}

Netty Client Handler

public class TimeClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf m = (ByteBuf) msg;
try {
long currentTimeMillis =
(m.readUnsignedInt() - 2208988800L) * 1000L;
System.out.println(new Date(currentTimeMillis));
ctx.close();
} finally {
m.release();
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx,
Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}

Netty服務端代碼

EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128) 
.childOption(ChannelOption.SO_KEEPALIVE, true);
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(port).sync();
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}

Netty Server Handler

public class TimeServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(final ChannelHandlerContext ctx) {
final ByteBuf time = ctx.alloc().buffer(4);
time.writeInt((int)
(System.currentTimeMillis() / 1000L + 2208988800L));
final ChannelFuture f = ctx.writeAndFlush(time);
f.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
assert f == future;
ctx.close();
}
});
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx,
Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}

我們從Netty服務器代碼來看，與Reactor模型進行對應！

EventLoopGroup就相當于是Reactor，bossGroup對應主Reactor,workerGroup對應從Reactor
TimeServerHandler就是Handler
child開頭的方法配置的是客戶端channel，非child開頭的方法配置的是服務端channel

具體Netty內容，請訪問Netty官網！

Netty的問題

Netty開發中一個很明顯的問題就是回調，一是打破了線性編碼習慣，
二就是Callback Hell！

看下面這個例子：

1
2
3

a.doing1(); //1
a.doing2(); //2
a.doing3(); //3

1,2,3處代碼如果是同步的，那么將按順序執行！但是如果不是同步的呢？我還是希望2在1之后執行，3在2之后執行！怎么辦呢？想想AJAX!我們需要寫類似如下這樣的代碼！

a.doing1(new Callback(){
 public void callback(){
a.doing2(new Callback(){
 public void callback(){
a.doing3();
}
})
}
});

那有沒有辦法解決這個問題呢？其實不難，實現一個類似Future的功能！當Client獲取結果時，進行阻塞，當得到結果后再繼續往下走！實現方案，一個就是使用鎖了，還有一個就是使用RingBuffer。經測試，使用RingBuffer比使用鎖TPS有2000左右的提高！

參考資料

Socket為什么要翻譯成套接字？
Reactor論文
Doug Lea《Scalable IO in Java》
Netty源碼
剖析Disruptor:為什么會這么快？
剖析Disruptor:為什么會這么快？(中文)
Java SE1.6中的Synchronized
線程安全的無鎖RingBuffer實現
Java NIO類庫Selector機制解析（上）
Java NIO類庫Selector機制解析（下）

轉載于:https://www.cnblogs.com/he-px/p/7368501.html