04. Apache thrift 之傳輸協(xié)議

我們都知道，數(shù)據(jù)在網(wǎng)絡(luò)上是以二進制方式傳輸?shù)摹?/p>

對于一個java對象，從客戶端通過網(wǎng)絡(luò)傳輸?shù)椒?wù)端時，客戶端需要將其轉(zhuǎn)換為二進制，然后寫入網(wǎng)絡(luò)IO；服務(wù)端從網(wǎng)絡(luò)IO接收到數(shù)據(jù)時，也需要將二進制數(shù)據(jù)轉(zhuǎn)換為對象，然后再進行操作。

這兩個轉(zhuǎn)換過程，統(tǒng)稱為編解碼操作。

在Thrift中，編解碼操作又叫傳輸協(xié)議，由抽象類 TProtocol 來定義，它的方法主要分為兩大類：

• TProtocol寫入方法：

• TProtocol讀取方法：

寫入方法規(guī)定了數(shù)據(jù)轉(zhuǎn)換成二進制的方式，讀取方法規(guī)定了將二進制轉(zhuǎn)換為數(shù)據(jù)的方式。

thrift中常用的傳輸協(xié)議有以下幾種：

• TBinaryProtocol：二進制編碼格式進行數(shù)據(jù)傳輸
• TCompactProtocol：高效率的、密集的二進制編碼格式進行數(shù)據(jù)傳輸
• TJSONProtocol：使用JSON文本的數(shù)據(jù)編碼協(xié)議進行數(shù)據(jù)傳輸
• TSimpleJSONProtocol：只提供JSON只寫的協(xié)議，適用于通過腳本語言解析
• TMultiplexedProtocol：復(fù)合協(xié)議，同時處理多個服務(wù)

這些傳輸協(xié)議都是TProtocol的子類，接下來我們來分析下這些方法的操作。

TBinaryProtocol

TBinaryProtocol 是最基本的實現(xiàn)，得到的二進制數(shù)據(jù)是原始數(shù)據(jù)（相比于壓縮數(shù)據(jù)而言），這里我們選取幾個方法了解轉(zhuǎn)換過程。

讀寫 I32 類型數(shù)據(jù)

所謂的I32數(shù)據(jù)，是指32位的整型數(shù)據(jù)，也就是java中的int類型，寫入方法如下：

public void writeI32(int i32) throws TException {
  // 將i32類型的數(shù)據(jù)轉(zhuǎn)換為byte數(shù)組
  inoutTemp[0] = (byte)(0xff & (i32 >> 24));
  inoutTemp[1] = (byte)(0xff & (i32 >> 16));
  inoutTemp[2] = (byte)(0xff & (i32 >> 8));
  inoutTemp[3] = (byte)(0xff & (i32));
  trans_.write(inoutTemp, 0, 4);
}

代碼中主要是通過位運算，將int類型轉(zhuǎn)換為長度為4的byte數(shù)組。

int類型的讀取操作如下：

  @Override
  public int readI32() throws TException {
    byte[] buf = inoutTemp;
    int off = 0;
    // 讀取4個byte
    if (trans_.getBytesRemainingInBuffer() >= 4) {
      buf = trans_.getBuffer();
      off = trans_.getBufferPosition();
      trans_.consumeBuffer(4);
    } else {
      readAll(inoutTemp, 0, 4);
    }
    // 通過位運算轉(zhuǎn)換為int類型
    return
      ((buf[off] & 0xff) << 24) |
      ((buf[off+1] & 0xff) << 16) |
      ((buf[off+2] & 0xff) <<  8) |
      ((buf[off+3] & 0xff));
  }

讀取操作比較簡單，先是在數(shù)據(jù)流上讀取4個byte，然后通過位運算將這4個byte轉(zhuǎn)換為int數(shù)據(jù)。

讀寫 I64 數(shù)據(jù)

基本數(shù)據(jù)類型都是這么操作的，比如long類型：

  @Override
  public void writeI64(long i64) throws TException {
    // 將i64類型的數(shù)據(jù)轉(zhuǎn)換為byte數(shù)組
    inoutTemp[0] = (byte)(0xff & (i64 >> 56));
    inoutTemp[1] = (byte)(0xff & (i64 >> 48));
    inoutTemp[2] = (byte)(0xff & (i64 >> 40));
    inoutTemp[3] = (byte)(0xff & (i64 >> 32));
    inoutTemp[4] = (byte)(0xff & (i64 >> 24));
    inoutTemp[5] = (byte)(0xff & (i64 >> 16));
    inoutTemp[6] = (byte)(0xff & (i64 >> 8));
    inoutTemp[7] = (byte)(0xff & (i64));
    trans_.write(inoutTemp, 0, 8);
  }

