点击刷新
Remote Procedure Call （RPC）
用RabbitMQ实现RPC还是比较简单，一个客户端发送一个请求，服务端返回一个响应。为了实现响应我们需要发送一个"callback" 指定某个请求的队列，也就是下面主要介绍的Correlation Id。
Correlation Id
实现RPC最简单的方式是为每个RPC请求创建一个callback回调函数，但这样是无效的或者说是冗余繁杂的。
这也会产生一个问题，就是返回的response不能清晰的知道属于哪个request，这就会用到CorrelationId，我们将会为每次请求创建唯一的值，然后，我们收到这个callback response，查找这个值通过匹配的方式能够知道response属于哪个RPC request，如果我们收到一个未知的CorrelationId 值，我们完全可以忽略它，不属于我们的请求。
这里可能会有所迷惑，为什么我们要忽略这个未知的值而不是直接报错，这个是由于RPC Server的竞争，虽然不太可能，这个可能是当RPC server发送一个reponse，但未发送确认（acknowledgment）消息给request之前宕机了，如果发生了，RPC Server重启之后会再次发送请求。这就是为什么在client端需要处理duplicate response。
PRC工作原理：
- 当Clientk开始，将会创建匿名唯一的callback队列
- 对于RPC请求，client会发送一个消息附带2个属性：ReplyTo通常是发送callback请求队列名，CorrelationId指定每次请求的唯一值。
- 请求发送到队列
- RPC Server等待请求对于指定的queue，当接收到请求，处理job并且会发送一个response消息给client，response的队列是ReplyTo request的队列。（就是说request和response的queue name和CorrelationId是一样的）
- Client等待callback queue的数据，当收到请求消息，判断CorrelcationId是否匹配，如果匹配了，返回response给application 应用程序。
下面以Fibonacci数列为例，实现简单的RPC Server / Client
RPC Server的逻辑比较简单：
- 建立连接connection，channel，声明queue
- 我们可能会运行多余一个的RPC Server，为了在多个Server之间实现负载均衡，需要设置prefetchCount 在channel.BasicQos settings。（解释一下：prefetchCount = 1是告诉RabbitMQ不要同时给一个worker分发多个消息，或者换句话说不要分发一个新消息给这个worker直到worker之前的消息已经处理并且被确认。而是分发新消息给下一个不繁忙的worker）
- 用BasiConsume访问queue，然后注册一个delivery handler处理和发送response back。
示例Code：
public static void RunRPCDemo() { var factory = new ConnectionFactory() { HostName = "10.2.87.39" }; 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] waiting RPC request"); consumer.Received += (model, ea) => { string response = null; var body = ea.Body.ToArray(); 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{message}"); response = fib(n).ToString(); } catch (Exception e) { Console.WriteLine($"[.] {e.Message}"); } finally { var reponseBytes = Encoding.UTF8.GetBytes(response); channel.BasicPublish(exchange: "", routingKey: props.ReplyTo, basicProperties: replyProps, body: reponseBytes); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); } }; Console.WriteLine("Press any key to exist"); Console.ReadLine(); } } } private static int fib(int n) { if (n == 0 || n == 1) { return n; } return fib(n - 1) + fib(n - 2); }
RPC Client 逻辑稍微复杂一些：
- 建立connection和channel，声明唯一的callback queue
- 订阅callback queue，以至于可以收到RPC 响应
- Call 方法是实际RPC 请求
- 创建一个唯一的CorrelationId number
- 发布request message附带ReplyTo和CorrelationId两个属性
- 接下来等待RPC Server响应
- 对于每个响应，判断是否是我们所期待的
- 最终返回response给user
示例Code
public class RPCClient { private readonly IConnection connection; private readonly IModel channel; private readonly string replyQueueName; private readonly EventingBasicConsumer consumer; private readonly BlockingCollection<string> respQueue = new BlockingCollection<string>(); private readonly IBasicProperties props; public RPCClient() { var factory = new ConnectionFactory() { HostName = "10.2.87.39" }; connection = factory.CreateConnection(); channel = connection.CreateModel(); replyQueueName = channel.QueueDeclare().QueueName; consumer = new EventingBasicConsumer(channel); props = channel.CreateBasicProperties(); var correlationId = $"{Guid.NewGuid()}"; props.CorrelationId = correlationId; props.ReplyTo = replyQueueName; consumer.Received += (model, ea) => { var body = ea.Body.ToArray(); 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(); } }
客户端Code:
public static void RunRPCDemo() { var rpcClient = new RPCClient(); Console.WriteLine($"[x] requesting fib(30)"); var response = rpcClient.Call("30"); Console.WriteLine($"[.] get {response}"); rpcClient.Close(); }
运行RPC Server和Client，结果如下：
简单的RPC已经实现了。
待解决问题：
- 没有server运行时，Client如何处理
- RPC Client长时间没有收到response，过期时间怎么处理
- 如果server 异常了，是否需要向client发送response
- 无效的incoming message，如何处理
© 著作权归作者所有
举报
相关热门文章
发表评论
0/200
