網路程式設計基礎(5) : IO多路複用(多Reactor)(主從式Reactor)
阿新 • • 發佈：2019-01-26
1. 介紹
把Reactor拆成兩個角色Main Reactor及Sub Reactor，以提升效能與資源利用率。
Main Reactor：負責監聽外部的連線請求，並派發給Acceptor處理。故Main Reactor中的selector只有註冊OP_ACCEPT事件，也只能監聽OP_ACCEPT事件。
Acceptor接受連線後會給client繫結一個Handler並註冊IO事件到Sub Reactor上監聽，
對於有多個Sub Reactor的情況下，IO事件選擇註冊給哪個Sub Reactor則是採用Round-robin的機制來分配。
Sub Reactor：負責監聽IO事件，並派發IO事件給Handler處理。Sub Reactor執行緒的數量可以設定為CPU核心數。
2. 網路模型
3. 原始碼
Main Reactor
[TCPReactor.java]
// 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; // mainReactor用的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(ssc)); // 給定key一個附加的Acceptor物件 } @Override public void run() { while (!Thread.interrupted()) { // 線上程被中斷前持續執行 System.out.println("mainReactor waiting for new event on port: " + ssc.socket().getLocalPort() + "..."); try { if (selector.select() == 0) // 若沒有事件就緒則不往下執行 continue; } catch (IOException e) { 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(); } }
Acceptor
[Acceptor.java]// 接受連線請求執行緒 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; // mainReactor監聽的socket通道 private final int cores = Runtime.getRuntime().availableProcessors(); // 取得CPU核心數 private final Selector[] selectors = new Selector[cores]; // 建立核心數個selector給subReactor用 private int selIdx = 0; // 當前可使用的subReactor索引 private TCPSubReactor[] r = new TCPSubReactor[cores]; // subReactor執行緒 private Thread[] t = new Thread[cores]; // subReactor執行緒 public Acceptor(ServerSocketChannel ssc) throws IOException { this.ssc = ssc; // 建立多個selector以及多個subReactor執行緒 for (int i = 0; i < cores; i++) { selectors[i] = Selector.open(); r[i] = new TCPSubReactor(selectors[i], ssc, i); t[i] = new Thread(r[i]); t[i].start(); } } @Override public synchronized void run() { try { SocketChannel sc = ssc.accept(); // 接受client連線請求 System.out.println(sc.socket().getRemoteSocketAddress().toString() + " is connected."); if (sc != null) { sc.configureBlocking(false); // 設定為非阻塞 r[selIdx].setRestart(true); // 暫停執行緒 selectors[selIdx].wakeup(); // 使一個阻塞住的selector操作立即返回 SelectionKey sk = sc.register(selectors[selIdx], SelectionKey.OP_READ); // SocketChannel向selector[selIdx]註冊一個OP_READ事件，然後返回該通道的key selectors[selIdx].wakeup(); // 使一個阻塞住的selector操作立即返回 r[selIdx].setRestart(false); // 重啟執行緒 sk.attach(new TCPHandler(sk, sc)); // 給定key一個附加的TCPHandler物件 if (++selIdx == selectors.length) selIdx = 0; } } catch (IOException e) { e.printStackTrace(); } } }
Sub Reactor
Sub Reactor在實作上有個重點要注意，
當一個監聽中而阻塞住的selector由於Acceptor需要註冊新的IO事件到該selector上時，
Acceptor會呼叫selector的wakeup()函式喚醒阻塞住的selector，以註冊新IO事件後再繼續監聽。
但Sub Reactor中迴圈呼叫selector.select()的執行緒迴圈可能會因為迴圈太快，導致selector被喚醒後再度於IO事件成功註冊前被呼叫selector.select()而阻塞住，
因此我們需要給Sub Reactor執行緒迴圈設定一個flag來控制，
讓selector被喚醒後不會馬上進入下回合調用selector.select()的Sub Reactor執行緒迴圈，
等待我們將新的IO事件註冊完之後才能讓Sub Reactor執行緒繼續執行。
[TCPSubReactor.java]
package server;
import java.io.IOException;
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 TCPSubReactor implements Runnable {
private final ServerSocketChannel ssc;
private final Selector selector;
private boolean restart = false;
int num;
public TCPSubReactor(Selector selector, ServerSocketChannel ssc, int num) {
this.ssc = ssc;
this.selector = selector;
this.num = num;
}
@Override
public void run() {
while (!Thread.interrupted()) { // 線上程被中斷前持續執行
//System.out.println("ID:" + num
// + " subReactor waiting for new event on port: "
// + ssc.socket().getLocalPort() + "...");
System.out.println("waiting for restart");
while (!Thread.interrupted() && !restart) { // 線上程被中斷前以及被指定重啟前持續執行
try {
if (selector.select() == 0)
continue; // 若沒有事件就緒則不往下執行
} catch (IOException e) {
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();
}
public void setRestart(boolean restart) {
this.restart = restart;
}
}
Handler
[TCPHandler.java]
// Handler執行緒
package server;
import java.io.IOException;
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;
private static final int THREAD_COUNTING = 10;
private static ThreadPoolExecutor pool = new ThreadPoolExecutor(
THREAD_COUNTING, THREAD_COUNTING, 10, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>()); // 執行緒池
HandlerState state; // 以狀態模式實現Handler
public TCPHandler(SelectionKey sk, SocketChannel sc) {
this.sk = sk;
this.sc = sc;
state = new ReadState(); // 初始狀態設定為READING
pool.setMaximumPoolSize(32); // 設定執行緒池最大執行緒數
}
@Override
public void run() {
try {
state.handle(this, sk, sc, pool);
} catch (IOException e) {
System.out.println("[Warning!] A client has been closed.");
closeChannel();
}
}
public void closeChannel() {
try {
sk.cancel();
sc.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
public void setState(HandlerState state) {
this.state = state;
}
}
狀態介面
[HandlerState.java]
package server;
import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ThreadPoolExecutor;
public interface HandlerState {
public void changeState(TCPHandler h);
public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,
ThreadPoolExecutor pool) throws IOException ;
}
狀態實現類
[ReadState.java]package server;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ThreadPoolExecutor;
public class ReadState implements HandlerState{
private SelectionKey sk;
public ReadState() {
}
@Override
public void changeState(TCPHandler h) {
// TODO Auto-generated method stub
h.setState(new WorkState());
}
@Override
public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,
ThreadPoolExecutor pool) throws IOException { // read()
this.sk = sk;
// 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.");
h.closeChannel();
return;
}
String str = new String(arr); // 將讀取到的byte內容轉為字串型態
if ((str != null) && !str.equals(" ")) {
h.setState(new WorkState()); // 改變狀態(READING->WORKING)
pool.execute(new WorkerThread(h, str)); // do process in worker thread
System.out.println(sc.socket().getRemoteSocketAddress().toString()
+ " > " + str);
}
}
/*
* 執行邏輯處理之函式
*/
synchronized void process(TCPHandler h, String str) {
// do process(decode, logically process, encode)..
// ..
h.setState(new WriteState()); // 改變狀態(WORKING->SENDING)
this.sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊的事件
this.sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
}
/*
* 工作者執行緒
*/
class WorkerThread implements Runnable {
TCPHandler h;
String str;
public WorkerThread(TCPHandler h, String str) {
this.h = h;
this.str=str;
}
@Override
public void run() {
process(h, str);
}
}
}
package server;
import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ThreadPoolExecutor;
public class WorkState implements HandlerState {
public WorkState() {
}
@Override
public void changeState(TCPHandler h) {
// TODO Auto-generated method stub
h.setState(new WriteState());
}
@Override
public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,
ThreadPoolExecutor pool) throws IOException {
// TODO Auto-generated method stub
}
}
package server;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ThreadPoolExecutor;
public class WriteState implements HandlerState{
public WriteState() {
}
@Override
public void changeState(TCPHandler h) {
// TODO Auto-generated method stub
h.setState(new ReadState());
}
@Override
public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,
ThreadPoolExecutor pool) throws IOException { // send()
// 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位置為止之間的內容
}
h.setState(new ReadState()); // 改變狀態(SENDING->READING)
sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊的事件
sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
}
}
主程式
[Main.java]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);
new Thread(reactor).start();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
網路程式設計中有關於IO複用的系列介紹文章就到此結束。