码迷，mamicode.com
首页 > 其他好文 > 详细

Thrift源码分析（二）-- 协议和编解码

时间：2018-06-23 21:00:56      阅读：130      评论：0      收藏：0      [点我收藏+]

标签：程序   switch   hand   throws   while   框架   UI   mask   rds   

协议和编解码是一个网络应用程序的核心问题之一，客户端和服务器通过约定的协议来传输消息(数据)，通过特定的格式来编解码字节流，并转化成业务消息，提供给上层框架调用。

Thrift的协议比较简单，它把协议和编解码整合在了一起。抽象类TProtocol定义了协议和编解码的顶层接口。个人感觉采用抽象类而不是接口的方式来定义顶层接口并不好，TProtocol关联了一个TTransport传输对象，而不是提供一个类似getTransport()的接口，导致抽象类的扩展性比接口差。

TProtocol主要做了两个事情:

1. 关联TTransport对象

2.定义一系列读写消息的编解码接口，包括两类，一类是复杂数据结构比如readMessageBegin, readMessageEnd,  writeMessageBegin, writMessageEnd.还有一类是基本数据结构，比如readI32, writeI32, readString, writeString

 

[java] view plain copy
 
  1. public abstract class TProtocol {  
  2.   
  3.   /** 
  4.    * Transport 
  5.    */  
  6.   protected TTransport trans_;  
  7.   
  8. 　public abstract void writeMessageBegin(TMessage message) throws TException;  
  9.   
  10.   public abstract void writeMessageEnd() throws TException;  
  11.   
  12.   public abstract void writeStructBegin(TStruct struct) throws TException;  
  13.   
  14.   public abstract void writeStructEnd() throws TException;  
  15.   
  16.   public abstract void writeFieldBegin(TField field) throws TException;  
  17.   
  18.   public abstract void writeFieldEnd() throws TException;  
  19.   
  20.   public abstract void writeFieldStop() throws TException;  
  21.   
  22.   public abstract void writeMapBegin(TMap map) throws TException;  
  23.   
  24.   public abstract void writeMapEnd() throws TException;  
  25.   
  26.   public abstract void writeListBegin(TList list) throws TException;  
  27.   
  28.   public abstract void writeListEnd() throws TException;  
  29.   
  30.   public abstract void writeSetBegin(TSet set) throws TException;  
  31.   
  32.   public abstract void writeSetEnd() throws TException;  
  33.   
  34.   public abstract void writeBool(boolean b) throws TException;  
  35.   
  36.   public abstract void writeByte(byte b) throws TException;  
  37.   
  38.   public abstract void writeI16(short i16) throws TException;  
  39.   
  40.   public abstract void writeI32(int i32) throws TException;  
  41.   
  42.   public abstract void writeI64(long i64) throws TException;  
  43.   
  44.   public abstract void writeDouble(double dub) throws TException;  
  45.   
  46.   public abstract void writeString(String str) throws TException;  
  47.   
  48.   public abstract void writeBinary(ByteBuffer buf) throws TException;  
  49.   
  50.   /** 
  51.    * Reading methods. 
  52.    */  
  53.   
  54.   public abstract TMessage readMessageBegin() throws TException;  
  55.   
  56.   public abstract void readMessageEnd() throws TException;  
  57.   
  58.   public abstract TStruct readStructBegin() throws TException;  
  59.   
  60.   public abstract void readStructEnd() throws TException;  
  61.   
  62.   public abstract TField readFieldBegin() throws TException;  
  63.   
  64.   public abstract void readFieldEnd() throws TException;  
  65.   
  66.   public abstract TMap readMapBegin() throws TException;  
  67.   
  68.   public abstract void readMapEnd() throws TException;  
  69.   
  70.   public abstract TList readListBegin() throws TException;  
  71.   
  72.   public abstract void readListEnd() throws TException;  
  73.   
  74.   public abstract TSet readSetBegin() throws TException;  
  75.   
  76.   public abstract void readSetEnd() throws TException;  
  77.   
  78.   public abstract boolean readBool() throws TException;  
  79.   
  80.   public abstract byte readByte() throws TException;  
  81.   
  82.   public abstract short readI16() throws TException;  
  83.   
  84.   public abstract int readI32() throws TException;  
  85.   
  86.   public abstract long readI64() throws TException;  
  87.   
  88.   public abstract double readDouble() throws TException;  
  89.   
  90.   public abstract String readString() throws TException;  
  91.   
  92.   public abstract ByteBuffer readBinary() throws TException;  
  93.   
  94.   /** 
  95.    * Reset any internal state back to a blank slate. This method only needs to 
  96.    * be implemented for stateful protocols. 
  97.    */  
  98.   public void reset() {}  
  99.     
  100.   /** 
  101.    * Scheme accessor 
  102.    */  
  103.   public Class<? extends IScheme> getScheme() {  
  104.     return StandardScheme.class;  
  105.   }  
  106.   
  107. }  


