Seata 协议编码

男娘i 2022-12-18 05:51 113阅读 0赞

# Seata 协议编码 #

协议格式如下:

<table> 
 <thead> 
  <tr> 
   <th>Len</th> 
   <th>Param</th> 
   <th>Desc</th> 
   <th>Desc in chinese</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>2B</td> 
   <td>Magic Code</td> 
   <td>0xdada</td> 
   <td>魔术位</td> 
  </tr> 
  <tr> 
   <td>1B</td> 
   <td>ProtocolVersion</td> 
   <td>1</td> 
   <td>协议版本：用于非兼容性升级</td> 
  </tr> 
  <tr> 
   <td>4B</td> 
   <td>FullLength</td> 
   <td>include front 3 bytes and self 4 bytes</td> 
   <td>总长度 ：用于拆包，包括前3位和自己4位</td> 
  </tr> 
  <tr> 
   <td>2B</td> 
   <td>HeadLength</td> 
   <td>include front 7 bytes, self 4 bytes, and head map</td> 
   <td>头部长度：包括前面7位，自己4位，以及 HeadMap</td> 
  </tr> 
  <tr> 
   <td>1B</td> 
   <td>Message type</td> 
   <td>request(oneway/twoway)/response/heartbeat/callback</td> 
   <td>消息类型：请求（单向/双向）/响应/心跳/回调/go away等</td> 
  </tr> 
  <tr> 
   <td>1B</td> 
   <td>Serialization</td> 
   <td>custom, hessian, pb</td> 
   <td>序列化类型：内置/hessian/protobuf等</td> 
  </tr> 
  <tr> 
   <td>1B</td> 
   <td>CompressType</td> 
   <td>None/gzip/snappy...</td> 
   <td>压缩算法：无/gzip/snappy</td> 
  </tr> 
  <tr> 
   <td>4B</td> 
   <td>MessageId</td> 
   <td>Integer</td> 
   <td>消息 Id</td> 
  </tr> 
  <tr> 
   <td>2B</td> 
   <td>TypeCode</td> 
   <td>code in AbstractMessage</td> 
   <td>消息类型: AbstractMessage 里的类型</td> 
  </tr> 
  <tr> 
   <td>?B</td> 
   <td>HeadMap[Optional]</td> 
   <td>exists when if head length &gt; 16</td> 
   <td>消息Map（可选的，如果头部长度大于16，代表存在HeadMap）</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td></td> 
   <td>ATTR_KEY(？B) key:string:length(2B)+data ATTR_TYPE(1B) 1:int; 2:string; 3:byte; 4:short ATTR_VAL(?B) int:(4B); string:length(2B)+data; byte:(1B); short:(2B) }</td> 
   <td>Key: 字符串 Value 类型Value 值</td> 
  </tr> 
  <tr> 
   <td>?b</td> 
   <td>Body</td> 
   <td>(FullLength-HeadLength)</td> 
   <td>请求体:长度为总长度-头长度</td> 
  </tr> 
 </tbody> 
</table>

1.  前面两个字节(0,2)存储的是魔数
2.  第三个字节(2,3)存储协议版本, 默认为 1, 目前没什么用
3.  后续4个字节(3-7)存储的是所有内容的长度(请求头长度+请求体长度)
4.  后续2个字节(7-9)存储的是请求头的长度
5.  后续的3个字节(9-11) 存储的分别是消息类型，序列化方式(默认jackson)，压缩方式(默认不压缩)
6.  后续的4个字节(12-16) 存储的是消息类型
7.  请求头的内容(可选)
8.  请求体的内容

Seata 由于只需要和协调者交互， 所以把消息分为两种类型， 一种是事务管理器和协调者(TC)的交互，另一种是资源管理其(RM)和协调者(TC)的交互

魔数是一个固定的常量，通常定义协议都会定义一个魔术，主要用于处理粘包， 协议版本在1.3以及以前都是1, 是一个常量，

后面接着的 头和体的长度，头的长度都比较容易理解， 后面的消息类型，序列化，压缩方式 看后续的单独说明， 接着是请求的ID,

对于同步请求而言，由发起者生成请求ID， 放在ConcurrentHashMap 中， 然后向协调者发起请求，当前线程等待， 当协调者发送响应消息时， 会把requestId返回回来， 然后响应现线程移除Map里面的id， 发送通知，让请求线程继续处理。