long類型占8個byte，因此轉(zhuǎn)換后的byte數(shù)組長度為8.

寫入方法如下：

  @Override
  public long readI64() throws TException {
    byte[] buf = inoutTemp;
    int off = 0;
    // 讀取8個byte
    if (trans_.getBytesRemainingInBuffer() >= 8) {
      buf = trans_.getBuffer();
      off = trans_.getBufferPosition();
      trans_.consumeBuffer(8);
    } else {
      readAll(inoutTemp, 0, 8);
    }
    // 將byte數(shù)組轉(zhuǎn)換為log
    return
      ((long)(buf[off]   & 0xff) << 56) |
      ((long)(buf[off+1] & 0xff) << 48) |
      ((long)(buf[off+2] & 0xff) << 40) |
      ((long)(buf[off+3] & 0xff) << 32) |
      ((long)(buf[off+4] & 0xff) << 24) |
      ((long)(buf[off+5] & 0xff) << 16) |
      ((long)(buf[off+6] & 0xff) <<  8) |
      ((long)(buf[off+7] & 0xff));
  }

long數(shù)據(jù)的讀取操作與int類型非常相似，先是在數(shù)據(jù)流上讀取8個byte，然后通過位運算將這8個byte轉(zhuǎn)換為long數(shù)據(jù)。

像byte、char、short 等整形數(shù)據(jù)的操作方式類似：先是在數(shù)據(jù)流上讀取固定長度的byte，然后通過位運算將這些byte轉(zhuǎn)換為指定類型的數(shù)據(jù)，這里就不多說了。

讀寫 boolean 類型數(shù)據(jù)

thrift 在處理boolean類型數(shù)據(jù)時，是將其按照byte類型來處理：

  public void writeBool(boolean b) throws TException {
    writeByte(b ? (byte)1 : (byte)0);
  }

在寫入boolean類型數(shù)據(jù)時，直接按byte類型來處理，0為false，1為true.

讀取操作也是按byte類型來處理，這里就不多說了。

讀寫String 類型數(shù)據(jù)

前面介紹的數(shù)據(jù)類型都是固定長度的，如int類型占4個byte，long類型占8個byte，byte類型占1個byte，對于String這種可變長度的數(shù)據(jù)，該如何寫入與讀取呢？

String 類型的數(shù)據(jù)寫入方式如下：

public void writeString(String str) throws TException {
    byte[] dat = str.getBytes(StandardCharsets.UTF_8);
    // 寫入長度數(shù)據(jù)
    writeI32(dat.length);
    trans_.write(dat, 0, dat.length);
}

寫入操作包含兩部分：

• 寫入長度
• 寫入二進制數(shù)據(jù)

與數(shù)據(jù)類型不同，String類型的數(shù)據(jù)長度并不固定，因此在寫入時，需要先指定數(shù)據(jù)長度，這樣才能在讀取時，知道讀取多少個byte，它的讀取方法如下：

  @Override
  public String readString() throws TException {
    // 先讀取 String 的長度
    int size = readI32();

    // 讀取固定長度的byte，然后轉(zhuǎn)換為String
    if (trans_.getBytesRemainingInBuffer() >= size) {
      String s = new String(trans_.getBuffer(), trans_.getBufferPosition(),
          size, StandardCharsets.UTF_8);
      trans_.consumeBuffer(size);
      return s;
    }
    return readStringBody(size);
  }

讀寫復(fù)雜對象數(shù)據(jù)

以上分析了基本類型的寫入與讀取，對于復(fù)雜對象，該如何讀寫呢？

需要說明的，復(fù)雜對象也是由基本對象構(gòu)成的，比如QryResult：

public class QryResult implements ... {

  public int code;

  public String msg;

  ...
}

它的寫入方法為QryResult.QryResultStandardScheme#write：

public void write(org.apache.thrift.protocol.TProtocol oprot, QryResult struct) 
    throws org.apache.thrift.TException {
  struct.validate();
  // 寫入對象開始標(biāo)識
  oprot.writeStructBegin(STRUCT_DESC);
  // 寫入屬性開始標(biāo)識
  oprot.writeFieldBegin(CODE_FIELD_DESC);
  oprot.writeI32(struct.code);
  // 寫入結(jié)束開始標(biāo)識
  oprot.writeFieldEnd();
  if (struct.msg != null) {
    // 寫入屬性開始標(biāo)識
    oprot.writeFieldBegin(MSG_FIELD_DESC);
    oprot.writeString(struct.msg);
    // 寫入結(jié)束開始標(biāo)識
    oprot.writeFieldEnd();
  }
  oprot.writeFieldStop();
  // 寫入對象結(jié)束標(biāo)識
  oprot.writeStructEnd();
}

