前言

Thrift提供的網路服務模型：單執行緒、多執行緒、事件驅動，從另一個角度劃分為：阻塞服務模型、非阻塞服務模型。

• 阻塞服務模型：TSimpleServer、TThreadPoolServer。

• 非阻塞服務模型：TNonblockingServer、THsHaServer和TThreadedSelectorServer。

TServer類的層次關係：

正文

TServer

TServer定義了靜態內部類Args，Args繼承自抽象類AbstractServerArgs。AbstractServerArgs採用了建造者模式，向TServer提供各種工廠：

工廠屬性	工廠型別	作用
ProcessorFactory	TProcessorFactory	處理層工廠類，用於具體的TProcessor物件的建立
InputTransportFactory	TTransportFactory	傳輸層輸入工廠類，用於具體的TTransport物件的建立
OutputTransportFactory	TTransportFactory	傳輸層輸出工廠類，用於具體的TTransport物件的建立
InputProtocolFactory	TProtocolFactory	協議層輸入工廠類，用於具體的TProtocol物件的建立
OutputProtocolFactory	TProtocolFactory	協議層輸出工廠類，用於具體的TProtocol物件的建立

下面是TServer的部分核心程式碼：

public abstract class TServer {
    public static class Args extends org.apache.thrift.server.TServer.AbstractServerArgs<org.apache.thrift.server.TServer.Args> {
        public Args(TServerTransport transport) {
            super
 
(transport);
        }
    }

    public static abstract class AbstractServerArgs<T extends org.apache.thrift.server.TServer.AbstractServerArgs<T>> {
        final TServerTransport serverTransport;
        TProcessorFactory processorFactory;
        TTransportFactory inputTransportFactory = new TTransportFactory();
        TTransportFactory outputTransportFactory = new TTransportFactory();
        TProtocolFactory inputProtocolFactory = new TBinaryProtocol.Factory();
        TProtocolFactory outputProtocolFactory = new TBinaryProtocol.Factory();

        public AbstractServerArgs(TServerTransport transport) {
            serverTransport = transport;
        }
    }

    protected TProcessorFactory processorFactory_;
    protected TServerTransport serverTransport_;
    protected TTransportFactory inputTransportFactory_;
    protected TTransportFactory outputTransportFactory_;
    protected TProtocolFactory inputProtocolFactory_;
    protected TProtocolFactory outputProtocolFactory_;
    private boolean isServing;

    protected TServer(org.apache.thrift.server.TServer.AbstractServerArgs args) {
        processorFactory_ = args.processorFactory;
        serverTransport_ = args.serverTransport;
        inputTransportFactory_ = args.inputTransportFactory;
        outputTransportFactory_ = args.outputTransportFactory;
        inputProtocolFactory_ = args.inputProtocolFactory;
        outputProtocolFactory_ = args.outputProtocolFactory;
    }

    public abstract void serve();
    public void stop() {}

    public boolean isServing() {
        return isServing;
    }

    protected void setServing(boolean serving) {
        isServing = serving;
    }
}

TServer的三個方法：serve()、stop()和isServing()。serve()用於啟動服務，stop()用於關閉服務，isServing()用於檢測服務的起停狀態。

TServer的不同實現類的啟動方式不一樣，因此serve()定義為抽象方法。不是所有的服務都需要優雅的退出, 因此stop()方法沒有被定義為抽象。

TSimpleServer

TSimpleServer的工作模式採用最簡單的阻塞IO，實現方法簡潔明瞭，便於理解，但是一次只能接收和處理一個socket連線，效率比較低。它主要用於演示Thrift的工作過程，在實際開發過程中很少用到它。

(一) 工作流程

(二) 使用入門