后面就是用于扩展的请求头了, 目前没有地方使用。 然后就是具体的内容了。

对具体的内容进行序列化得到内容长度之后， 然后在回到头部，将Full Length 和 Head Length 会写

## 消息类型 ##

消息类型在seata中用常量表示: 在类 io.seata.core.protocol.ProtocolConstants 中, 是字节类型

1.  MSGTYPE\_RESQUEST\_SYNC = 0x0; 同步请求
2.  MSGTYPE\_RESPONSE = 0x1; 响应
3.  MSGTYPE\_RESQUEST\_ONEWAY = 0x3; 不需要响应的请求
4.  MSGTYPE\_HEARTBEAT\_REQUEST = 0x4; 心跳请求
5.  MSGTYPE\_HEARTBEAT\_RESPONSE = 0x5; 心跳响应

## 事物管理器 ##

事物管理器™用于向事物协调者(TC)发起请求, 事物管理器需要和协调者交互的消息如下:

1.  全局事物开始请求(GlobalBeginRequest)，向协调者发起全局事物开始请求
2.  全局事物提交请求(GlobalCommitRequest)，向协调者发起全局事物提交请求
3.  全局事物回滚请求(GlobalRollbackRequest) , 向协调者发起全局事物回滚请求
4.  全局报告请求(GlobalReportRequest), 用于告诉协调者分支事物的成功状态
5.  全局状态请求(GlobalStatusRequest), 回滚失败，提交失败，重试回滚的时候， 会向协调者请求全局事物的状态

## 资源管理器 ##

资源管理器用于处理由协调者发起的请求，资源管理器需要和协调者交互的消息如下:

1.  分支注册请求(BranchRollbackRequest), 执行分支事物回滚. AT模式下会恢复原先的数据.
2.  分支提交请求(BranchCommitRequest), 执行分支事物提交， AT模式下提交成功时，会删除回滚日志(undo\_log).
3.  删除回滚日志请求(UndoLogDeleteRequest), 删除本地的回滚日志, 只有AT模式才需要处理
4.  分支报告请求(BranchReportRequest), 如果开启了配置`client.rm.reportSuccessEnable=true`, 在AT，XA模式本地的分支事物执行完成后， 向协调者报告执行结果

# 序列化方式 #

Serializer 只有两个方法:

public interface Serializer { 
    
        <T> byte[] serialize(T t);
    
        <T> T deserialize(byte[] bytes);
    }

Seata 1.3 提供了五种序列化方式:

1.  FstSerializer (io.seata.serializer.fst)
2.  HessianSerializer (io.seata.serializer.hessian)
3.  KryoSerializer (io.seata.serializer.kryo)
4.  SeataSerializer (io.seata.serializer.seata)
5.  ProtobufSerializer (io.seata.serializer.protobuf)

在和协调者交互的时候，需要将消息序列化， 通过 SerializerFactory 进行序列化， SerializerFactory 内部使用了 Seata 的SPI机制，默认设置为: SeataSerializer。

可以通过配置 transport.serialization 修改序列化方式, 每个序列化类都有注解， 例如想要设置为 KryoSerializer , 就可以设置:

transport.serialization = KRYO

# 数据压缩 #

数据压缩接口: io.seata.core.compressor.Compressor

这里同样也是通过Seata的SPI机制加载

Seata 1.3 提供了五种数据压缩方式

1.  io.seata.core.compressor.CompressorFactory.NoneCompressor
2.  io.seata.compressor.zip.ZipCompressor
3.  io.seata.compressor.sevenz.SevenZCompressor
4.  io.seata.compressor.lz4.Lz4Compressor
5.  io.seata.compressor.gzip.GzipCompressor
6.  io.seata.compressor.bzip2.BZip2Compressor

压缩数据能减少网络传输的数据，同时也会增加应用服务器的CPU负担。

在和协调者交互的时候，需要将消息序列化， 通过 SerializerFactory 进行序列化， SerializerFactory 内部使用了 Seata 的SPI机制，默认设置为: NoneCompressor,也就是不压缩数据.

可以通过配置 transport.compressor 修改压缩方式, 每种压缩方式都有注解， 例如想要设置为 GzipCompressor , 就可以设置:

transport.compressor = GZIP