從寫入操作來看，

• 對象寫入前后，都會寫入一個標(biāo)識，表明對象的開始與結(jié)束位置
• 屬性寫入前后，都會寫入一個標(biāo)識，表明屬性的開始與結(jié)束位置，屬性會有多個
• 屬性就是基本類型，如byte、int、String等

寫入的屬性開始標(biāo)記CODE_FIELD_DESC與MSG_FIELD_DESC如下：

  // code
  private static final TField CODE_FIELD_DESC = new TField("code", I32, (short)1);
  // msg
  private static final TField MSG_FIELD_DESC = new TField("msg", TType.STRING, (short)2);

再來看看QryResult對象的讀取操作，進入QryResult.QryResultStandardScheme#read方法：

public void read(org.apache.thrift.protocol.TProtocol iprot, QryResult struct) 
    throws org.apache.thrift.TException {
  org.apache.thrift.protocol.TField schemeField;
  // 讀取對象開始標(biāo)記
  iprot.readStructBegin();
  while (true)
  {
    // 讀取屬性開始標(biāo)記，標(biāo)記會包含屬性類型
    schemeField = iprot.readFieldBegin();
    if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
      break;
    }
    switch (schemeField.id) {
      case 1: // CODE
        // 判斷類型
        if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
          // 讀取 code，按 i32（也就是int）類型讀取
          struct.code = iprot.readI32();
          struct.setCodeIsSet(true);
        } else { 
          org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
        }
        break;
      case 2: // MSG
        // 判斷類型
        if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
          // 讀取 msg， 按 String 類型讀取
          struct.msg = iprot.readString();
          struct.setMsgIsSet(true);
        } else { 
          org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
        }
        break;
      default:
        org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
    }
    // 讀取屬性結(jié)束標(biāo)記
    iprot.readFieldEnd();
  }
  // 讀取對象結(jié)束標(biāo)記
  iprot.readStructEnd();

  struct.validate();
}

讀取操作如下：

1. 讀取對象開始標(biāo)記
2. 循環(huán)讀取屬性：
1. 讀取屬性開始標(biāo)記
2. 讀取屬性值，由標(biāo)識里包含里了屬性值類型，讀取屬性值時只需要調(diào)用對應(yīng)方法讀取即可
3. 讀取屬性結(jié)束標(biāo)記
3. 讀取對象結(jié)束標(biāo)記

以上就是復(fù)雜對象的讀取方式了。

TCompactProtocol

接著我們來看看TCompactProtocol，它的注釋如下：

TCompactProtocol2 is the Java implementation of the compact protocol specified in THRIFT-110. The fundamental approach to reducing the overhead of structures is a) use variable-length integers all over the place and b) make use of unused bits wherever possible. Your savings will obviously vary based on the specific makeup of your structs, but in general, the more fields, nested structures, short strings and collections, and low-value i32 and i64 fields you have, the more benefit you'll see.

TCompactProtocol2是THRIFT-110中指定的緊湊協(xié)議的Java實現(xiàn)。減少結(jié)構(gòu)開銷的基本方法是：a）在整個位置使用長度可變的整數(shù)， b）盡可能使用未使用的位。根據(jù)結(jié)構(gòu)的具體組成，您的節(jié)省顯然會有所不同，但通常，您擁有的字段，嵌套結(jié)構(gòu)，短字符串和集合以及低價值的i32和i64字段越多，您將看到更多的好處。

從注釋來看，它是用來節(jié)省數(shù)據(jù)空間的，使用的是緊湊協(xié)議，我們來看看它是如何做到的。

讀寫 i32類型數(shù)據(jù)

我們進入writeI32方法：

public void writeI32(int i32) throws TException {
  writeVarint32(intToZigZag(i32));
}

/**
 * 轉(zhuǎn)換
 */
private int intToZigZag(int n) {
  return (n << 1) ^ (n >> 31);
}

/**
 * 寫入變長操作
 */
private void writeVarint32(int n) throws TException {
  int idx = 0;
  while (true) {
    if ((n & ~0x7F) == 0) {
      temp[idx++] = (byte)n;
      // writeByteDirect((byte)n);
      break;
      // return;
    } else {
      temp[idx++] = (byte)((n & 0x7F) | 0x80);
      // writeByteDirect((byte)((n & 0x7F) | 0x80));
      n >>>= 7;
    }
  }
  trans_.write(temp, 0, idx);
}

從代碼來看，寫入方法有兩個操作：

