做pc端网站教程,聊城市住房和城乡建设局网站首页,潮州seo网站推广,广州建站外包先决条件本教程假定 RabbitMQ 已经安装#xff0c;并运行在localhost标准端口#xff08;5672#xff09;。如果你使用不同的主机、端口或证书#xff0c;则需要调整连接设置。从哪里获得帮助如果您在阅读本教程时遇到困难#xff0c;可以通过邮件列表 联系我们。在第 教程… 先决条件本教程假定 RabbitMQ 已经安装并运行在localhost标准端口5672。如果你使用不同的主机、端口或证书则需要调整连接设置。从哪里获得帮助如果您在阅读本教程时遇到困难可以通过邮件列表 联系我们。在第 教程[2] 中我们学习了如何使用工作队列在多个工作单元之间分配耗时任务。但是如果我们想要运行一个在远程计算机上的函数并等待其结果呢这将是另外一回事了。这种模式通常被称为 远程过程调用 或 RPC 。在本篇教程中我们将使用 RabbitMQ 构建一个 RPC 系统一个客户端和一个可扩展的 RPC 服务器。由于我们没有什么耗时任务值得分发那干脆就创建一个返回斐波那契数列的虚拟 RPC 服务吧。客户端接口为了说明如何使用 RPC 服务我们将创建一个简单的客户端类。该类将暴露一个名为Call的方法用来发送 RPC 请求并且保持阻塞状态直到接收到应答为止。var rpcClient new RPCClient();Console.WriteLine( [x] Requesting fib(30));
var response rpcClient.Call(30);
Console.WriteLine( [.] Got {0}, response);rpcClient.Close();关于 RPC 的说明尽管 RPC 在计算机中是一种很常见的模式但它经常受到批评。问题出现在当程序员不知道一个函数是本地调用还是一个耗时的 RPC 请求。这样的混淆会导致系统不可预测以及给调试增加不必要的复杂性。误用 RPC 可能会导致不可维护的混乱代码而不是简化软件。牢记这些限制请考虑如下建议确保可以明显区分哪些函数是本地调用哪些是远程调用。为您的系统编写文档明确组件之间的依赖关系。捕获异常当 RPC 服务长时间宕机时客户端该如何应对。当有疑问的时候可以先避免使用 RPC。如果可以的话考虑使用异步管道 - 而不是类似 RPC 的阻塞其会将结果以异步的方式推送到下一个计算阶段。回调队列一般来讲基于 RabbitMQ 进行 RPC 通信是非常简单的客户端发送一个请求消息然后服务端用一个响应消息作为应答。为了能接收到响应我们需要在发送请求过程中指定一个callback队列地址。Copyvar props channel.CreateBasicProperties(); props.ReplyTo replyQueueName;var messageBytes Encoding.UTF8.GetBytes(message); channel.BasicPublish(exchange: , routingKey: rpc_queue, basicProperties: props, body: messageBytes);// ... then code to read a response message from the callback_queue ...消息属性AMQP 0-9-1 协议在消息中预定义了一个包含 14 个属性的集合大多数属性很少使用但以下情况除外Persistent将消息标记为持久的值为2或者瞬时的其他值可以参考 教程[2]。DeliveryMode熟悉 AMQP 协议的人可以选择此属性而不是熟悉协议的人可以选择使用此属性而不是Persistent它们控制的东西是一样的。ContentType用于描述编码的 mime 类型。例如对于经常使用的 JSON 编码将此属性设置为application/json是一种很好的做法。ReplyTo通常用于命名回调队列。CorrelationId用于将 RPC 响应与请求相关联。关联ID在上面介绍的方法中我们建议为每个 RPC 请求创建一个回调队列但是这种方式效率低。幸运的是我们有一种更好的方式那就是为每个客户端创建一个独立的回调队列。这种方式会引出一个新的问题在收到响应的回调队列中它无法区分响应属于哪一个请求此时便是CorrelationId属性的所用之处。我们将为每个请求的CorrelationId设置一个唯一值。之后当我们在回调队列接收到响应的时候再去检查下这个属性是否和请求中的值匹配如此一来我们就可以把响应和请求关联起来了。如果出现一个未知的CorrelationId值我们可以安全的销毁这个消息因为这个消息不属于我们的请求。你可能会问为什么我们应该忽略回调队列中的未知的消息而不是用错误来标识失败呢这是因为于服务器端可能存在竞争条件。虽然不太可能但是 RPC 服务器可能在仅发送了响应消息而未发送消息确认的情况下挂掉如果出现这种情况RPC 服务器重启之后将会重新处理该请求。这就是为什么在客户端上我们必须优雅地处理重复的响应并且理想情况下 RPC 应该是幂等的。总结我们的 RPC 会是这样工作客户端启动时会创建一个匿名的独占回调队列。对于 RPC 请求客户端发送带有两个属性的消息ReplyTo设置为回调队列和CorrelationId为每个请求设置唯一值。请求被发送到rpc_queue队列。RPC 工作线程或者叫服务器正在等待该队列上的请求。当出现请求时它会执行该作业并使用ReplyTo属性设置的队列将带有结果的消息发送回客户端。客户端等待回调队列上的数据。出现消息时它会检查CorrelationId属性。如果它与请求中的值匹配则返回对应用程序的响应。组合在一起斐波纳契 任务private static int fib(int n){ if (n 0 || n 1) return n; return fib(n - 1) fib(n - 2);
}我们宣布我们的斐波那契函数。并假定只允许有效的正整数输入。 不要期望这个适用于大数字它可能是最慢的递归实现。我们的 RPC 服务端代码 RPCServer.cs 看起来如下所示using System;using RabbitMQ.Client;using RabbitMQ.Client.Events;using System.Text;class RPCServer{ public static void Main() { var factory new ConnectionFactory() { HostName localhost }; using (var connection factory.CreateConnection()) using (var channel connection.CreateModel()){channel.QueueDeclare(queue: rpc_queue, durable: false,exclusive: false, autoDelete: false, arguments: null);channel.BasicQos(0, 1, false); var consumer new EventingBasicConsumer(channel);channel.BasicConsume(queue: rpc_queue,autoAck: false, consumer: consumer);Console.WriteLine( [x] Awaiting RPC requests);consumer.Received (model, ea) { string response null; var body ea.Body; var props ea.BasicProperties; var replyProps channel.CreateBasicProperties();replyProps.CorrelationId props.CorrelationId; try{ var message Encoding.UTF8.GetString(body); int n int.Parse(message);Console.WriteLine( [.] fib({0}), message);response fib(n).ToString();} catch (Exception e){Console.WriteLine( [.] e.Message);response ;} finally{ var responseBytes Encoding.UTF8.GetBytes(response);channel.BasicPublish(exchange: , routingKey: props.ReplyTo,basicProperties: replyProps, body: responseBytes);channel.BasicAck(deliveryTag: ea.DeliveryTag,multiple: false);}};Console.WriteLine( Press [enter] to exit.);Console.ReadLine();}} /// /// Assumes only valid positive integer input./// Dont expect this one to work for big numbers, and its/// probably the slowest recursive implementation possible./// private static int fib(int n) { if (n 0 || n 1){ return n;} return fib(n - 1) fib(n - 2);}
}服务端代码非常简单像往常一样首先建立连接通道和声明队列。我们可能希望运行多个服务器进程。为了在多个服务器上平均分配负载我们需要设置channel.BasicQos中的prefetchCount值。使用BasicConsume访问队列然后注册一个交付处理程序并在其中完成工作并发回响应。我们的 RPC 客户端 RPCClient.cs 代码using System;using System.Collections.Concurrent;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;public class RpcClient{ private readonly IConnection connection; private readonly IModel channel; private readonly string replyQueueName; private readonly EventingBasicConsumer consumer; private readonly BlockingCollectionstring respQueue new BlockingCollectionstring(); private readonly IBasicProperties props;public RpcClient(){ var factory new ConnectionFactory() { HostName localhost };connection factory.CreateConnection();channel connection.CreateModel();replyQueueName channel.QueueDeclare().QueueName;consumer new EventingBasicConsumer(channel);props channel.CreateBasicProperties(); var correlationId Guid.NewGuid().ToString();props.CorrelationId correlationId;props.ReplyTo replyQueueName;consumer.Received (model, ea) { var body ea.Body; var response Encoding.UTF8.GetString(body); if (ea.BasicProperties.CorrelationId correlationId){respQueue.Add(response);}};} public string Call(string message) { var messageBytes Encoding.UTF8.GetBytes(message);channel.BasicPublish(exchange: ,routingKey: rpc_queue,basicProperties: props,body: messageBytes);channel.BasicConsume(consumer: consumer,queue: replyQueueName,autoAck: true); return respQueue.Take(); ;} public void Close() {connection.Close();}
} public class Rpc{ public static void Main() { var rpcClient new RpcClient();Console.WriteLine( [x] Requesting fib(30)); var response rpcClient.Call(30);Console.WriteLine( [.] Got {0}, response);rpcClient.Close();}
}客户端代码稍微复杂一些建立连接和通道并为响应声明一个独有的 callback 队列。订阅这个 callback 队列以便可以接收到 RPC 响应。Call方法用来生成实际的 RPC 请求。在这里我们首先生成一个唯一的CorrelationId编号并保存它while 循环会使用该值来捕获匹配的响应。接下来我们发布请求消息其中包含两个属性ReplyTo和CorrelationId。此时我们可以坐下来稍微一等直到指定的响应到来。while 循环做的工作非常简单对于每个响应消息它都会检查CorrelationId是否是我们正在寻找的那一个。如果是这样它就会保存该响应。最后我们将响应返回给用户。客户发出请求var rpcClient new RPCClient(); Console.WriteLine( [x] Requesting fib(30)); var response rpcClient.Call(30); Console.WriteLine( [.] Got {0}, response); rpcClient.Close();现在是查看 RPCClient.cs 和 RPCServer.cs 的完整示例源代码包括基本异常处理的好时机哦。像往常一样设置请参见 教程[1]]我们的 RPC 服务现已准备就绪现在可以启动服务端cd RPCServer dotnet run# [x] Awaiting RPC requests要请求斐波纳契数请运行客户端cd RPCClient dotnet run# [x] Requesting fib(30)这里介绍的设计并不是 RPC 服务的唯一可能实现但它仍具有一些重要优势如果 RPC 服务器太慢您可以通过运行另一个服务器来扩展。尝试在新开一个控制台运行第二个 RPCServer。在客户端RPC 只需要发送和接收一条消息。不需要像QueueDeclare一样同步调用。因此对于单个 RPC 请求RPC 客户端只需要一次网络往返。我们的代码很简单也并没有尝试去解决更复杂但很重要的问题比如就像如果服务端没有运行客户端应该如何反应客户端是否应该为 RPC 设置某种超时机制如果服务端出现故障并引发异常是否应将其转发给客户端在处理之前防止无效的传入消息例如检查边界、类型。如果您想进行实验您可能会发现 管理 UI 对于查看队列非常有用。写在最后本文翻译自 RabbitMQ 官方教程 C# 版本。如本文介绍内容与官方有所出入请以官方最新内容为准。水平有限翻译的不好请见谅如有翻译错误还请指正。原文链接RabbitMQ tutorial - Remote procedure call (RPC)实验环境RabbitMQ 3.7.4 、.NET Core 2.1.3、Visual Studio Code最后更新2018-11-17相关文章分布式系统消息中间件——RabbitMQ的使用基础篇ASP.NET Core 2.0利用MassTransit集成RabbitMQ浅谈surging服务引擎中的rabbitmq组件和容器化部署RabbitMQ一个简单可靠的方案(.Net Core实现).NetCore Cap 结合 RabbitMQ 实现消息订阅[译]RabbitMQ教程C#版 - 发布订阅RabbitMQ教程C#版 - 工作队列RabbitMQ教程C#版 “Hello World”原文地址: https://www.cnblogs.com/esofar/p/rabbitmq-rpc.html.NET社区新闻深度好文欢迎访问公众号文章汇总 http://www.csharpkit.com