服務端：

    ServerSocket serverSocket = new ServerSocket(ServerConfig.SERVER_PORT);
    TServerSocket serverTransport = new TServerSocket(serverSocket);
    HelloWorldService.Processor processor =
            new HelloWorldService.Processor<HelloWorldService.Iface>(new HelloWorldServiceImpl());
    TBinaryProtocol.Factory protocolFactory = new TBinaryProtocol.Factory();

    TSimpleServer.Args tArgs = new TSimpleServer.Args(serverTransport);
    tArgs.processor(processor);
    tArgs.protocolFactory(protocolFactory);
    // 簡單的單執行緒服務模型 一般用於測試
    TServer tServer = new TSimpleServer(tArgs);
    System.out.println("Running Simple Server");
    tServer.serve();

客戶端：

    TTransport transport = new TSocket(ServerConfig.SERVER_IP, ServerConfig.SERVER_PORT, ServerConfig.TIMEOUT);
    TProtocol protocol = new TBinaryProtocol(transport);
    HelloWorldService.Client client = new HelloWorldService.Client(protocol);
    transport.open();

    String result = client.say("Leo");
    System.out.println("Result =: " + result);
    transport.close();

(三) 原始碼分析

檢視上述流程的原始碼，即TSimpleServer.java中的serve()方法如下：

serve()方法的操作：

1. 設定TServerSocket的listen()方法啟動連線監聽。
2. 以阻塞的方式接受客戶端地連線請求，每進入一個連線即為其建立一個通道TTransport物件。
3. 為客戶端建立處理器物件、輸入傳輸通道物件、輸出傳輸通道物件、輸入協議物件和輸出協議物件。
4. 通過TServerEventHandler物件處理具體的業務請求。

ThreadPoolServer

TThreadPoolServer模式採用阻塞socket方式工作，主執行緒負責阻塞式監聽是否有新socket到來，具體的業務處理交由一個執行緒池來處理。

(一) 工作流程

(二) 使用入門

服務端：

    ServerSocket serverSocket = new ServerSocket(ServerConfig.SERVER_PORT);
    TServerSocket serverTransport = new TServerSocket(serverSocket);
    HelloWorldService.Processor<HelloWorldService.Iface> processor =
            new HelloWorldService.Processor<>(new HelloWorldServiceImpl());

    TBinaryProtocol.Factory protocolFactory = new TBinaryProtocol.Factory();
    TThreadPoolServer.Args ttpsArgs = new TThreadPoolServer.Args(serverTransport);
    ttpsArgs.processor(processor);
    ttpsArgs.protocolFactory(protocolFactory);

    // 執行緒池服務模型 使用標準的阻塞式IO 預先建立一組執行緒處理請求
    TServer ttpsServer = new TThreadPoolServer(ttpsArgs);
    System.out.println("Running ThreadPool Server");
    ttpsServer.serve();

客戶端：

    TTransport transport = new TSocket(ServerConfig.SERVER_IP, ServerConfig.SERVER_PORT, ServerConfig.TIMEOUT);
    TProtocol protocol = new TBinaryProtocol(transport);
    HelloWorldService.Client client = new HelloWorldService.Client(protocol);

    transport.open();
    String result = client.say("ThreadPoolClient");
    System.out.println("Result =: " + result);
    transport.close();

(三) 原始碼分析

ThreadPoolServer解決了TSimpleServer不支援併發和多連線的問題，引入了執行緒池。實現的模型是One Thread Per Connection。檢視上述流程的原始碼，先檢視執行緒池的程式碼片段：

TThreadPoolServer.java中的serve()方法如下：

serve()方法的操作：

1. 設定TServerSocket的listen()方法啟動連線監聽。
2. 以阻塞的方式接受客戶端的連線請求，每進入一個連線，將通道物件封裝成一個WorkerProcess物件(WorkerProcess實現了Runnabel介面)，並提交到執行緒池。
3. WorkerProcess的run()方法負責業務處理，為客戶端建立了處理器物件、輸入傳輸通道物件、輸出傳輸通道物件、輸入協議物件和輸出協議物件。
4. 通過TServerEventHandler物件處理具體的業務請求。

WorkerProcess的run()方法：

(四) 優缺點

TThreadPoolServer模式的優點

拆分了監聽執行緒(Accept Thread)和處理客戶端連線的工作執行緒(Worker Thread)，資料讀取和業務處理都交給執行緒池處理。因此在併發量較大時新連線也能夠被及時接受。