• 使用intToZigZag將int類型數(shù)據(jù)轉(zhuǎn)換為ZigZag
• 使用writeVarint32將得到的ZigZag寫入

關(guān)于ZigZag是個啥，我在網(wǎng)上找了一篇介紹文章：小而巧的數(shù)字壓縮算法：zigzag（(https://blog.csdn.net/zgwangbo/article/details/51590186)[https://blog.csdn.net/zgwangbo/article/details/51590186]）

以int類型為例，它的核心思想如下：對于int類型的整數(shù)1（二進制為00000000_00000000_00000000_00000001），如果直接按int傳輸，前3個byte都是0，有效值僅在第4個byte上，這種情況下對于前3個byte的傳輸就是一種浪費。針對這種小整數(shù)的情況，ZigZag的處理是，通過只傳輸有效的部分，降低傳輸?shù)淖止?jié)數(shù)，即最終只需要傳輸00000001，這就可以少傳輸3個byte。

根據(jù)ZigZag的思想，ZigZag僅對大于1個byte的整型數(shù)據(jù)類型有效，比如char、short、int、long等。

讀寫double類型數(shù)據(jù)

ZigZag可處理大于1個byte的整型數(shù)據(jù)類型，那能不能處理double類型數(shù)據(jù)呢？

我們來看看double類型的操作：

public void writeDouble(double dub) throws TException {
  fixedLongToBytes(Double.doubleToLongBits(dub), temp, 0);
  trans_.write(temp, 0, 8);
}

從代碼來看，依然是寫入了8個字節(jié)，因此并不能對double進行壓縮操作。

讀寫String類型

再看看String類型：

public void writeString(String str) throws TException {
  byte[] bytes = str.getBytes(StandardCharsets.UTF_8);
  // 長度還是可以使用的
  writeVarint32(bytes.length);
  trans_.write(bytes, 0, bytes.length);
}

在writeString方法中，寫入了兩部分?jǐn)?shù)據(jù)：

• String數(shù)據(jù)轉(zhuǎn)化為二進制后的長度：int類型，可以使用僅是對長度使用了ZigZag壓縮
• String具體數(shù)據(jù)：不是整數(shù)類型，不能使用ZigZag壓縮

可以看到，由于長度是int類型，因此可以使用僅是對長度使用了ZigZag算法，數(shù)據(jù)部分就無能為力.

從以上來分析來看，ZigZag僅是對整形數(shù)據(jù)類型（char、short、int、long）的壓縮，除此之外，對其他類型的數(shù)據(jù)就無能為力了。

TJSONProtocol 與 TSimpleJSONProtocol

關(guān)于json的操作，想必各位小伙伴已經(jīng)很熟悉了，這里我們簡單地了解下thrfit的json序列化協(xié)議。

TJSONProtocol 的注釋如下：

JSON protocol implementation for thrift. This is a full-featured protocol supporting write and read. Please see the C++ class header for a detailed description of the protocol's wire format.

thrift 的 JSON 協(xié)議實現(xiàn)。這是一個支持讀寫的全功能協(xié)議。請參閱 C++ 類頭文件以獲取協(xié)議的有線格式的詳細(xì)說明。

注意到注釋中的「這是一個支持讀寫的全功能協(xié)議」，這個是相對于TSimpleJSONProtocol而言的，它的注釋如下：

JSON protocol implementation for thrift. This protocol is write-only and produces a simple output format suitable for parsing by scripting languages. It should not be confused with the full-featured TJSONProtocol. JSON協(xié)議實現(xiàn)節(jié)儉。該協(xié)議是只寫的，并且會生成一種簡單的輸出格式，適合通過腳本語言進行解析。不應(yīng)將其與功能齊全的TJSONProtocol混淆。

thrift 中提供了兩個json序列化協(xié)議：

• TJSONProtocol：支持「讀寫的全功能協(xié)議」
• TSimpleJSONProtocol：「只寫協(xié)議」

在thrift中傳輸多是二進制協(xié)議，json協(xié)議用的比較少，關(guān)于json的讀寫操作，本文就不過多分析了，

TMultiplexedProtocol

TMultiplexedProtocol并不是一個傳輸協(xié)議，而是協(xié)議的包裝器，它允許Thrift客戶端在函數(shù)調(diào)用期間通過在服務(wù)名稱之前添加服務(wù)名稱來與Thrift服務(wù)器進行通信。

如果使用了客戶端使用了TMultiplexedProtocol作為傳輸協(xié)議，在服務(wù)端，需要使用TMultiplexedProcessor處理來自多路復(fù)用客戶端的請求。

TMultiplexedProtocol示例

