C#队列学习笔记：RabbitMQ安装及使用(3)

当前位置:

首页 > Python基础教程 >

C#队列学习笔记：RabbitMQ安装及使用(3)

    class Program
    {
        static void Main(string[] args)
        {
            #region rpc
            //1.实例化连接工厂
            var factory = new ConnectionFactory
            {
                HostName = "localhost",
                UserName = "hello",
                Password = "world"
            };
            //2.建立连接
            using (var connection = factory.CreateConnection())
            {
                //3.建立信道
                using (var channel = connection.CreateModel())
                {
                    //4.声明唯一guid用来标识此次发送的远程调用请求
                    var correlationId = Guid.NewGuid().ToString();
                    //5.声明需要监听的回调队列
                    var replyQueue = channel.QueueDeclare().QueueName;
                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;//将消息标记为持久性
                    properties.ReplyTo = replyQueue;//指定回调队列
                    properties.CorrelationId = correlationId;//指定消息唯一标识
                    //6.构建byte消息数据包
                    string number = args.Length > 0 ? args[0] : "30";
                    var body = Encoding.UTF8.GetBytes(number);
                    //7.发送数据包
                    channel.BasicPublish(exchange: "", routingKey: "rpc_queue", basicProperties: properties, body: body);
                    Console.WriteLine($"Request fib({number})");
                    //8.创建消费者用于处理消息回调（远程调用返回结果）
                    var callbackConsumer = new EventingBasicConsumer(channel);
                    channel.BasicConsume(queue: replyQueue, noAck: false, consumer: callbackConsumer);
                    callbackConsumer.Received += (model, ea) =>
                    {
                        //仅当消息回调的ID与发送的ID一致时，说明远程调用结果正确返回。
                        if (ea.BasicProperties.CorrelationId == correlationId)
                        {
                            var responseMsg = $"Get Response: {Encoding.UTF8.GetString(ea.Body)}";
                            Console.WriteLine($"{responseMsg}");
                        }
                    };
                    Console.Read();
                }
            }
            #endregion
        }
    }

			
    class Program
    {
        static void Main(string[] args)
        {
            #region rpc
            //1.实例化连接工厂
            var factory = new ConnectionFactory
            {
                HostName = "localhost",
                UserName = "hello",
                Password = "world"
            };
            //2.建立连接
            using (var connection = factory.CreateConnection())
            {
                //3.创建信道
                using (var channel = connection.CreateModel())
                {
                    //4.声明队列接收远程调用请求
                    channel.QueueDeclare(queue: "rpc_queue", durable: true, exclusive: false, autoDelete: false, arguments: null);
                    var consumer = new EventingBasicConsumer(channel);
                    Console.WriteLine("Waiting for message.");
                    //5.请求处理逻辑
                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body);
                        int n = int.Parse(message);
                        Console.WriteLine($"Receive request of Fib({n})");
                        int result = Fib(n);
                        //6.从请求的参数中获取请求的唯一标识，在消息回传时同样绑定。
                        var properties = ea.BasicProperties;
                        var replyProerties = channel.CreateBasicProperties();
                        replyProerties.CorrelationId = properties.CorrelationId;
                        //7.将远程调用结果发送到客户端监听的队列上
                        channel.BasicPublish(exchange: "", routingKey: properties.ReplyTo,
                            basicProperties: replyProerties, body: Encoding.UTF8.GetBytes(result.ToString()));
                        //8.手动发回消息确认
                        channel.BasicAck(ea.DeliveryTag, false);
                        Console.WriteLine($"Return result: Fib({n})= {result}");
                    };
                    channel.BasicConsume(queue: "rpc_queue", noAck: false, consumer: consumer);
                    Console.Read();
                }
            }

            int Fib (int n)
            {
                if (n <= 2)
                    return 1;
                else
                    return Fib(n - 1) + Fib(n - 2);
            }
            #endregion
        }
    }
			

六、总结

本文介绍了RabbitMQ消息代理在Windows上的安装以及在.NET中的使用。消息队列在构建分布式系统、提高系统的可扩展性及响应性方面，有着很重要的作用。

参考自：

https://www.cnblogs.com/yangecnu/p/Introduce-RabbitMQ.html#!comments

https://www.cnblogs.com/sheng-jie/p/7192690.html

栏目列表