也談Reactor模式
何謂Reactor模式？它是實(shí)現(xiàn)高性能IO的一種設(shè)計(jì)模式。網(wǎng)上資料有很多，有些寫(xiě)的也很好，但大多不知其所以然。這里博主按自己的思路簡(jiǎn)單介紹下，有不對(duì)的地方敬請(qǐng)指正。
BIO
Java1.4（2002年）以前，IO都是Blocking的，也就是常說(shuō)的BIO，它在等待請(qǐng)求、讀、寫(xiě)（返回）三個(gè)環(huán)節(jié)都是阻塞的。在等待請(qǐng)求階段，系統(tǒng)無(wú)法知道請(qǐng)求何時(shí)到達(dá)，因此需要一個(gè)主線(xiàn)程一直守著，當(dāng)有請(qǐng)求進(jìn)來(lái)時(shí)，將請(qǐng)求分發(fā)給讀寫(xiě)線(xiàn)程。如圖：
代碼如下：
ExecutorService executor = Excutors.newFixedThreadPollExecutor(100);//線(xiàn)程池 ServerSocket serverSocket = new ServerSocket(); serverSocket.bind(8088); while(!Thread.currentThread.isInturrupted()){//主線(xiàn)程死循環(huán)等待新連接到來(lái) Socket socket = serverSocket.accept(); executor.submit(new ConnectIOnHandler(socket));//為新的連接創(chuàng)建新的線(xiàn)程 }
class ConnectIOnHandler extends Thread{ private Socket socket; public ConnectIOnHandler(Socket socket){ this.socket = socket; } public void run(){ while(!Thread.currentThread.isInturrupted()&&!socket.isClosed()){//死循環(huán)處理讀寫(xiě)事件 String someThing = socket.read()....//讀取數(shù)據(jù) if(someThing!=null){
......//處理數(shù)據(jù)
socket.write()....//寫(xiě)數(shù)據(jù) } } }
需知，請(qǐng)求進(jìn)來(lái)（accept），并不表示數(shù)據(jù)馬上達(dá)到了，可能隔一段時(shí)間才會(huì)傳進(jìn)來(lái)，這個(gè)時(shí)候socket.read()也是一直阻塞的狀態(tài)。socket.write()也同理，當(dāng)向磁盤(pán)或其它socket寫(xiě)數(shù)據(jù)時(shí)，也要等對(duì)方準(zhǔn)備好才能寫(xiě)入，在對(duì)方準(zhǔn)備階段，socket.write()也是阻塞的。這兩個(gè)環(huán)節(jié)可能的無(wú)效阻塞導(dǎo)致讀寫(xiě)線(xiàn)程的低效。
NIO
Java1.4開(kāi)始，引入了NIO。NIO有三個(gè)概念：Selector、Buffer、Channel。與BIO的區(qū)別是，請(qǐng)求進(jìn)來(lái)后，并不會(huì)馬上分派IO線(xiàn)程，而是依靠操作系統(tǒng)底層的多路復(fù)用機(jī)制（select/poll/epoll等），在監(jiān)聽(tīng)到socket讀寫(xiě)就緒之后，再分配IO線(xiàn)程（實(shí)際可由當(dāng)前線(xiàn)程[使用Buffer和Channel]直接讀寫(xiě)，因?yàn)樽x寫(xiě)本身的效率很高），這就避免了線(xiàn)程等待。且與BIO多線(xiàn)程方式相比，使用I/O多路復(fù)用技術(shù)，系統(tǒng)不必創(chuàng)建和維護(hù)龐大的線(xiàn)程池，從而大大減小了開(kāi)銷(xiāo)。這部分工作是NIO的核心，由Selector負(fù)責(zé)，本質(zhì)上是多路復(fù)用的Java封裝。而B(niǎo)uffer和Channel又封裝了一層socket的讀寫(xiě)，應(yīng)該為的是將IO與業(yè)務(wù)代碼徹底分離。以下圖示為本人理解：
如圖示，與BIO中監(jiān)聽(tīng)線(xiàn)程職責(zé)不同，Selector監(jiān)聽(tīng)的不只是連接請(qǐng)求，還有讀寫(xiě)就緒事件，當(dāng)某個(gè)事件發(fā)生時(shí)，即通知注冊(cè)了該事件的Channel，由Channel操作socket讀寫(xiě)B(tài)uffer。虛線(xiàn)表示需要具體的NIO框架或業(yè)務(wù)代碼自己處理，比如Channel如何注冊(cè)以及注冊(cè)何種事件，Channel處理IO的方式（如在當(dāng)前線(xiàn)程處理還是新開(kāi)線(xiàn)程，若新開(kāi)線(xiàn)程，則可看作是AIO模式）等。NIO只是提供了一套機(jī)制，具體使用還是需要編程實(shí)現(xiàn)（Reactor模式就是OO的一種實(shí)現(xiàn)）。
示例代碼（摘自Java NIO詳解）
服務(wù)端：
1 package cn.blog.test.NioTest; 2 3 4 import java.io.IOException; 5 import java.net.InetSocketAddress; 6 import java.nio.ByteBuffer; 7 import java.nio.channels.*; 8 import java.nio.charset.Charset; 9 import java.util.Iterator; 10 import java.util.Set; 11 12 13 public class MyNioServer { 14 private Selector selector; //創(chuàng)建一個(gè)選擇器 15 private final static int port = 8686; 16 private final static int BUF_SIZE = 10240; 17 18 private void initServer() throws IOException { 19 //創(chuàng)建通道管理器對(duì)象selector 20 this.selector=Selector.open(); 21 22 //創(chuàng)建一個(gè)通道對(duì)象channel 23 ServerSocketChannel channel = ServerSocketChannel.open(); 24 channel.configureBlocking(false); //將通道設(shè)置為非阻塞 25 channel.socket().bind(new InetSocketAddress(port)); //將通道綁定在8686端口 26 27 //將上述的通道管理器和通道綁定，并為該通道注冊(cè)O(shè)P_ACCEPT事件 28 //注冊(cè)事件后，當(dāng)該事件到達(dá)時(shí)，selector.select()會(huì)返回（一個(gè)key），如果該事件沒(méi)到達(dá)selector.select()會(huì)一直阻塞 29 SelectionKey selectionKey = channel.register(selector,SelectionKey.OP_ACCEPT); 30 31 while (true){ //輪詢(xún) 32 selector.select(); //這是一個(gè)阻塞方法，一直等待直到有數(shù)據(jù)可讀，返回值是key的數(shù)量（可以有多個(gè)） 33 Set keys = selector.selectedKeys(); //如果channel有數(shù)據(jù)了，將生成的key訪(fǎng)入keys集合中 34 Iterator iterator = keys.iterator(); //得到這個(gè)keys集合的迭代器 35 while (iterator.hasNext()){ //使用迭代器遍歷集合 36 SelectionKey key = (SelectionKey) iterator.next(); //得到集合中的一個(gè)key實(shí)例 37 iterator.remove(); //拿到當(dāng)前key實(shí)例之后記得在迭代器中將這個(gè)元素刪除，非常重要，否則會(huì)出錯(cuò) 38 if (key.isAcceptable()){ //判斷當(dāng)前key所代表的channel是否在Acceptable狀態(tài)，如果是就進(jìn)行接收 39 doAccept(key); 40 }else if (key.isReadable()){ 41 doRead(key); 42 }else if (key.isWritable() && key.isValid()){ 43 doWrite(key); 44 }else if (key.isConnectable()){ 45 System.out.println("連接成功！"); 46 } 47 } 48 } 49 } 50 51 public void doAccept(SelectionKey key) throws IOException { 52 ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); 53 System.out.println("ServerSocketChannel正在循環(huán)監(jiān)聽(tīng)"); 54 SocketChannel clientChannel = serverChannel.accept(); 55 clientChannel.configureBlocking(false); 56 clientChannel.register(key.selector(),SelectionKey.OP_READ); 57 } 58 59 public void doRead(SelectionKey key) throws IOException { 60 SocketChannel clientChannel = (SocketChannel) key.channel(); 61 ByteBuffer byteBuffer = ByteBuffer.allocate(BUF_SIZE); 62 long bytesRead = clientChannel.read(byteBuffer); 63 while (bytesRead>0){ 64 byteBuffer.flip(); 65 byte[] data = byteBuffer.array(); 66 String info = new String(data).trim(); 67 System.out.println("從客戶(hù)端發(fā)送過(guò)來(lái)的消息是："+info); 68 byteBuffer.clear(); 69 bytesRead = clientChannel.read(byteBuffer); 70 } 71 if (bytesRead==-1){ 72 clientChannel.close(); 73 } 74 } 75 76 public void doWrite(SelectionKey key) throws IOException { 77 ByteBuffer byteBuffer = ByteBuffer.allocate(BUF_SIZE); 78 byteBuffer.flip(); 79 SocketChannel clientChannel = (SocketChannel) key.channel(); 80 while (byteBuffer.hasRemaining()){ 81 clientChannel.write(byteBuffer); 82 } 83 byteBuffer.compact(); 84 } 85 86 public static void main(String[] args) throws IOException { 87 MyNioServer myNioServer = new MyNioServer(); 88 myNioServer.initServer(); 89 } 90 }
客戶(hù)端：
1 package cn.blog.test.NioTest; 2 3 4 import java.io.IOException; 5 import java.net.InetSocketAddress; 6 import java.nio.ByteBuffer; 7 import java.nio.channels.SelectionKey; 8 import java.nio.channels.Selector; 9 import java.nio.channels.SocketChannel; 10 import java.util.Iterator; 11 12 public class MyNioClient { 13 private Selector selector; //創(chuàng)建一個(gè)選擇器 14 private final static int port = 8686; 15 private final static int BUF_SIZE = 10240; 16 private static ByteBuffer byteBuffer = ByteBuffer.allocate(BUF_SIZE); 17 18 private void initClient() throws IOException { 19 this.selector = Selector.open(); 20 SocketChannel clientChannel = SocketChannel.open(); 21 clientChannel.configureBlocking(false); 22 clientChannel.connect(new InetSocketAddress(port)); 23 clientChannel.register(selector, SelectionKey.OP_CONNECT); 24 while (true){ 25 selector.select(); 26 Iterator<SelectionKey> iterator = selector.selectedKeys().iterator(); 27 while (iterator.hasNext()){ 28 SelectionKey key = iterator.next(); 29 iterator.remove(); 30 if (key.isConnectable()){ 31 doConnect(key); 32 }else if (key.isReadable()){ 33 doRead(key); 34 } 35 } 36 } 37 } 38 39 public void doConnect(SelectionKey key) throws IOException { 40 SocketChannel clientChannel = (SocketChannel) key.channel(); 41 if (clientChannel.isConnectionPending()){ 42 clientChannel.finishConnect(); 43 } 44 clientChannel.configureBlocking(false); 45 String info = "服務(wù)端你好!!"; 46 byteBuffer.clear(); 47 byteBuffer.put(info.getBytes("UTF-8")); 48 byteBuffer.flip(); 49 clientChannel.write(byteBuffer); 50 //clientChannel.register(key.selector(),SelectionKey.OP_READ); 51 clientChannel.close(); 52 } 53 54 public void doRead(SelectionKey key) throws IOException { 55 SocketChannel clientChannel = (SocketChannel) key.channel(); 56 clientChannel.read(byteBuffer); 57 byte[] data = byteBuffer.array(); 58 String msg = new String(data).trim(); 59 System.out.println("服務(wù)端發(fā)送消息："+msg); 60 clientChannel.close(); 61 key.selector().close(); 62 } 63 64 public static void main(String[] args) throws IOException { 65 MyNioClient myNioClient = new MyNioClient(); 66 myNioClient.initClient(); 67 } 68 }
在早期的JDK1.4和1.5 update10版本之前，Selector基于select/poll模型實(shí)現(xiàn)，是基于IO復(fù)用技術(shù)的非阻塞IO，不是異步IO。在JDK1.5 update10和linux core2.6以上版本，sun優(yōu)化了Selctor的實(shí)現(xiàn)，底層使用epoll替換了select/poll。另?yè)?jù)說(shuō)Buffer指向的并非堆內(nèi)內(nèi)存，NIO使用 Native 函數(shù)庫(kù)直接分配堆外內(nèi)存，然后通過(guò)一個(gè)存儲(chǔ)在 Java 堆的 DirectByteBuffer 對(duì)象作為這塊內(nèi)存的引用進(jìn)行操作，避免了在 Java 堆和 Native 堆中來(lái)回復(fù)制數(shù)據(jù)。
NIO的實(shí)現(xiàn)解析可參看：深入淺出NIO Socket實(shí)現(xiàn)機(jī)制
Reactor模式
NIO為實(shí)現(xiàn)Reactor模式提供了基礎(chǔ)，上面的NIO圖示其實(shí)就是Reactor模式的雛形，只是Reactor以O(shè)O的方式抽象出了幾個(gè)概念，使得職責(zé)劃分更加明確。
- Reactor：Reactor是IO事件的派發(fā)者，對(duì)應(yīng)NIO的Selector；
- Acceptor：Acceptor接受client連接，建立對(duì)應(yīng)client的Handler，并向Reactor注冊(cè)此Handler，對(duì)應(yīng)NIO中注冊(cè)Channel和事件觸發(fā)時(shí)的判斷分支（上述NIO服務(wù)端示例代碼的38-46行）；
- Handler：IO處理類(lèi)，對(duì)應(yīng)NIO中Channel[使用socket]操作Buffer的過(guò)程。
基于上述三個(gè)角色畫(huà)出Reactor模式圖如下：
如此，Reactor模式便非常清晰地展現(xiàn)在我們眼前。那么業(yè)務(wù)線(xiàn)程如何與Reactor交互呢？由前文所知，數(shù)據(jù)存取于Buffer，具體操作由Handler負(fù)責(zé)。socket.read()將數(shù)據(jù)讀入Buffer，需要一種機(jī)制將Buffer引用推送給業(yè)務(wù)線(xiàn)程；同樣，業(yè)務(wù)線(xiàn)程返回的數(shù)據(jù)需要寫(xiě)入Buffer，按Reactor模式，寫(xiě)入后還需要注冊(cè)write事件，socket可寫(xiě)后write()。如果直接調(diào)用的話(huà)，至少Handler和業(yè)務(wù)代碼會(huì)耦合在一起，常見(jiàn)的解耦方式是定義接口，或使用消息中間件。
其它
話(huà)說(shuō)回來(lái)，由于相對(duì)短暫的歷史以及相對(duì)封閉的環(huán)境，.Net社區(qū)缺少很多概念的演化、探究和討論，這也導(dǎo)致了.Neter們這些概念的缺失。雖然從語(yǔ)言層面上來(lái)說(shuō)，C#和Java大同小異，前者甚至一定程度的有語(yǔ)法上的便利，然而只有認(rèn)識(shí)到了其背后的思想和模式，才能真正用好這門(mén)語(yǔ)言，這就是.Neter需要了解Java及其歷史的原因，畢竟.Net一開(kāi)始就是參照著Java來(lái)的。
比如.Net里的堆棧概念，就算一些經(jīng)典書(shū)籍都沒(méi)有非常深入的說(shuō)明，而Java方面的資料就很多了，參看深入理解JVM—JVM內(nèi)存模型
其它參考資料：
網(wǎng)絡(luò)通信socket連接數(shù)上限
轉(zhuǎn)載請(qǐng)注明本文出處：http://www.rzrgm.cn/newton/p/9776821.html
浙公網(wǎng)安備 33010602011771號(hào)