所谓协议就是客户端和服务器端约定传输什么数据，如何解析传输的数据。对于一个RPC调用的协议来说，要传输的数据主要有:

 

调用方

1. 方法的名称，包括类的名称和方法的名称

2. 方法的参数，包括类型和参数值

3.一些附加的数据，比如附件，超时事件，自定义的控制信息等等

返回方

1. 调用的返回码

2. 返回值

3.异常信息

 

从TProtocol的定义我们可以看出Thrift的协议约定如下事情:

1. 先writeMessageBegin表示开始传输消息了，写消息头。Message里面定义了方法名，调用的类型，版本号，消息seqId

2. 接下来是写方法的参数，实际就是写消息体。如果参数是一个类，就writeStructBegin

3. 接下来写字段，writeFieldBegin, 这个方法会写接下来的字段的数据类型和顺序号。这个顺序号是Thrfit对要传输的字段的一个编码，从１开始

4. 如果是一个集合就writeListBegin/writeMapBegin，如果是一个基本数据类型，比如int, 就直接writeI32

5. 每个复杂数据类型写完都调用writeXXXEnd，直到writeMessageEnd结束

6. 读消息时根据数据类型读取相应的长度

 

每个writeXXX都是采用消息头+消息体的方式。我们来看TBinaryProtocol的实现。

1. writeMessgeBegin方法写了消息头，包括4字节的版本号和类型信息，字符串类型的方法名，４字节的序列号seqId

2. writeFieldBegin，写了１个字节的字段数据类型，和2个字节字段的顺序号

3. writeI32，写了４个字节的字节数组

4. writeString,先写４字节消息头表示字符串长度，再写字符串字节

5. writeBinary,先写４字节消息头表示字节数组长度，再写字节数组内容

6.readMessageBegin时，先读４字节版本和类型信息，再读字符串，再读４字节序列号

7.readFieldBegin，先读1个字节的字段数据类型，再读2个字节的字段顺序号

8. readString时，先读４字节字符串长度，再读字符串内容。字符串统一采用UTF-8编码

 

