前言:
Thrift作為Facebook開源的RPC框架, 通過IDL中間語言, 並借助代碼生成引擎生成各種主流語言的rpc框架服務端/客戶端代碼. 不過Thrift的實現, 簡單使用離實際生產環境還是有一定距離, 本系列將對Thrift作代碼解讀和框架擴充, 使得它更加貼近生產環境. 本文主要講解Thrift的高性能網絡框架模型, 講解各種網絡模型的特點和區別.
Thrift 高性能網絡服務模型
1). TServer類層次體系
TSimpleServer/TThreadPoolServer是阻塞服務模型
TNonblockingServer/THsHaServer/TThreadedSelectotServer是非阻塞服務模型(NIO)
2). TServer抽象類的定義
內部靜態類Args的定義, 用於TServer類用於串聯軟件棧(傳輸層, 協議層, 處理層)
public abstract class TServer {
public static class Args extends AbstractServerArgs<Args> {
public Args(TServerTransport transport) {
super(transport);
}
}
public static abstract class AbstractServerArgs<T extends AbstractServerArgs<T>> {
public AbstractServerArgs(TServerTransport transport);
public T processorFactory(TProcessorFactory factory);
public T processor(TProcessor processor);
public T transportFactory(TTransportFactory factory);
public T protocolFactory(TProtocolFactory factory);
}
}
TServer類定義的抽象類
public abstract class TServer {
public abstract void serve();
public void stop();
public boolean isServing();
public void setServerEventHandler(TServerEventHandler eventHandler);
}
評注:
抽象函數serve由具體的TServer實例來實現, 而並非所有的服務都需要優雅的退出, 因此stop沒有被定義為抽象
3). TSimpleServer
TSimpleServer實現, 正如其名Simple, 其實現非常的簡單, 是個單線程阻塞模型, 只適合測試開發使用
抽象的代碼可簡單描述如下:
// *) server socket進行監聽
serverSocket.listen();
while ( isServing() ) {
// *) 接受socket鏈接
client = serverSocket.accept();
// *) 封裝處理器
processor = factory.getProcess(client);
while ( true ) {
// *) 阻塞處理rpc的輸入/輸出
if ( !processor.process(input, output) ) {
break;
}
}
}
4). ThreadPoolServer
ThreadPoolServer解決了TSimple不支持並發和多連接的問題, 引入了線程池. 實現的模型是One Thread Per Connection
線程池代碼片段:
private static ExecutorService createDefaultExecutorService(Args args) {
SynchronousQueue<Runnable> executorQueue =
new SynchronousQueue<Runnable>();
return new ThreadPoolExecutor(args.minWorkerThreads,
args.maxWorkerThreads,
60,
TimeUnit.SECONDS,
executorQueue);
}
評注:
采用同步隊列(SynchronousQueue), 線程池采用能線程數可伸縮的模式.
主線程循環
setServing(true);
while (!stopped_) {
try {
TTransport client = serverTransport_.accept();
WorkerProcess wp = new WorkerProcess(client);
executorService_.execute(wp);
} catch (TTransportException ttx) {
}
}
評注:
拆分了監聽線程(accept)和處理客戶端連接的工作線程(worker), 監聽線程每接到一個客戶端, 就投給線程池去處理. 這種模型能提高並發度, 但並發數取決於線程數, IO依舊阻塞, 從而限制該服務的服務能力.
5). TNonblockingServer
TNonblockingServer采用NIO的模式, 借助Channel/Selector機制, 采用IO事件模型來處理.
private void select() {
try {
selector.select(); // wait for io events.
// process the io events we received
Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
while (!stopped_ && selectedKeys.hasNext()) {
SelectionKey key = selectedKeys.next();
selectedKeys.remove();
if (key.isAcceptable()) {
handleAccept(); // deal with accept
} else if (key.isReadable()) {
handleRead(key); // deal with reads
} else if (key.isWritable()) {
handleWrite(key); // deal with writes
}
}
} catch (IOException e) {
}
}
評注:
select代碼里對accept/read/write等IO事件進行監控和處理, 唯一可惜的這個單線程處理. 當遇到handler里有阻塞的操作時, 會導致整個服務被阻塞住.
6). THsHaServer
鑒於TNonblockingServer的缺點, THsHa引入了線程池去處理, 其模型把讀寫任務放到線程池去處理.
HsHa是: Half-sync/Half-async的處理模式, Half-aysnc是在處理IO事件上(accept/read/write io), Half-sync用於handler對rpc的同步處理上.
7). TThreadedSelectorServer
TThreadedSelectorServer是最成熟,也是被業界所推崇的RPC服務模型
TThreadedSelectorServer是對以上NonblockingServer的擴充, 其分離了Accept和Read/Write的Selector線程, 同時引入Worker工作線程池. 它也是種Half-sync/Half-async的服務模型.
總結:
MainReactor就是Accept線程, 用於監聽客戶端連接, SubReactor采用IO事件線程(多個), 主要負責對所有客戶端的IO讀寫事件進行處理. 而Worker工作線程主要用於處理每個rpc請求的handler回調處理(這部分是同步的).
問題:
這邊提幾個小小的問題, 考考讀者?
1). Java NIO中 Selector采用什么方式實現? c++中的select/poll/epool? 如果是epool的話, 采用的是水平觸發,還是邊緣觸發?
2). 這邊非阻塞模型是HsHa, 有沒有全異步的模式? 為何通用的模型是采用TThreadedSelectorServer這種模式呢?
期待你的回答, 也敬請關注后續的文章.