Apache Thrift 与 Rebus 服务总线集成实战基于 RabbitMQ 的异步 oneway RPC 示例剖析【免费下载链接】thriftApache Thrift项目地址: https://gitcode.com/GitHub_Trending/thr/thrift导读本文围绕 Apache Thrift 仓库contrib/Rebus目录下的示例工程完整讲解如何将 Thrift 的 RPC 调用与 Rebus一个轻量级 .NET 服务总线相结合通过 RabbitMQ 消息队列实现异步的请求—应答通信。读者将掌握 Thrift 序列化字节流与消息队列负载的桥接方法、oneway void异步调用的设计范式、代码与配置文件混合的 Rebus 配置方式以及该示例为何被官方标记为弃用并迁移到 netstd 的来龙去脉。Rebus 是什么为什么要与 Thrift 结合关联文档 开门见山地说明了本示例的定位它是 Thrift 与 Rebus 组合使用的示例代码。Rebus 是一个 .NET 服务总线与 NServiceBus 定位类似但更加轻量主要由 Mogens Heller Grabe 编写。在典型的 Thrift RPC 场景中客户端直接通过 Socket 同步调用服务端而引入 Rebus 之后调用不再是直连而是变成投递客户端把 Thrift 调用序列化成一个消息通过 Rebus 发往 RabbitMQ服务端从队列中取出消息、反序列化、执行调用再把结果同样通过队列送回客户端。这种架构带来了天然的异步解耦、削峰填谷和进程间隔离能力代价是放弃了同步调用的即时响应。核心设计原则一切调用都是 oneway void文档强调了一个贯穿全示例的硬性约束As with all ServiceBus or MQ scenarios, due to the highly asynchronous operations it is recommended to do all calls as oneway void calls.服务总线/消息队列场景天然是异步的因此所有调用都应该声明为oneway void。这一点在 sample.thrift 中被严格执行——两个服务共 5 个方法全部是oneway voidservice BasicMathServer { oneway void DoTheMath( 1: i32 arg1, 2: i32 arg2) oneway void Ping(1: i64 value) } service BasicMathClient { oneway void ThreeResults( 1 : i32 added, 2 : i32 multiplied, 3 : i32 subtracted); oneway void FourResults( 1 : i32 added, 2 : i32 multiplied, 3 : i32 subtracted, 4 : i32 divided); oneway void Pong(1: i64 value) }oneway意味着调用方发出请求后立即返回不等待任何应答这与消息队列的投递语义完全吻合。示例中的请求—应答实际上是靠两个独立队列MathRequests与MathResponses和两个 Thrift 服务拼出来的BasicMathServer承载请求BasicMathClient承载应答二者互为对方的远程服务形成双向异步对等通信。示例架构单进程内的两个队列监听器在 Program.cs 中可以看到示例在同一个进程里启动了两个 Rebus 端点adapter分别监听两个队列static BuiltinContainerAdapter StartRequestServer(string server) { var adapter new BuiltinContainerAdapter(); Configure.With(adapter) .Transport(t t.UseRabbitMq(amqp:// server, MathRequests, MathRequestErrors)) .MessageOwnership(o o.FromRebusConfigurationSection()) .CreateBus().Start(); adapter.Register(typeof(MathRequestCallHandler)); return adapter; } static BuiltinContainerAdapter StartResponseServer(string server) { var adapter new BuiltinContainerAdapter(); Configure.With(adapter) .Transport(t t.UseRabbitMq(amqp:// server, MathResponses, MathResponseErrors)) .MessageOwnership(o o.FromRebusConfigurationSection()) .CreateBus().Start(); adapter.Register(typeof(MathResponseCallHandler)); return adapter; }MathRequests/MathRequestErrors承载计算请求的输入队列与错误队列MathResponses/MathResponseErrors承载计算结果应答的输入队列与错误队列。Main方法依次启动两个端点随后创建MathRequestClient并发送第一条DoTheMath消息之后阻塞等待用户按回车退出var req StartRequestServer(server); var rsp StartResponseServer(server); var random new Random(); var client new MathRequestClient(server); client.DoTheMath(random.Next(), random.Next()); Console.Write(Hit ENTER to stop ... ); Console.ReadLine();整体消息流转如下客户端将DoTheMath调用序列化后投递到MathRequests队列 → 服务端MathRequestCallHandler反序列化并执行计算 → 服务端将结果通过MathResponseClient序列化后投递到MathResponses队列 → 客户端MathResponseCallHandler反序列化并打印结果随后主动发起下一次计算请求形成乒乓式的连续对话。消息容器设计通用字节载体 每服务专用子类Rebus 依靠消息类型做 handler 路由因此示例在 ServiceImpl/Both.cs 中定义了一个巧妙的类层次// generic data container for serialized Thrift calls public class GenericThriftServiceCall { public byte[] rawBytes; } // specific containers (one per Thrift service) to leverage Rebus handler routing public class MathRequestCall : GenericThriftServiceCall { } public class MathResponseCall : GenericThriftServiceCall { }GenericThriftServiceCall只是一个装 Thrift 序列化字节的通用载体MathRequestCall与MathResponseCall各对应一个 Thrift 服务目的是让 Rebus 能根据消息的具体类型把消息路由到正确的 handler。这是通用序列化 类型化路由的经典组合序列化格式统一Thrift 二进制协议路由维度独立消息类型。客户端实现序列化 Thrift 调用并投递到队列ServiceImpl/Client.cs 中MathRequestClient是客户端一侧的核心它实现BasicMathServer.Iface从而对上层代码隐藏队列细节——调用方就像在调用一个普通的 Thrift 客户端public void SerializeThriftCall(ActionBasicMathServer.Iface action) { // Thrift protocol/transport stack var stm new MemoryStream(); var trns new TStreamTransport(null, stm); var prot new TBinaryProtocol(trns); // serialize the call into a bunch of bytes var client new BasicMathServer.Client(prot); if (action ! null) action(client); else throw new ArgumentException(action must not be null); // make sure everything is written to the MemoryStream trns.Flush(); // send the message var msg new MathRequestCall() { rawBytes stm.ToArray() }; MQAdapter.Bus.Send(msg); } public void Ping(long value) { SerializeThriftCall(client { client.Ping(value); }); } public void DoTheMath(int arg1, int arg2) { SerializeThriftCall(client { client.DoTheMath(arg1, arg2); }); }这段代码演示了本示例最关键的模式——MemoryStream TStreamTransport TBinaryProtocol 生成的 Client/Processor四件套构造一个写方向的MemoryStream用TStreamTransport(null, stm)把 Thrift 传输层接到该内存流上用TBinaryProtocol(trns)叠加二进制协议层用生成的BasicMathServer.Client(prot)执行一次方法调用调用结果被序列化进MemoryStreamtrns.Flush()确保字节全部写入取出stm.ToArray()装入消息容器通过MQAdapter.Bus.Send(msg)投递到 Rebus 消息总线。MathRequestClient的构造函数使用了UseRabbitMqInOneWayMode即只发不收的单向模式因为客户端只需要发送请求Configure.With(MQAdapter) .Transport(t t.UseRabbitMqInOneWayMode(amqp:// server)) // we need send only .MessageOwnership(o o.FromRebusConfigurationSection()) .CreateBus().Start();客户端另一侧是响应处理MathResponseCallHandler实现IHandleMessagesMathResponseCall从消息中取出rawBytes用读方向的TStreamTransport(stm, null)反序列化再交给BasicMathClient.Processor执行public void Handle(MathResponseCall message) { // Thrift protocol/transport stack var stm new MemoryStream(message.rawBytes); var trns new TStreamTransport(stm, null); var prot new TBinaryProtocol(trns); // create a processor and let him handle the call var hndl new MathResponsesHandler(); var proc new BasicMathClient.Processor(hndl); proc.Process(prot, null); // oneway only }proc.Process(prot, null)中第二个参数传null注释标明oneway only——因为所有方法都是oneway void不存在需要写回的应答输出 protocol 可以直接置空。MathResponsesHandler在收到结果后打印四则运算结果并立即发起Ping与下一轮DoTheMath形成持续对话public void FourResults(int added, int multiplied, int subtracted, int divided) { Console.WriteLine(added {0}, added); Console.WriteLine(multiplied {0}, multiplied); Console.WriteLine(subtracted {0}, subtracted); Console.WriteLine(divided {0}, divided); PingAndDoAnotherCalculation(); }Pong(long value)则利用DateTime.Now.Ticks - value粗略计算一次 Ping 的往返延迟。服务端实现处理请求、回投结果ServiceImpl/Server.cs 与客户端严格对称。MathRequestCallHandler从MathRequests队列取出请求字节反序列化后交给BasicMathServer.Processor执行MathRequestsHandler实现BasicMathServer.Iface执行真实计算逻辑并通过MathResponseClient把结果投回MathResponses队列public void DoTheMath(int arg1, int arg2) { var client new MathResponseClient(localhost); if (arg2 ! 0) client.FourResults(arg1 arg2, arg1 * arg2, arg1 - arg2, arg1 / arg2); else client.ThreeResults(arg1 arg2, arg1 * arg2, arg1 - arg2); }注意对除数为 0 的防御性处理arg2 0时退化为ThreeResults不返回除法结果并在客户端一侧打印 DIV/0 error during division。MathResponseClient与客户端的MathRequestClient结构完全一致同样使用UseRabbitMqInOneWayModeSerializeThriftCall模式只是换成了BasicMathClient的生成代码。由此可以看出示例中的服务端与客户端本质上是对等的两个队列监听者一个负责接收请求BasicMathServer一个负责接收应答BasicMathClient二者通过两个队列形成闭环。配置方式App.config 与代码混合文档明确指出 Rebus 的配置可以通过 App.Config、纯代码或两者混合完成并说明本示例因为在单个进程中实现了两个队列监听器所以把输入队列和错误队列放在代码里配置即前文UseRabbitMq的传入参数而消息到端点的路由映射则放在 App.config 中configSections section namerebus typeRebus.Configuration.RebusConfigurationSection, Rebus/ /configSections rebus inputQueueMyResponses errorQueueMyErrors workers1 endpoints add messagesRebusSample.MathRequestCall, RebusSample endpointMathRequests/ add messagesRebusSample.MathResponseCall, RebusSample endpointMathResponses/ /endpoints /rebus配置要点configSections中注册rebus配置节类型为Rebus.Configuration.RebusConfigurationSectionrebus元素的inputQueue/errorQueue定义了默认输入与错误队列示例中为MyResponses/MyErrorsworkers指定并发 worker 数endpoints内的每条add将某个消息类型路由到指定队列端点MathRequestCall→MathRequestsMathResponseCall→MathResponses。这正是代码中.MessageOwnership(o o.FromRebusConfigurationSection())所读取的内容——MessageOwnership决定某类消息应该发给哪个端点。这种队列 传输配置放代码、消息路由映射放配置文件的混合方式正是文档所描述的mixed from both locations的典型实践。构建与运行环境RebusSample.csproj 揭示了工程的依赖与构建方式目标框架.NET Framework 4.5TargetFrameworkVersionv4.5解决方案文件 RebusSample.sln 为 Visual Studio 2013 格式外部依赖需额外安装 RabbitMQ .NET 客户端文档明确要求可经 NuGet 获取同时引用 Rebus 与 Rebus.RabbitMQ 程序集以及 Thrift 的 C# 库工程ProjectReference指向lib/csharp/src/Thrift.csproj代码生成工程通过PreBuildEvent在构建前自动执行thrift -gen csharp sample.thrift把 sample.thrift 编译成gen-csharp\BasicMathServer.cs与gen-csharp\BasicMathClient.cs再参与编译。这意味着开发者本地需先安装 Thrift 编译器并配置好上述 DLL 引用路径运行前提本地需有一个可用的 RabbitMQ 服务默认连接amqp://localhost可通过Main中的server变量修改启动后程序会自动创建队列并持续进行请求—应答循环。跨语言互通需要自定义序列化器文档特别提醒如果希望与非 .NET 语言通信需要自定义序列化器以覆盖 Rebus 的默认线格式。原因在于 Rebus 默认使用自己的 .NET 序列化格式BinaryFormatter 一类其他语言无法解析而本示例正是通过先把 Thrift 调用序列化为rawBytes再放进 Rebus 消息的方式绕开了这一限制——消息正文对 Rebus 而言只是一段不透明字节真正的语义由 Thrift 二进制协议承载。从仓库结构看同样的Thrift 字节入队列思路在 contrib/zeromqZeroMQ 传输示例中也有体现说明以字节数组为载荷、跨传输层搬运 Thrift 帧是 Apache Thrift 生态中连接各类消息中间件的通用手法。弃用说明从 C# 迁移到 netstd文档末尾附有明确的弃用声明C# 已不再是 Apache Thrift 官方支持的目标语言netstd.NET Standard是推荐的替代方案本示例代码原样保留仅供教学目的除非有人将其移植到 netstd。这一声明与仓库当前状态吻合在 lib 目录下已经不存在csharp子目录本示例 csproj 中引用的lib/csharp/src/Thrift.csproj路径如今已失效取而代之的是 lib/netstd其中包含 Thrift/Protocol/TBinaryProtocol.cs、Thrift/Transport/Client/TStreamTransport.cs 等与现代 .NET 生态.NET Standard配套的实现。示例中使用的TStreamTransport、TBinaryProtocol、生成的Client/Processor等核心抽象在 netstd 中均有对应实现因此本文所述的序列化字节 → 消息队列投递模式在迁移到 netstd 后依然成立迁移工作主要集中在工程格式csproj/包引用与 API 命名空间层面。小结contrib/Rebus是一个信息密度很高的教学示例它演示了三条可复用的工程经验一是服务总线场景下坚持oneway void的异步调用设计二是用MemoryStreamTStreamTransportTBinaryProtocol 生成代码把 Thrift 调用编码为字节载荷从而与任何消息中间件解耦三是通过通用字节容器 每服务专用消息子类同时满足序列化统一与 Rebus 类型化路由的需求。对于需要在 .NET 生态中把 Thrift 与消息队列结合使用的开发者这份代码配合其 README是理解该集成模式的直接入口同时务必注意其 C# 目标已弃用新项目应转向 netstd 版本。【免费下载链接】thriftApache Thrift项目地址: https://gitcode.com/GitHub_Trending/thr/thrift创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考