執行緒池模式比較適合伺服器端能預知最多有多少個客戶端併發的情況，這時每個請求都能被業務執行緒池及時處理，效能也非常高。

TThreadPoolServer模式的缺點

執行緒池模式的處理能力受限於執行緒池的工作能力，當併發請求數大於執行緒池中的執行緒數時，新請求也只能排隊等待。

TNonblockingServer

TNonblockingServer模式也是單執行緒工作，但是採用NIO的模式，藉助Channel/Selector機制, 採用IO事件模型來處理。

所有的socket都被註冊到selector中，在一個執行緒中通過seletor迴圈監控所有的socket。

每次selector迴圈結束時，處理所有的處於就緒狀態的socket，對於有資料到來的socket進行資料讀取操作，對於有資料傳送的socket則進行資料傳送操作，對於監聽socket則產生一個新業務socket並將其註冊到selector上。

注意：TNonblockingServer要求底層的傳輸通道必須使用TFramedTransport。

(一) 工作流程

(二) 使用入門

服務端：

    TProcessor tprocessor = new HelloWorldService.Processor<HelloWorldService.Iface>(new HelloWorldServiceImpl());
    TNonblockingServerSocket tnbSocketTransport = new TNonblockingServerSocket(ServerConfig.SERVER_PORT);

    TNonblockingServer.Args tnbArgs = new TNonblockingServer.Args(tnbSocketTransport);
    tnbArgs.processor(tprocessor);
    tnbArgs.transportFactory(new TFramedTransport.Factory());
    tnbArgs.protocolFactory(new TCompactProtocol.Factory());

    // 使用非阻塞式IO服務端和客戶端需要指定TFramedTransport資料傳輸的方式
    TServer server = new TNonblockingServer(tnbArgs);
    System.out.println("Running Non-blocking Server");
    server.serve();

客戶端：

    TTransport transport = new TFramedTransport(new TSocket(ServerConfig.SERVER_IP, ServerConfig.SERVER_PORT, ServerConfig.TIMEOUT));
    // 協議要和服務端一致
    TProtocol protocol = new TCompactProtocol(transport);
    HelloWorldService.Client client = new HelloWorldService.Client(protocol);
    transport.open();

    String result = client.say("NonBlockingClient");
    System.out.println("Result =: " + result);
    transport.close();

(三) 原始碼分析

TNonblockingServer繼承於AbstractNonblockingServer，這裡我們更關心基於NIO的selector部分的關鍵程式碼。

(四) 優缺點

TNonblockingServer模式優點

相比於TSimpleServer效率提升主要體現在IO多路複用上，TNonblockingServer採用非阻塞IO，對accept/read/write等IO事件進行監控和處理，同時監控多個socket的狀態變化。

TNonblockingServer模式缺點

TNonblockingServer模式在業務處理上還是採用單執行緒順序來完成。在業務處理比較複雜、耗時的時候，例如某些介面函式需要讀取資料庫執行時間較長，會導致整個服務被阻塞住，此時該模式效率也不高，因為多個呼叫請求任務依然是順序一個接一個執行。

THsHaServer

鑑於TNonblockingServer的缺點，THsHaServer繼承於TNonblockingServer，引入了執行緒池提高了任務處理的併發能力。THsHaServer是半同步半非同步(Half-Sync/Half-Async)的處理模式，Half-Aysnc用於IO事件處理(Accept/Read/Write)，Half-Sync用於業務handler對rpc的同步處理上。

注意：THsHaServer和TNonblockingServer一樣，要求底層的傳輸通道必須使用TFramedTransport。

(一) 工作流程

(二) 使用入門

服務端：

    TNonblockingServerSocket tnbSocketTransport = new TNonblockingServerSocket(ServerConfig.SERVER_PORT);
    TProcessor tprocessor = new HelloWorldService.Processor<HelloWorldService.Iface>(new HelloWorldServiceImpl());
    // 半同步半非同步
    THsHaServer.Args thhsArgs = new THsHaServer.Args(tnbSocketTransport);
    thhsArgs.processor(tprocessor);
    thhsArgs.transportFactory(new TFramedTransport.Factory());
    thhsArgs.protocolFactory(new TBinaryProtocol.Factory());

    TServer server = new THsHaServer(thhsArgs);
    System.out.println("Running HsHa Server");
    server.serve();

