RabbitMQ(十)——消息優先級

RabbitMQ系列

RabbitMQ(一)——簡介

RabbitMQ(二)——模式類型

RabbitMQ(三)——簡單模式

RabbitMQ(四)——工作隊列模式

RabbitMQ(五)——發布訂閱模式

RabbitMQ(六)——路由模式

RabbitMQ(七)——主題模式

RabbitMQ(八)——消息確認

RabbitMQ(九)——消息持久化

RabbitMQ(十)——消息優先級

 

前言

　　通過前面的學習我們將消息寫入隊列中都是按順序寫的，消費時也是按順序進行消費，即隊列queue是先進先出的(FIFO)，在一些場景中我們需要將某些消息提前處理，首先定義隊列時先設置隊列為優先隊列，然后設置優先等級。

channel.QueueDeclare("Priqueue", true, false, false, new Dictionary<string, object>() { { "x-max-priority",10} });

 

生產者：

        //初始化工廠
            ConnectionFactory factory = new ConnectionFactory()
            {
                HostName = "127.0.0.1",
                UserName = "guest",
                Password = "guest"
            };
            //2.創建連接
            using (var connection = factory.CreateConnection())
            //3.創建管道
            using (var channel = connection.CreateModel())
            {
                //4.創建交換器
                channel.ExchangeDeclare("exchange", "fanout", true);
                //定義隊列，設置優先級
                channel.QueueDeclare("Priqueue", true, false, false, new Dictionary<string, object>() { { "x-max-priority",10} });
                //交換器綁定隊列
                channel.QueueBind("Priqueue", "exchange", "", null);

                //消息持久化
                IBasicProperties basicProperties = channel.CreateBasicProperties();
                basicProperties.Persistent = true;

                string msg = "";

                for (int i = 0; i < 10; i++)
                {
                    msg = $"發布消息{i}";
                    if (i % 2 == 0)
                    {
                        msg += "——vip用戶消息(優先處理)";
                        basicProperties.Priority = 9;
                    }
                    else
                    {
                        msg += "——普通用戶消息";
                        basicProperties.Priority = 1;
                    }
                    var body = Encoding.UTF8.GetBytes(msg);
                    channel.BasicPublish("exchange", "", basicProperties, body);
                    Console.WriteLine($"發布成功：{msg}");
                    Thread.Sleep(1000);
                }
                Console.ReadKey();
            }    

 

消費者：

//1.創建連接工廠
            ConnectionFactory factory = new ConnectionFactory()
            {
                HostName = "127.0.0.1",
                UserName = "guest",
                Password = "guest"
            };
            //2.創建連接
            using (var connection = factory.CreateConnection())
            {
                //3.創建通道
                using (var channel = connection.CreateModel())
                {
                    //4.定義交換器  持久化
                    channel.ExchangeDeclare("exchange", "fanout",true);
                    //5.創建匿名隊列，綁定交換器  durable:true 持久化 沒有被消費不會丟失
                    var queueName = channel.QueueDeclare("Priqueue", true, false, false, new Dictionary<string, object>() { { "x-max-priority", 10 } });
                    channel.QueueBind(queueName, "exchange", "",null);

                    //6.創建消費者
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                      {
                          //接收消息
                          var body = Encoding.UTF8.GetString(ea.Body.ToArray());
                          Console.WriteLine($"接收消息：{body.ToString()}");
                      };
                    //7.消費消息
                    channel.BasicConsume(queueName, true, consumer);
                    Console.ReadKey();
                }
            }

 

總結：

　　開始生產消息，生產完成后啟動消費者控制台，出現以下的效果，消費者會優先消費優先級別高的消息，也就是生產者中定義的vip消息，消費完優先級高的vip消息才去消費優先級低的普通用戶的消息。