[java] view plain copy
 
  1.  public void writeMessageBegin(TMessage message) throws TException {  
  2.     if (strictWrite_) {  
  3.       int version = VERSION_1 | message.type;  
  4.       writeI32(version);  
  5.       writeString(message.name);  
  6.       writeI32(message.seqid);  
  7.     } else {  
  8.       writeString(message.name);  
  9.       writeByte(message.type);  
  10.       writeI32(message.seqid);  
  11.     }  
  12.   }  
  13.   
  14. public void writeFieldBegin(TField field) throws TException {  
  15.     writeByte(field.type);  
  16.     writeI16(field.id);  
  17.   }  
  18.   
  19. private byte[] i32out = new byte[4];  
  20.   public void writeI32(int i32) throws TException {  
  21.     i32out[0] = (byte)(0xff & (i32 >> 24));  
  22.     i32out[1] = (byte)(0xff & (i32 >> 16));  
  23.     i32out[2] = (byte)(0xff & (i32 >> 8));  
  24.     i32out[3] = (byte)(0xff & (i32));  
  25.     trans_.write(i32out, 0, 4);  
  26.   }  
  27.   
  28. public void writeString(String str) throws TException {  
  29.     try {  
  30.       byte[] dat = str.getBytes("UTF-8");  
  31.       writeI32(dat.length);  
  32.       trans_.write(dat, 0, dat.length);  
  33.     } catch (UnsupportedEncodingException uex) {  
  34.       throw new TException("JVM DOES NOT SUPPORT UTF-8");  
  35.     }  
  36.   }  
  37.   
  38. public void writeBinary(ByteBuffer bin) throws TException {  
  39.     int length = bin.limit() - bin.position();  
  40.     writeI32(length);  
  41.     trans_.write(bin.array(), bin.position() + bin.arrayOffset(), length);  
  42.   }  
  43.   
  44. public TMessage readMessageBegin() throws TException {  
  45.     int size = readI32();  
  46.     if (size < 0) {  
  47.       int version = size & VERSION_MASK;  
  48.       if (version != VERSION_1) {  
  49.         throw new TProtocolException(TProtocolException.BAD_VERSION, "Bad version in readMessageBegin");  
  50.       }  
  51.       return new TMessage(readString(), (byte)(size & 0x000000ff), readI32());  
  52.     } else {  
  53.       if (strictRead_) {  
  54.         throw new TProtocolException(TProtocolException.BAD_VERSION, "Missing version in readMessageBegin, old client?");  
  55.       }  
  56.       return new TMessage(readStringBody(size), readByte(), readI32());  
  57.     }  
  58.   }  
  59.   
  60. public TField readFieldBegin() throws TException {  
  61.     byte type = readByte();  
  62.     short id = type == TType.STOP ? 0 : readI16();  
  63.     return new TField("", type, id);  
  64.   }  
  65.   
  66. public String readString() throws TException {  
  67.     int size = readI32();  
  68.   
  69.     if (trans_.getBytesRemainingInBuffer() >= size) {  
  70.       try {  
  71.         String s = new String(trans_.getBuffer(), trans_.getBufferPosition(), size, "UTF-8");  
  72.         trans_.consumeBuffer(size);  
  73.         return s;  
  74.       } catch (UnsupportedEncodingException e) {  
  75.         throw new TException("JVM DOES NOT SUPPORT UTF-8");  
  76.       }  
  77.     }  
  78.   
  79.     return readStringBody(size);  
  80.   }  

 

 

TProtocol定义了基本的协议信息，包括传输什么数据，如何解析传输的数据的基本方法。

技术分享图片

还存在一个问题，就是服务器端如何知道客户端发送过来的数据是怎么组合的，比如第一个字段是字符串类型，第二个字段是int。这个信息是在IDL生成客户端时生成的代码时提供了。Thrift生成的客户端代码提供了读写参数的方法，这两个方式是一一对应的，包括字段的序号，类型等等。客户端使用写参数的方法，服务器端使用读参数的方法。

关于IDL生成的客户端代码会在后面的文章具体描述。下面简单看一下自动生成的代码

1. 方法的调用从writeMessageBegin开始，发送了消息头信息

2. 写方法的参数，也就是写消息体。方法参数由一个统一的接口TBase描述，提供了read和write的统一接口。自动生成的代码提供了read, write方法参数的具体实现

3. 写完结束  

 

 

