reactor模式:單執行緒的reactor模式
阿新 • • 發佈:2019-02-18
reactor模式稱之為響應器模式,常用於nio的網路通訊框架,其服務架構圖如下
不同於傳統IO的序列排程方式,NIO把整個服務請求分為五個階段
read:接收到請求,讀取資料
decode:解碼資料
compute:業務邏輯處理
encode:返回資料編碼
send:傳送資料
其中,以read和send階段IO最為頻繁
程式碼實現
// Reactor執行緒 package server; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.util.Iterator; import java.util.Set; public class TCPReactor implements Runnable { private final ServerSocketChannel ssc; private final Selector selector; public TCPReactor(int port) throws IOException { selector = Selector.open(); ssc = ServerSocketChannel.open(); InetSocketAddress addr = new InetSocketAddress(port); ssc.socket().bind(addr); // 在ServerSocketChannel繫結監聽埠 ssc.configureBlocking(false); // 設定ServerSocketChannel為非阻塞 SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊一個OP_ACCEPT事件,然後返回該通道的key sk.attach(new Acceptor(selector, ssc)); // 給定key一個附加的Acceptor物件 } @Override public void run() { while (!Thread.interrupted()) { // 線上程被中斷前持續執行 System.out.println("Waiting for new event on port: " + ssc.socket().getLocalPort() + "..."); try { if (selector.select() == 0) // 若沒有事件就緒則不往下執行 continue; } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合 Iterator<SelectionKey> it = selectedKeys.iterator(); while (it.hasNext()) { dispatch((SelectionKey) (it.next())); // 根據事件的key進行排程 it.remove(); } } } /* * name: dispatch(SelectionKey key) * description: 排程方法,根據事件繫結的物件開新執行緒 */ private void dispatch(SelectionKey key) { Runnable r = (Runnable) (key.attachment()); // 根據事件之key繫結的物件開新執行緒 if (r != null) r.run(); } }
// 接受連線請求執行緒 package server; import java.io.IOException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; public class Acceptor implements Runnable { private final ServerSocketChannel ssc; private final Selector selector; public Acceptor(Selector selector, ServerSocketChannel ssc) { this.ssc=ssc; this.selector=selector; } @Override public void run() { try { SocketChannel sc= ssc.accept(); // 接受client連線請求 System.out.println(sc.socket().getRemoteSocketAddress().toString() + " is connected."); if(sc!=null) { sc.configureBlocking(false); // 設定為非阻塞 SelectionKey sk = sc.register(selector, SelectionKey.OP_READ); // SocketChannel向selector註冊一個OP_READ事件,然後返回該通道的key selector.wakeup(); // 使一個阻塞住的selector操作立即返回 sk.attach(new TCPHandler(sk, sc)); // 給定key一個附加的TCPHandler物件 } } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
// Handler執行緒 package server; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; public class TCPHandler implements Runnable { private final SelectionKey sk; private final SocketChannel sc; int state; public TCPHandler(SelectionKey sk, SocketChannel sc) { this.sk = sk; this.sc = sc; state = 0; // 初始狀態設定為READING } @Override public void run() { try { if (state == 0) read(); // 讀取網路資料 else send(); // 傳送網路資料 } catch (IOException e) { System.out.println("[Warning!] A client has been closed."); closeChannel(); } } private void closeChannel() { try { sk.cancel(); sc.close(); } catch (IOException e1) { e1.printStackTrace(); } } private synchronized void read() throws IOException { // non-blocking下不可用Readers,因為Readers不支援non-blocking byte[] arr = new byte[1024]; ByteBuffer buf = ByteBuffer.wrap(arr); int numBytes = sc.read(buf); // 讀取字串 if(numBytes == -1) { System.out.println("[Warning!] A client has been closed."); closeChannel(); return; } String str = new String(arr); // 將讀取到的byte內容轉為字串型態 if ((str != null) && !str.equals(" ")) { process(str); // 邏輯處理 System.out.println(sc.socket().getRemoteSocketAddress().toString() + " > " + str); state = 1; // 改變狀態 sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊的事件 sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回 } } private void send() throws IOException { // get message from message queue String str = "Your message has sent to " + sc.socket().getLocalSocketAddress().toString() + "\r\n"; ByteBuffer buf = ByteBuffer.wrap(str.getBytes()); // wrap自動把buf的position設為0,所以不需要再flip() while (buf.hasRemaining()) { sc.write(buf); // 回傳給client回應字串,傳送buf的position位置 到limit位置為止之間的內容 } state = 0; // 改變狀態 sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊的事件 sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回 } void process(String str) { // do process(decode, logically process, encode).. // .. } }
package server;
import java.io.IOException;
public class Main {
public static void main(String[] args) {
// TODO Auto-generated method stub
try {
TCPReactor reactor = new TCPReactor(1333);
reactor.run();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
客戶端程式碼
package main.pkg;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
public class Client {
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
String hostname=args[0];
int port = Integer.parseInt(args[1]);
//String hostname="127.0.0.1";
//int port=1333;
System.out.println("Connecting to "+ hostname +":"+port);
try {
Socket client = new Socket(hostname, port); // 連線至目的地
System.out.println("Connected to "+ hostname);
PrintWriter out = new PrintWriter(client.getOutputStream());
BufferedReader in = new BufferedReader(new InputStreamReader(client.getInputStream()));
BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));
String input;
while((input=stdIn.readLine()) != null) { // 讀取輸入
out.println(input); // 傳送輸入的字串
out.flush(); // 強制將緩衝區內的資料輸出
if(input.equals("exit"))
{
break;
}
System.out.println("server: "+in.readLine());
}
client.close();
System.out.println("client stop.");
} catch (UnknownHostException e) {
// TODO Auto-generated catch block
System.err.println("Don't know about host: " + hostname);
} catch (IOException e) {
// TODO Auto-generated catch block
System.err.println("Couldn't get I/O for the socket connection");
}
}
}
程式碼解讀:
1.建立TCPReactor 類的例項,啟動埠監聽
2.Acceptor 類只用於處理接受請求的時候,後續的讀寫跟其無任何關係
3.TCPReactor.run( )一直在進行,後續selectionkey有變動,會監聽到,一直執行dispatch方法
最後提醒一點,從效能來說,單執行緒的reactor沒過多的提升,因為IO和CPU的速度還是嚴重不匹配
參考文章:
https://blog.csdn.net/yehjordan/article/details/51012833