RabbitMQ 教程 - 远程过程调用 (RPC)

远程过程调用 (RPC)

(使用 AMQP 1.0 .NET 客户端)

信息

先决条件

本教程假设 RabbitMQ 已 安装 并在 localhost 上的 标准端口 (5672) 上运行。如果您使用不同的主机、端口或凭据，则需要调整连接设置。

哪里寻求帮助

如果您在学习本教程时遇到困难，可以通过 GitHub Discussions 或 RabbitMQ 社区 Discord 联系我们。

在上一篇教程中，我们使用了主题交换机 (topic exchanges)。在本教程中，我们将实现请求/回复 (RPC) 模式：客户端发送一个请求并等待响应。

示例 RPC 服务将在服务器上计算斐波那契数列。服务器在仲裁队列 (quorum queue) rpc_queue 上使用 IResponder；客户端使用 IRequester 发布请求并等待回复。

服务器 (RPCServer)

IResponder responder = await connection.ResponderBuilder()
    .RequestQueue(rpcQueueName)
    .Handler((ctx, request) =>
    {
        string response = "";
        try
        {
            string message = Encoding.UTF8.GetString(request.Body()!);
            int n = int.Parse(message);
            Console.WriteLine($" [.] fib({message})");
            response += Fib(n);
        }
        catch (Exception e)
        {
            Console.WriteLine($" [.] {e.Message}");
        }

        return Task.FromResult(ctx.Message(Encoding.UTF8.GetBytes(response)));
    })
    .BuildAsync();

客户端 (RPCClient)

IRequester requester = await connection.RequesterBuilder()
    .RequestAddress()
    .Queue("rpc_queue")
    .Requester()
    .BuildAsync();

IMessage request = new AmqpMessage(Encoding.UTF8.GetBytes(iStr));
IMessage reply = await requester.PublishAsync(request);

本教程的客户端在循环中请求 fib(0) 到 fib(31)，这与其他 AMQP 1.0 教程的端口保持一致。

运行

dotnet run --project RPCServer/RPCServer.csproj
dotnet run --project RPCClient/RPCClient.csproj

源码

• RPCServer/Program.cs
• RPCClient/Program.cs

关于 RPC 的说明

RPC 虽然常见但很容易被误用：请明确区分哪些调用是本地调用，哪些是远程调用，记录好依赖关系，并妥善处理代理或服务器故障。如有疑问，请优先选择异步流水线而不是阻塞式 RPC。

有关 AMQP 1.0 和 RabbitMQ 的更多信息，请参阅 RabbitMQ 中的 AMQP 和 AMQP 1.0 客户端库。