[java] view plain copy
 
  1.  public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {  
  2.         prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("handle", org.apache.thrift.protocol.TMessageType.CALL, 0));  
  3.         handle_args args = new handle_args();  
  4.         args.setIdentity(identity);  
  5.         args.setUid(uid);  
  6.         args.setSid(sid);  
  7.         args.setType(type);  
  8.         args.setMessage(message);  
  9.         args.setParams(params);  
  10.         args.write(prot);  
  11.         prot.writeMessageEnd();  
  12.       }  
  13.   
  14. public interface TBase<T extends TBase<?,?>, F extends TFieldIdEnum> extends Comparable<T>,  Serializable {  
  15.   
  16.   public void read(TProtocol iprot) throws TException;  
  17.   
  18.   public void write(TProtocol oprot) throws TException;  
  19. ｝  
  20.   
  21. public static class handle_args <strong>implements org.apache.thrift.TBase</strong><handle_args, handle_args._Fields>, java.io.Serializable, Cloneable   {  
  22.     private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("handle_args");  
  23.   
  24.     private static final org.apache.thrift.protocol.TField IDENTITY_FIELD_DESC = new org.apache.thrift.protocol.TField("identity", org.apache.thrift.protocol.TType.STRING, (short)1);  
  25.     private static final org.apache.thrift.protocol.TField UID_FIELD_DESC = new org.apache.thrift.protocol.TField("uid", org.apache.thrift.protocol.TType.I64, (short)2);  
  26.     private static final org.apache.thrift.protocol.TField SID_FIELD_DESC = new org.apache.thrift.protocol.TField("sid", org.apache.thrift.protocol.TType.STRING, (short)3);  
  27.     private static final org.apache.thrift.protocol.TField TYPE_FIELD_DESC = new org.apache.thrift.protocol.TField("type", org.apache.thrift.protocol.TType.I32, (short)4);  
  28.     private static final org.apache.thrift.protocol.TField MESSAGE_FIELD_DESC = new org.apache.thrift.protocol.TField("message", org.apache.thrift.protocol.TType.STRING, (short)5);  
  29.     private static final org.apache.thrift.protocol.TField PARAMS_FIELD_DESC = new org.apache.thrift.protocol.TField("params", org.apache.thrift.protocol.TType.MAP, (short)6);  
  30.   
  31.     private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();  
  32.     static {  
  33.       schemes.put(StandardScheme.class, new handle_argsStandardSchemeFactory());  
  34.       schemes.put(TupleScheme.class, new handle_argsTupleSchemeFactory());  
  35.     }  
  36.   
  37.     public String identity; // required  
  38.     public long uid; // required  
  39.     public String sid; // required  
  40.     public int type; // required  
  41.     public String message; // required  
  42.     public Map<String,String> params; // required  
  43.   
  44.     /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */  
  45.     public enum _Fields implements org.apache.thrift.TFieldIdEnum {  
  46.       IDENTITY((short)1, "identity"),  
  47.       UID((short)2, "uid"),  
  48.       SID((short)3, "sid"),  
  49.       TYPE((short)4, "type"),  
  50.       MESSAGE((short)5, "message"),  
  51.       PARAMS((short)6, "params");  
  52. ｝  
  53.   
  54. //　　自动生成的写方法参数的方法，按照字段顺序写，给客户端代码使用  
  55.       public void write(org.apache.thrift.protocol.TProtocol oprot, handle_args struct) throws org.apache.thrift.TException {  
  56.         struct.validate();  
  57.   
  58.         oprot.writeStructBegin(STRUCT_DESC);  
  59.         if (struct.identity != null) {  
  60.           oprot.writeFieldBegin(IDENTITY_FIELD_DESC);  
  61.           oprot.writeString(struct.identity);  
  62.           oprot.writeFieldEnd();  
  63.         }  
  64.         oprot.writeFieldBegin(UID_FIELD_DESC);  
  65.         oprot.writeI64(struct.uid);  
  66.         oprot.writeFieldEnd();  
  67.         if (struct.sid != null) {  
  68.           oprot.writeFieldBegin(SID_FIELD_DESC);  
  69.           oprot.writeString(struct.sid);  
  70.           oprot.writeFieldEnd();  
  71.         }  
  72.         oprot.writeFieldBegin(TYPE_FIELD_DESC);  
  73.         oprot.writeI32(struct.type);  
  74.         oprot.writeFieldEnd();  
  75.         if (struct.message != null) {  
  76.           oprot.writeFieldBegin(MESSAGE_FIELD_DESC);  
  77.           oprot.writeString(struct.message);  
  78.           oprot.writeFieldEnd();  
  79.         }  
  80. ｝  
  81.   
  82. <pre name="code" class="java">//　　自动生成的读方法参数的方法，按照字段顺序读，给服务器端代码使用  


 public void read(org.apache.thrift.protocol.TProtocol iprot, handle_args struct) throws org.apache.thrift.TException {
        org.apache.thrift.protocol.TField schemeField;
        iprot.readStructBegin();
        while (true)
        {
          schemeField = iprot.readFieldBegin();
          if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
            break;
          }
          switch (schemeField.id) {
            case 1: // IDENTITY
              if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
                struct.identity = iprot.readString();
                struct.setIdentityIsSet(true);
              } else { 
                org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
              }
              break;
            case 2: // UID
              if (schemeField.type == org.apache.thrift.protocol.TType.I64) {
                struct.uid = iprot.readI64();
                struct.setUidIsSet(true);
              } else { 
                org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
              }
              break;
            case 3: // SID
              if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
                struct.sid = iprot.readString();
                struct.setSidIsSet(true);
              } else { 
                org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
              }
              break;
            case 4: // TYPE
              if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
                struct.type = iprot.readI32();
                struct.setTypeIsSet(true);
              } else { 
                org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
              }
              break;
｝

Thrift源码分析（二）-- 协议和编解码

标签：程序   switch   hand   throws   while   框架   UI   mask   rds   

原文地址：https://www.cnblogs.com/heapStark/p/9218394.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！