rebbitmq-RPC(C#)

RPC（Remote Procedure Call Protocol）——远程过程调用协议

运行时,一次客户机对服务器的RPC调用,其内部操作大致有如下十步：

1.调用客户端句柄；执行传送参数

2.调用本地系统内核发送网络消息

3.消息传送到远程主机

4.服务器句柄得到消息并取得参数

5.执行远程过程

 

6.执行的过程将结果返回服务器句柄

7.服务器句柄返回结果，调用远程系统内核

8.消息传回本地主机

9.客户句柄由内核接收消息

10.客户接收句柄返回的数据

 

rebbitmq通过消息交互来实现远程方法的调用，

主要过程包括两端：

被调用端：

            //创建链接
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                //创建渠道
                using (var channel = connection.CreateModel())
                {
                    //创建消息队列，接收rpc请求
                    channel.QueueDeclare(queue: "rpc_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);

                    //监听并消费消息，收到消息需要通知发送方
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume(queue: "rpc_queue", noAck: false, consumer: consumer);
                    while (true)
                    {
                        Thread.Sleep(1000);
                        Console.WriteLine(string.Format("等待RPC请求..."));
                        string response = null;
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
                        var body = ea.Body;
                        var props = ea.BasicProperties;
                        var replyProps = channel.CreateBasicProperties();
                        //约定ID
                        replyProps.CorrelationId = props.CorrelationId;
                        var message = Encoding.UTF8.GetString(body);
                        try
                        {
                            Console.WriteLine(string.Format("执行方法{0}", message));
                            response = Request(message);
                        }
                        catch (Exception ex)
                        {
                            Console.WriteLine(string.Format("执行{0}异常，异常信息：{1}", message, ex.Message));
                            response = "";
                        }
                        finally
                        {
                            var responseBytes = Encoding.UTF8.GetBytes(response);
                            //按照发送方的要求（等待返回的接收队列、约定ID），返回消息
                            channel.BasicPublish(exchange: "", routingKey: props.ReplyTo, basicProperties: replyProps, body: responseBytes);
                            //收到消息需要通知发送方
                            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                        }
                    }
                }
            }

调用端：

            //创建链接
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                //创建渠道
                using (var channel = connection.CreateModel())
                {
                    //创建接收返回消息的队列
                    string replyQueueName = channel.QueueDeclare().QueueName;

                    //监听并消费消息，收到消息无需通知发送方
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume(queue: replyQueueName, noAck: true, consumer: consumer);

                    var props = channel.CreateBasicProperties();
                    props.ReplyTo = replyQueueName;
                    var corrId= Guid.NewGuid().ToString();
                    props.CorrelationId =corrId;

                    var messageByte = Encoding.UTF8.GetBytes(message);
                    //发送消息，并携带接收返回消息的相关内容（接收队列、约定ID）
                    channel.BasicPublish(exchange: "", routingKey: "rpc_queue", basicProperties: props, body: messageByte);

                    while (true)
                    {
                        Thread.Sleep(1000);
                        Console.WriteLine(string.Format("等待请求requst:{0}返回...", message));
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
                        if (ea.BasicProperties.CorrelationId==corrId)
                        {
                            Console.WriteLine(string.Format("请求requst:{0};response:{1}", message, Encoding.UTF8.GetString(ea.Body)));
                            break;
                        }
                    }
                }
            }