如何使用 Thrift 代替 RPC 进行消息传递答案

【问题标题】：How to use Thrift for message passing instead of RPC如何使用 Thrift 代替 RPC 进行消息传递
【发布时间】：2018-08-31 22:20:20
【问题描述】：

我知道 Thrift 主要针对成熟的客户端-服务器 RPC，但从高级架构来看，在我看来它也应该完全适用于双向消息传递。

我想在两端（C、.NET Core）上构建的内容如下：

接收方法：引用套接字，读取完整消息，反序列化，返回
Send 方法：引用套接字，序列化给定消息，将其发送到线路。

我不需要线程服务器，......任何花哨的东西。基本上我想得到什么，例如Protobuffs 提供的是开箱即用的处理方式，可以在接收端缓冲整个消息，通常是消息帧。

问题是我找不到任何关于如何使用当前库（我个人对 .NET Core 和 C 库）API 开始构建的文档。我发现的唯一东西是这个question，但它并没有真正指向任何资源。

【问题讨论】：

标签： c .net-core thrift

【解决方案1】：

做一些非常相似的事情的一些笔记：

同时使用 C#（.Net Core 和 Framework 的组合）和 C++ 的客户端
使用 Thrift RPC 以及“普通消息”进行发布/订阅和常规序列化

The suggestion to put all messages in a top-level union 是一个不错的选择，因为它可以更轻松地反序列化消息。

鉴于以下节俭：

struct SubscribeRequest {
    1: string topic,
    2: string appid,
}

struct SubscribeReply {
    1: bool success,
    2: string topic,
}
service HttpService {
    HttpSDKDataTypes.SubscribeReply Subscribe(1: HttpSDKDataTypes.SubscribeRequest message),
}

thrift -gen netcore 给你：

  public async Task<Ruyi.SDK.Http.SubscribeReply> SubscribeAsync(Ruyi.SDK.Http.SubscribeRequest message, CancellationToken cancellationToken)
  {
    await OutputProtocol.WriteMessageBeginAsync(new TMessage("Subscribe", TMessageType.Call, SeqId), cancellationToken);

消息标识符包含在 RPC 调用中。如果您不使用 RPC 调用，您将获得“原始”结构，而没有指示如何反序列化它们。

将它们放入union：

union UnionExample {
    1: SubscribeRequest request,
    2: SubscribeReply reply,    
}

为您解决这个问题：

public async Task WriteAsync(TProtocol oprot, CancellationToken cancellationToken)
{
  oprot.IncrementRecursionDepth();
  try
  {
    var struc = new TStruct("SubscribeRequest");
    await oprot.WriteStructBeginAsync(struc, cancellationToken);
    var field = new TField();
    if (Topic != null && __isset.topic)
    {
      field.Name = "topic";
      field.Type = TType.String;
      field.ID = 1;

反序列化后，可以用：

    public void handler(UnionExample example)
    {
        if (example.__isset.request)
        {
            SubscribeRequest msg = example.request;
            // ...
        }
        else if (example.__isset.reply)
        {
            SubscribeReply msg = example.reply;
            // ...
        }

检查生成器中可用的选项。 thrift -gen "csharp:union,async"让你用pattern matching：

   public void handler(UnionExample example)
    {
        switch (example.Data)
        {
            case SubscribeRequest msg:
                //...
            case SubscribeReply msg:
                //...

不幸的是，netcore 生成器在 0.11.0 中没有这样做。

thrift github 存储库有examples of serializing into memory（而不是使用 RPC）。一般来说，它类似于：

Stream stm = new MemoryStream();
TTransport trans = new TStreamTransport(null, stm);
TProtocol prot = new TJSONProtocol(trans);

如果您要创建大量 MemoryStream 实例，请查看 Microsoft.IO.RecycableMemoryStream。

使用自定义协议/传输将简化发送消息的过程，因为它会处理样板文件（并避免先序列化到内存的额外对象，然后再用它做什么）。提到了thrift contrib/ folder。 Here's our C# example 在 ZeroMQ 上使用 Thrift。

最后一点。如果您打算使用 RPC 功能，请使用单个结构参数编写服务方法。含义：

service HttpService {
    // Do this
    string Subscribe(1: SubscribeRequest message),
    // Not this
    string Subscribe(1: string topic, 2: string appid,),
}

这将使摆脱 RPC 和/或重用消息变得更加容易。

【讨论】：

【解决方案2】：

Thrift 是一个 RPC 和序列化框架。这意味着，您也可以只使用没有 RPC 的序列化部分。

结合消息传递系统，通常（大致）如下：

将消息序列化到缓冲区中
以任何您想要的方式发送该缓冲区
接收端反序列化缓冲区并处理数据

如果您计划通过同一渠道发送不同类型的消息，最好有一个包含所有可能消息正文的union 信封结构：

 struct MessageOne {
      // contents of this message type
 }

 struct MessageTwo {
      // contents of this message type
 }

 struct MessageThree {
      // contents of this message type
 }

 union MyMessageEnvelope {
      1: MessageOne   one
      2: MessageTwo   two
      3: MessageThree  three
      // easily extendable 
 }

为了使其更加优雅/可重用，还可以实现自定义传输来满足需求并更多地封装逻辑。 Thrift 的模块化结构使其变得容易（您链接的帖子也提到了这一点）。源代码树的/contrib 文件夹中有一些示例，可以作为起点。

如果您完全不知道从哪里开始：查看教程，然后查看测试套件程序，这两者都是 Thrift 初学者的学习资源。

【讨论】：