客戶端：

    TTransport transport = new TFramedTransport(new TSocket(ServerConfig.SERVER_IP, ServerConfig.SERVER_PORT, ServerConfig.TIMEOUT));
    // 協議要和服務端一致
    TProtocol protocol = new TBinaryProtocol(transport);
    HelloWorldService.Client client = new HelloWorldService.Client(protocol);
    transport.open();

    String result = client.say("HsHaClient");
    System.out.println("Result =: " + result);
    transport.close();

(三) 原始碼分析

THsHaServer繼承於TNonblockingServer，新增了執行緒池併發處理工作任務的功能，檢視執行緒池的相關程式碼：

任務執行緒池的建立過程：

下文的TThreadedSelectorServer囊括了THsHaServer的大部分特性，原始碼分析可參考TThreadedSelectorServer。

(四) 優缺點

THsHaServer的優點

THsHaServer與TNonblockingServer模式相比，THsHaServer在完成資料讀取之後，將業務處理過程交由一個執行緒池來完成，主執行緒直接返回進行下一次迴圈操作，效率大大提升。

THsHaServer的缺點

主執行緒仍然需要完成所有socket的監聽接收、資料讀取和資料寫入操作。當併發請求數較大時，且傳送資料量較多時，監聽socket上新連線請求不能被及時接受。

TThreadedSelectorServer

TThreadedSelectorServer是對THsHaServer的一種擴充，它將selector中的讀寫IO事件(read/write)從主執行緒中分離出來。同時引入worker工作執行緒池，它也是種Half-Sync/Half-Async的服務模型。

TThreadedSelectorServer模式是目前Thrift提供的最高階的執行緒服務模型，它內部有如果幾個部分構成：

1. 一個AcceptThread執行緒物件，專門用於處理監聽socket上的新連線。
2. 若干個SelectorThread物件專門用於處理業務socket的網路I/O讀寫操作，所有網路資料的讀寫均是有這些執行緒來完成。
3. 一個負載均衡器SelectorThreadLoadBalancer物件，主要用於AcceptThread執行緒接收到一個新socket連線請求時，決定將這個新連線請求分配給哪個SelectorThread執行緒。
4. 一個ExecutorService型別的工作執行緒池，在SelectorThread執行緒中，監聽到有業務socket中有呼叫請求過來，則將請求資料讀取之後，交給ExecutorService執行緒池中的執行緒完成此次呼叫的具體執行。主要用於處理每個rpc請求的handler回撥處理(這部分是同步的)。

(一) 工作流程

(二) 使用入門

服務端：

    TNonblockingServerSocket serverSocket = new TNonblockingServerSocket(ServerConfig.SERVER_PORT);
    TProcessor processor = new HelloWorldService.Processor<HelloWorldService.Iface>(new HelloWorldServiceImpl());
    // 多執行緒半同步半非同步
    TThreadedSelectorServer.Args ttssArgs = new TThreadedSelectorServer.Args(serverSocket);
    ttssArgs.processor(processor);
    ttssArgs.protocolFactory(new TBinaryProtocol.Factory());
    // 使用非阻塞式IO時 服務端和客戶端都需要指定資料傳輸方式為TFramedTransport
    ttssArgs.transportFactory(new TFramedTransport.Factory());

    // 多執行緒半同步半非同步的服務模型
    TThreadedSelectorServer server = new TThreadedSelectorServer(ttssArgs);
    System.out.println("Running ThreadedSelector Server");
    server.serve();

客戶端：

