thrift學習筆記（一） thrift簡介及第一個helloword程式
簡介
facebook開源的RPC框架，秉承了Facebook一貫的只管拉屎不管擦屁股的作風.
Thrift是一個軟體框架，用來進行可擴充套件且跨語言的服務的開發。它結合了功能強大的軟體堆疊和程式碼生成引擎，以構建在 C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, JavaScript, Node.js, Smalltalk, and OCaml 等等程式語言間無縫結合的、高效的服務。
Thrift最初由facebook開發，07年四月開放原始碼，08年5月進入apache孵化器。thrift允許你定義一個簡單的定義檔案中的資料型別和服務介面。以作為輸入檔案，編譯器生成程式碼用來方便地生成RPC客戶端和伺服器通訊的無縫跨程式語言。
Windows下安裝thrift
- 下載完成後放到【C:\Program Files\Thrift】目錄下，並將名字改為【thrift.exe】。注：這裡改名只是為了敲命令列方便
- 將目錄【C:\Program Files\Thrift】新增到環境變數即可
到此thrift安裝完成，簡單吧，我們來驗證一下。
編寫helloword服務介面描述檔案程式碼：
Hello.thrift
namespace java service.demo
service Hello{
string helloString(1:string para)
i32 helloInt(1 命名為【Hello.thrift】放到【D:\workSpaceIdea\thrift】目錄下，執行命令【thrift –gen java Hello.thrift】生成java檔案：
建立第一個thrift程式
建立一個maven專案
目錄結構：
pom 檔案：
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project 注：Hello.java是通過thrift自動生成的
HelloServiceImpl.java是實現類：
package service.demo;
import org.apache.thrift.TException;
public class HelloServiceImpl implements Hello.Iface {
@Override
public boolean helloBoolean(boolean para) throws TException {
return para;
}
@Override
public int helloInt(int para) throws TException {
try {
Thread.sleep(20000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return para;
}
@Override
public String helloNull() throws TException {
return null;
}
@Override
public String helloString(String para) throws TException {
System.out.println("入參：" + para);
return "hello " + para;
}
@Override
public void helloVoid() throws TException {
System.out.println("Hello World");
}
}使用執行緒池模式的server:
HelloServiceServer.java
package server;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransportException;
import service.demo.Hello;
import service.demo.HelloServiceImpl;
public class HelloServiceServer {
/**
* 啟動 Thrift 伺服器
* @param args
*/
public static void main(String[] args) {
try {
// 設定服務埠為 7911
TServerSocket serverTransport = new TServerSocket(7911);
// 設定協議工廠為 TBinaryProtocol.Factory
TBinaryProtocol.Factory proFactory = new TBinaryProtocol.Factory();
// 關聯處理器與 Hello 服務的實現
TProcessor processor = new Hello.Processor(new HelloServiceImpl());
TThreadPoolServer.Args args1 = new TThreadPoolServer.Args(serverTransport);
args1.processor(processor);
args1.protocolFactory(proFactory);
TServer server = new TThreadPoolServer(args1);
System.out.println("Start server on port 7911...");
server.serve();
} catch (TTransportException e) {
e.printStackTrace();
}
}
}HelloServiceClient.java
package client;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import service.demo.Hello;
public class HelloServiceClient {
/**
* 呼叫 Hello 服務
* @param args
*/
public static void main(String[] args) {
try {
// 設定呼叫的服務地址為本地，埠為 7911
TTransport transport = new TSocket("localhost", 7911);
transport.open();
// 設定傳輸協議為 TBinaryProtocol
TProtocol protocol = new TBinaryProtocol(transport);
Hello.Client client = new Hello.Client(protocol);
// 呼叫服務的 helloVoid 方法
// client.helloVoid();
String resp = client.helloString("wertyuiopsdfghjkl;");
System.out.println(resp);
transport.close();
} catch (TTransportException e) {
e.printStackTrace();
} catch (TException e) {
e.printStackTrace();
}
}
}執行
執行server以後，執行client，會在server的console控制檯看到輸出內容
注意：
客戶端和服務端要使用同一中 Protocol 和 Transport，否則會丟擲異常
TNonblockingServer 服務模型 的 server
HelloServiceServer1.java
package server;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.TNonblockingServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException;
import service.demo.Hello;
import service.demo.HelloServiceImpl;
public class HelloServiceServer1 {
public static final int SERVER_PORT = 8090;
/**
* 啟動 Thrift 伺服器
* @param args
*/
public static void main(String[] args) {
try {
System.out.println("HelloWorld TNonblockingServer start ....");
TProcessor tprocessor = new Hello.Processor(new HelloServiceImpl());
TNonblockingServerSocket tnbSocketTransport = new TNonblockingServerSocket(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);
server.serve();
} catch (TTransportException e) {
e.printStackTrace();
}
}
}HelloServiceClient1.java
package client;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import service.demo.Hello;
public class HelloServiceClient1 {
public static final String SERVER_IP = "localhost";
public static final int SERVER_PORT = 8090;
public static final int TIMEOUT = 30000;
/**
* 呼叫 Hello 服務 * @param args
*/
public static void main(String[] args) {
TTransport transport = null;
try {
transport = new TFramedTransport(new TSocket(SERVER_IP, SERVER_PORT, TIMEOUT));
// 協議要和服務端一致
TProtocol protocol = new TCompactProtocol(transport);
Hello.Client client = new Hello.Client(protocol);
transport.open();
String result = client.helloString("hfjtydiyghfkjdhftgh");
System.out.println("Thrify client result =: " + result);
} catch (TTransportException e) {
e.printStackTrace();
} catch (TException e) {
e.printStackTrace();
} finally {
if (null != transport) {
transport.close();
}
}
}
}我在開發中遇到的錯誤及解決辦法
- java.lang.NoSuchMethodError: org.apache.thrift.EncodingUtils.setBit(BIZ)B
出現這個問題的原因是dubbox本身整合的thrift版本是0.8.0，我新增的版本是0.9.3導致的，解決方法：去掉對0.8.0版本的依賴
這個問題是因為我thrift的IDL描述檔案多了個逗號，在thrift生產java檔案的時候沒有報錯，但是在dubbox執行的時候就出錯了。
IDL檔案簡介
- 基本型別
- bool：布林值，true 或 false，對應 Java 的 boolean
- byte：8 位有符號整數，對應 Java 的 byte
- i16：16 位有符號整數，對應 Java 的 short
- i32：32 位有符號整數，對應 Java 的 int
- i64：64 位有符號整數，對應 Java 的 long
- double：64 位浮點數，對應 Java 的 double
- string：utf-8編碼的字串，對應 Java 的 String
- 結構體型別：
- struct：定義公共的物件，類似於 C 語言中的結構體定義，在 Java 中是一個 JavaBean
- 容器型別：
- list：對應 Java 的 ArrayList
- set：對應 Java 的 HashSet
- map：對應 Java 的 HashMap
- 異常型別：
- exception：對應 Java 的 Exception
- 服務型別：
- service：對應服務的類
資料傳輸協議
不同版本稍有區別
- TBinaryProtocol : 二進位制格式.
- TCompactProtocol : 壓縮格式
- TJSONProtocol : JSON格式
- TSimpleJSONProtocol : 提供JSON只寫協議, 生成的檔案很容易通過指令碼語言解析
| 資料傳輸格式 | 型別 | 優點 | 缺點 |
|---|---|---|---|
| Xml | 文字 | 1、良好的可讀性。 2、序列化的資料包含完整的結構。3、調整不同屬性的順序對序列化/反序列化不影響 | 1、資料傳輸量大。2、不支援二進位制資料型別 |
| Json | 文字 | 1、良好的可讀性。2、調整不同屬性的順序對序列化/反序列化不影響 | 1、丟棄了型別資訊, 比如”price”:100, 對price型別是int/double解析有二義性。2、不支援二進位制資料型別 |
| Thrift | 二進位制 | 高效 | 1、不宜讀。2、向後相容有一定的約定限制，採用id遞增的方式標識並以optional修飾來新增 |
| Google Protobuf | 二進位制 | 高效 | 1、不宜讀。2、向後相容有一定的約定限制 |
服務模型
Thrif 提供網路模型：單執行緒、多執行緒、事件驅動。從另一個角度劃分為：阻塞服務模型、非阻塞服務模型。
- 阻塞服務
- TSimpleServer
- TThreadPoolServer
- 非阻塞服務模型
- TNonblockingServer
- THsHaServer
- TThreadedSelectorServer
TSimpleServer
TSimpleServer實現是非常的簡單，迴圈監聽新請求的到來並完成對請求的處理，是個單執行緒阻塞模型。由於是一次只能接收和處理一個socket連線，效率比較低，在實際開發過程中很少用到它。
TThreadPoolServer
ThreadPoolServer為解決了TSimpleServer不支援併發和多連線的問題, 引入了執行緒池。但仍然是多執行緒阻塞模式即實現的模型是One Thread Per Connection。
執行緒池採用能執行緒數可伸縮的模式，執行緒池中的佇列採用同步佇列(SynchronousQueue)。
ThreadPoolServer拆分了監聽執行緒(accept)和處理客戶端連線的工作執行緒(worker), 監聽執行緒每接到一個客戶端, 就投給執行緒池去處理。
TThreadPoolServer模式優點：
執行緒池模式中，資料讀取和業務處理都交由執行緒池完成，主執行緒只負責監聽新連線，因此在併發量較大時新連線也能夠被及時接受。執行緒池模式比較適合伺服器端能預知最多有多少個客戶端併發的情況，這時每個請求都能被業務執行緒池及時處理，效能也非常高。
TThreadPoolServer模式缺點：
執行緒池模式的處理能力受限於執行緒池的工作能力，當併發請求數大於執行緒池中的執行緒數時，新請求也只能排隊等待。
TNonblockingServer
TNonblockingServer採用單執行緒非阻塞(NIO)的模式, 藉助Channel/Selector機制, 採用IO事件模型來處理。所有的socket都被註冊到selector中，在一個執行緒中通過seletor迴圈監控所有的socket，每次selector結束時，處理所有的處於就緒狀態的socket，對於有資料到來的socket進行資料讀取操作，對於有資料傳送的socket則進行資料傳送，對於監聽socket則產生一個新業務socket並將其註冊到selector中。
select程式碼裡對accept/read/write等IO事件進行監控和處理, 唯一可惜的這個單執行緒處理. 當遇到handler裡有阻塞的操作時, 會導致整個服務被阻塞住。
TNonblockingServer模式優點：
相比於TSimpleServer效率提升主要體現在IO多路複用上，TNonblockingServer採用非阻塞IO，同時監控多個socket的狀態變化；
TNonblockingServer模式缺點：
TNonblockingServer模式在業務處理上還是採用單執行緒順序來完成，在業務處理比較複雜、耗時的時候，例如某些介面函式需要讀取資料庫執行時間較長，此時該模式效率也不高，因為多個呼叫請求任務依然是順序一個接一個執行。
THsHaServer
THsHaServer類是TNonblockingServer類的子類，為解決TNonblockingServer的缺點, THsHa引入了執行緒池去處理, 其模型把讀寫任務放到執行緒池去處理即多執行緒非阻塞模式。HsHa是: Half-sync/Half-async的處理模式, Half-aysnc是在處理IO事件上(accept/read/write io), Half-sync用於handler對rpc的同步處理上。因此可以認為THsHaServer半同步半非同步。
THsHaServer的優點：
與TNonblockingServer模式相比，THsHaServer在完成資料讀取之後，將業務處理過程交由一個執行緒池來完成，主執行緒直接返回進行下一次迴圈操作，效率大大提升；
THsHaServer的缺點：
主執行緒需要完成對所有socket的監聽以及資料讀寫的工作，當併發請求數較大時，且傳送資料量較多時，監聽socket上新連線請求不能被及時接受。
TThreadedSelectorServer
TThreadedSelectorServer是大家廣泛採用的服務模型，其多執行緒伺服器端使用非堵塞式I/O模型，是對TNonblockingServer的擴充, 其分離了Accept和Read/Write的Selector執行緒, 同時引入Worker工作執行緒池。
（1）一個AcceptThread執行緒物件，專門用於處理監聽socket上的新連線；
（2）若干個SelectorThread物件專門用於處理業務socket的網路I/O操作，所有網路資料的讀寫均是有這些執行緒來完成；
（3）一個負載均衡器SelectorThreadLoadBalancer物件，主要用於AcceptThread執行緒接收到一個新socket連線請求時，決定將這個新連線請求分配給哪個SelectorThread執行緒。
（4）一個ExecutorService型別的工作執行緒池，在SelectorThread執行緒中，監聽到有業務socket中有呼叫請求過來，則將請求讀取之後，交個ExecutorService執行緒池中的執行緒完成此次呼叫的具體執行
MainReactor就是Accept執行緒, 用於監聽客戶端連線, SubReactor採用IO事件執行緒(多個), 主要負責對所有客戶端的IO讀寫事件進行處理. 而Worker工作執行緒主要用於處理每個rpc請求的handler回撥處理(這部分是同步的)。因此其也是Half-Sync/Half-Async（半非同步-半同步）的 。
TThreadedSelectorServer模式對於大部分應用場景效能都不會差，因為其有一個專門的執行緒AcceptThread用於處理新連線請求，因此能夠及時響應大量併發連線請求；另外它將網路I/O操作分散到多個SelectorThread執行緒中來完成，因此能夠快速對網路I/O進行讀寫操作，能夠很好地應對網路I/O較多的情況。