下面我們使用TMultiplexedProtocol實現(xiàn)單個套接字傳輸來調(diào)用兩個服務(wù)的功能：

客戶端：

TTransport transport = new TSocket("localhost", SERVER_PORT);
transport.open();
TProtocol protocol = new TBinaryProtocol(transport);

// 指定serviceName與處理的service
// helloService
TMultiplexedProtocol helloService = new TMultiplexedProtocol(
        protocol, "helloService");
HelloService.Client client = new HelloService.Client(helloService);
System.out.println(client.hello("thrift world"));

// queryService
TMultiplexedProtocol helloProtocol = new TMultiplexedProtocol(
        protocol, "queryService");
QueryService.Client queryClient = new QueryService.Client(helloProtocol);
System.out.println(queryClient.query(1));

服務(wù)端：

// 構(gòu)建 processor，分別指定serivceNam與對應(yīng)的Processor
TMultiplexedProcessor processor = new TMultiplexedProcessor();
processor.registerProcessor("helloService",
        new HelloService.Processor<>(new HelloServiceImpl()));
processor.registerProcessor("queryService",
        new QueryService.Processor<>(new QueryServiceImpl()));
// 生成 TServer 實例
TServer server = new TSimpleServer(
  new TServer.Args(new TServerSocket(port)).processor(processor));
System.out.println("Starting the simple server...");
server.serve();

TMultiplexedProtocol的實現(xiàn)原理

TMultiplexedProtocol是如何做到區(qū)分服務(wù)的呢？答應(yīng)就在serviceName！

我們來看看TMultiplexedProtocol構(gòu)造方法：

public TMultiplexedProtocol(TProtocol protocol, String serviceName) {
    super(protocol);
    SERVICE_NAME = serviceName;
}

在TMultiplexedProtocol的構(gòu)造方法中，需要指定服務(wù)名(serviceName)。

再來看看數(shù)據(jù)寫入操作：

@Override
public void writeMessageBegin(TMessage tMessage) throws TException {
  if (tMessage.type == TMessageType.CALL || tMessage.type == TMessageType.ONEWAY) {
      super.writeMessageBegin(new TMessage(
              // 寫數(shù)據(jù)時，需要指定服務(wù)名
              SERVICE_NAME + SEPARATOR + tMessage.name,
              tMessage.type,
              tMessage.seqid
      ));
  } else {
      super.writeMessageBegin(tMessage);
  }
}

從代碼來看，與其他Protocol不同的是，TMultiplexedProtocol在寫入數(shù)據(jù)時，需要指定服務(wù)名。

thrift客戶端在請求服務(wù)端時，會帶上當(dāng)前請求的「服務(wù)名」，表明要調(diào)用的是哪個服務(wù)的方法；當(dāng)服務(wù)端收到數(shù)據(jù)后，根據(jù)這個「服務(wù)名」來獲取對應(yīng)的服務(wù)，再調(diào)用該服務(wù)的方法，從而實現(xiàn)「多服務(wù)復(fù)用」功能。

總結(jié)

數(shù)據(jù)在網(wǎng)絡(luò)傳輸中是以字節(jié)流的形式傳輸?shù)?，即使用的是二進制，所謂的傳輸協(xié)議本質(zhì)上就是二進制與其他數(shù)據(jù)類型的轉(zhuǎn)換操作。

本文介紹了thrift提供的幾種傳輸協(xié)議的實現(xiàn)，其中重點介紹了二進制傳輸協(xié)議TBinaryProtocol的功能，具體介紹了int、long、String、boolean以及對象類型與二進制相互轉(zhuǎn)化的操作；

接著介紹了TCompactProtocol的功能，它可以對整數(shù)類型的數(shù)據(jù)進行壓縮，從而減少數(shù)據(jù)的傳輸，提高傳輸效率；

對于thrfit提供的兩個json序列化協(xié)議，本文并沒有深究，個人覺得了解即可；

最后介紹了TMultiplexedProtocol協(xié)議，這是個包裝協(xié)議，并不實現(xiàn)具體的讀寫操作，它的作用是，如果一個服務(wù)器上有多個thrift服務(wù)，可以通過它來指定serviceName從而確定調(diào)用的是哪個服務(wù)。

限于作者個人水平，文中難免有錯誤之處，歡迎指正！原創(chuàng)不易，商業(yè)轉(zhuǎn)載請聯(lián)系作者獲得授權(quán)，非商業(yè)轉(zhuǎn)載請注明出處。

本文首發(fā)于微信公眾號 「Java技術(shù)探秘」，如果您喜歡本文，歡迎關(guān)注該公眾號，讓我們一起在技術(shù)的世界里探秘吧！