for (int i = 0; i < 10; i++) {
    new Thread("Thread " + i) {
        @Override
        public void run() {
            // 設定傳輸通道 對於非阻塞服務 需要使用TFramedTransport(用於將資料分塊傳送)
            for (int j = 0; j < 10; j++) {
                TTransport transport = null;
                try {
                    transport = new TFramedTransport(new TSocket(ServerConfig.SERVER_IP, ServerConfig.SERVER_PORT, ServerConfig.TIMEOUT));
                    TProtocol protocol = new TBinaryProtocol(transport);
                    HelloWorldService.Client client = new HelloWorldService.Client(protocol);
                    transport.open();
                    String result = client.say("ThreadedSelector Client");
                    System.out.println("Result =: " + result);
                    transport.close();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 關閉傳輸通道
                    transport.close();
                }
            }
        }
    }.start();
}

(三) 核心程式碼

以上工作流程的三個元件AcceptThread、SelectorThread和ExecutorService在原始碼中的定義如下：

TThreadedSelectorServer模式中有一個專門的執行緒AcceptThread用於處理新連線請求，因此能夠及時響應大量併發連線請求；另外它將網路I/O操作分散到多個SelectorThread執行緒中來完成，因此能夠快速對網路I/O進行讀寫操作，能夠很好地應對網路I/O較多的情況。

TThreadedSelectorServer預設引數定義如下：

• 負責網路IO讀寫的selector預設執行緒數(selectorThreads)：2
• 負責業務處理的預設工作執行緒數(workerThreads)：5
• 工作執行緒池單個執行緒的任務佇列大小(acceptQueueSizePerThread)：4

建立、初始化並啟動AcceptThread和SelectorThreads，同時啟動selector執行緒的負載均衡器(selectorThreads)。

AcceptThread原始碼

AcceptThread繼承於Thread，可以看出包含三個重要的屬性：非阻塞式傳輸通道(TNonblockingServerTransport)、NIO選擇器(acceptSelector)和選擇器執行緒負載均衡器(threadChooser)。

檢視AcceptThread的run()方法，可以看出accept執行緒一旦啟動，就會不停地呼叫select()方法：

檢視select()方法，acceptSelector選擇器等待IO事件的到來，拿到SelectionKey即檢查是不是accept事件。如果是，通過handleAccept()方法接收一個新來的連線；否則，如果是IO讀寫事件，AcceptThread不作任何處理，交由SelectorThread完成。

在handleAccept()方法中，先通過doAccept()去拿連線通道，然後Selector執行緒負載均衡器選擇一個Selector執行緒，完成接下來的IO讀寫事件。

接下來繼續檢視doAddAccept()方法的實現，毫無懸念，它進一步呼叫了SelectorThread的addAcceptedConnection()方法，把非阻塞傳輸通道物件傳遞給選擇器執行緒做進一步的IO讀寫操作。

SelectorThreadLoadBalancer原始碼

SelectorThreadLoadBalancer如何建立？

SelectorThreadLoadBalancer是一個基於輪詢演算法的Selector執行緒選擇器，通過執行緒迭代器為新進來的連線順序分配SelectorThread。

SelectorThread原始碼

SelectorThread和AcceptThread一樣，是TThreadedSelectorServer的一個成員內部類，每個SelectorThread執行緒物件內部都有一個阻塞式的佇列，用於存放該執行緒被接收的連線通道。

阻塞佇列的大小可由建構函式指定：

上面看到，在AcceptThread的doAddAccept()方法中呼叫了SelectorThread的addAcceptedConnection()方法。

這個方法做了兩件事：

1. 將被此SelectorThread執行緒接收的連線通道放入阻塞佇列中。
2. 通過wakeup()方法喚醒SelectorThread中的NIO選擇器selector。

既然SelectorThread也是繼承於Thread，檢視其run()方法的實現：

SelectorThread方法的select()監聽IO事件，僅僅用於處理資料讀取和資料寫入。如果連線有資料可讀，讀取並以frame的方式快取；如果需要向連線中寫入資料，快取併發送客戶端的資料。且在資料讀寫處理完成後，需要向NIO的selector清空和登出自身的SelectionKey。

• 資料寫操作完成以後，整個rpc呼叫過程也就結束了，handleWrite()方法如下：

相關推薦