RabbitMQ (十二) 远程过程调用（RPC）

    internal class Program
    {
        private const string RequestQueueName = "rpc_queue";

        private static void Main(string[] args)
        {
            using (RabbitMQ.Client.IConnection connection = ConnectionHelper.GetConnection())
            using (RabbitMQ.Client.IModel channel = connection.CreateModel())
            {
                channel.QueueDeclare(queue: RequestQueueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
                channel.BasicQos(0, 1, false);

                EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                channel.BasicConsume(queue: RequestQueueName, autoAck: false, consumerTag: "", noLocal: false, exclusive: false, arguments: null, consumer: consumer);
                Console.WriteLine("server 开始等待 RPC 请求");

                consumer.Received += (s, e) =>
                {
                    string response = null;
                    byte[] bytes = e.Body;
                    RabbitMQ.Client.IBasicProperties pros = e.BasicProperties;//拿到这条请求消息的属性
                    RabbitMQ.Client.IBasicProperties replyPros = channel.CreateBasicProperties();//创建响应消息的属性
                    replyPros.CorrelationId = pros.CorrelationId;//将请求消息的id赋值给响应消息,这个id就相当于请求消息的身份证

                    try
                    {
                        string msg = Encoding.UTF8.GetString(bytes);
                        int n = int.Parse(msg);
                        Console.WriteLine($"执行函数 Fib(int n) , 入参为 {msg}");
                        response = Fib(n).ToString();//运行函数,拿到结果
                    }
                    catch (Exception exception)
                    {
                        Console.WriteLine(exception);
                        response = string.Empty;
                    }
                    finally
                    {
                        byte[] responseBytes = Encoding.UTF8.GetBytes(response);//创建响应消息的字节码
                        //将响应消息发送到请求消息的属性中指定的响应队列
                        channel.BasicPublish(exchange: "", routingKey: pros.ReplyTo, mandatory: false, basicProperties: replyPros, body: responseBytes);

                        //发送响应消息后,手动确认已经收到请求消息
                        channel.BasicAck(deliveryTag: e.DeliveryTag, multiple: false);
                    }
                };

                Console.WriteLine("按 enter 退出");
                Console.ReadLine();
            }
        }

        /// <summary>
        /// 服务器的函数
        /// </summary>
        /// <param name="n"></param>
        /// <returns></returns>
        private static int Fib(int n)
        {
            if (n == 0 || n == 1)
            {
                return n;
            }
            return Fib(n - 1) + Fib(n - 2);
        }
    }

    public class MyClient
    {
        private readonly IConnection connection;
        private readonly IModel channel;
        private readonly IBasicProperties pros;//请求消息属性
        private readonly EventingBasicConsumer consumer;
        private readonly string replyQueueName;//响应队列名称
        private const string requestQueueName = "rpc_queue";//请求队列名称
        private readonly BlockingCollection<string> responseQueue = new BlockingCollection<string>();//存储响应消息

        public MyClient()
        {
            connection = ConnectionHelper.GetConnection();
            channel = connection.CreateModel();
            replyQueueName = channel.QueueDeclare().QueueName;//声明一个随机的,独占的,自动删除的,非持久化的响应队列
            consumer = new EventingBasicConsumer(channel);//创建一个消费者
            pros = channel.CreateBasicProperties();
            string correlationId = Guid.NewGuid().ToString();//创建一个"身份证"
            pros.CorrelationId = correlationId;
            pros.ReplyTo = replyQueueName;//设置回调队列

            consumer.Received += (s, e) =>
            {
                string response = Encoding.UTF8.GetString(e.Body);//拿到响应消息
                if (e.BasicProperties.CorrelationId.Equals(correlationId))//确认身份
                {
                    responseQueue.Add(response);
                }
            };
        }


        /// <summary>
        /// 发起请求
        /// </summary>
        /// <param name="msg">请求消息</param>
        /// <returns>请求的结果</returns>
        public string Call(string msg)
        {
            byte[] bytes = Encoding.UTF8.GetBytes(msg);
            channel.BasicPublish(exchange: "", routingKey: requestQueueName, basicProperties: pros, body: bytes);//向请求队列发送请求消息.

            //在发送请求消息(发起请求)后,再定义客户端需要消费的回复队列,并且设置应答模式为 自动应答.因为RPC中,服务器不用关心客户端是否收到了响应
            channel.BasicConsume(queue: replyQueueName, autoAck: true, consumer: consumer);

            return responseQueue.Take();//返回本次请求的结果
        }


        /// <summary>
        /// 关闭客户端
        /// </summary>
        public void Close()
        {
            channel.Close();
            connection.Close();
        }
    }


    internal class Program
    {
        private static void Main(string[] args)
        {
            MyClient client = new MyClient();
            while (true)
            {
                Console.WriteLine("请输入您要发送的请求消息 : ");
                string request = Console.ReadLine();
                if (string.IsNullOrWhiteSpace(request))
                {
                    continue;
                }
                if (request.ToLower().Equals("q"))
                {
                    break;
                }
                string response = client.Call(request);
                Console.WriteLine("请求的结果 : " + response);
            }
            client.Close();
        }
    }