RabbitMQ学习路由模式(以C#代码为例)

路由模式（Routing）

有选择地（Routing key）接收消息，发送消息到交换机并且要指定路由key ，消费者将队列绑定到交换机时需要指定路由key，仅消费指定路由key的消息

Exchange不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey进行判断，只有队列的RoutingKey与消息的RoutingKey完全一致，才会接收消息，路由模式使用的Exchange类型为Direct类型

应用场景： 如在商品库存中增加了1台iphone12，iphone12促销活动消费者指定routing key为iphone12，只有此促销活动会接收到消息，其它促销活动不关心也不会消费此routing key的消息

 Demo 架构与订阅模式一样

生产者类WeatherDirect.cs

using System;
using System.Collections.Generic;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Common;

namespace RabbitMQ.Producer.Producer
{
    public class WeatherDirect
    {
        public static void SendWeather()
        {
            Dictionary<string, string> area = new Dictionary<string, string>();
            area.Add("china.hunan.changsha.20211231", "中国湖南长沙20211231天气数据");
            area.Add("china.hubei.wuhan.20211231", "中国湖北武汉20211231天气数据");
            area.Add("china.hubei.xiangyang.20211231", "中国湖北襄阳20211231天气数据");
            area.Add("us.cal.lsj.20211231", "美国加州洛杉矶20211231天气数据");

            using (var connection = RabbitUtils.GetConnection().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    foreach (var item in area)
                    {
                        channel.BasicPublish(RabbitConstant.EXCHANGE_WEATHER_ROUTING, item.Key, null, Encoding.UTF8.GetBytes(item.Value));
                    }
                    Console.WriteLine("气象信息发送成功！");
                    Console.WriteLine("Press [Enter] to exit");
                    Console.ReadLine();
                }
            }
        }
    }
}

消费者1 WeatherDirect.cs

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Common;
using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQ.Consumer01.Consumer
{
    public class WeatherDirect
    {
        public static void ReceiveWeatherInfo()
        {
            using (var connection = RabbitUtils.GetConnection().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(RabbitConstant.EXCHANGE_WEATHER_ROUTING, ExchangeType.Direct);
                    channel.QueueDeclare(RabbitConstant.QUEUE_BAIDU, true, false, false, null);
                    /*
                     * QueueBind用于将队列与交换机绑定
                     * queue:队列名
                     * exchange:交换机名
                     * routingKey：路由key
                     */
                    channel.QueueBind(queue: RabbitConstant.QUEUE_BAIDU,
                                      exchange: RabbitConstant.EXCHANGE_WEATHER_ROUTING,
                                      routingKey: "china.hunan.changsha.20211231");
                    channel.QueueBind(queue: RabbitConstant.QUEUE_BAIDU,
                                    exchange: RabbitConstant.EXCHANGE_WEATHER_ROUTING,
                                    routingKey: "china.hubei.wuhan.20211231");
                    channel.BasicQos(0, 1, false);
                    var consumer = new EventingBasicConsumer(channel);

                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"百度收到的气象信息：{message}");
                        channel.BasicAck(ea.DeliveryTag, false);
                    };
                    channel.BasicConsume(queue: RabbitConstant.QUEUE_BAIDU,
                                 autoAck: false,
                                 consumer: consumer);
                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }
    }
}

消费者2  WeatherDirect.cs

using System;
using System.Collections.Generic;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Common;

namespace RabbitMQ.Consumer02.Consumer
{
    public class WeatherDirect
    {
        public static void ReceiveWeatherInfo()
        {
            using (var connection = RabbitUtils.GetConnection().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(RabbitConstant.EXCHANGE_WEATHER_ROUTING, ExchangeType.Direct);
                    channel.QueueDeclare(RabbitConstant.QUEUE_SINA, true, false, false, null);
                    /*
                     * QueueBind用于将队列与交换机绑定
                     * queue:队列名
                     * exchange:交换机名
                     * routingKey：路由key
                     */
                    channel.QueueBind(queue: RabbitConstant.QUEUE_SINA,
                                      exchange: RabbitConstant.EXCHANGE_WEATHER_ROUTING,
                                      routingKey: "china.hubei.xiangyang.20211231");
                    channel.QueueBind(queue: RabbitConstant.QUEUE_SINA,
                                    exchange: RabbitConstant.EXCHANGE_WEATHER_ROUTING,
                                    routingKey: "china.hubei.wuhan.20211231");
                    channel.QueueBind(queue: RabbitConstant.QUEUE_SINA,
                                    exchange: RabbitConstant.EXCHANGE_WEATHER_ROUTING,
                                    routingKey: "us.cal.lsj.20211231");
                    channel.BasicQos(0, 1, false);
                    var consumer = new EventingBasicConsumer(channel);

                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"新浪收到的气象信息：{message}");
                        channel.BasicAck(ea.DeliveryTag, false);
                    };
                    channel.BasicConsume(queue: RabbitConstant.QUEUE_SINA,
                                 autoAck: false,
                                 consumer: consumer);
                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }
    }
}

执行结果

 备注：若执行报错：'PRECONDITION_FAILED - inequivalent arg 'type' for exchange 

原因：一旦创建了exchange, RabbitMQ是不允许对其改变的,不然会报错

解决方法：在web上Exchanges模块把之前的exchange删除。

 

参考连接：https://www.bilibili.com/video/BV1GU4y1w7Yq?p=8

                  https://mp.weixin.qq.com/s/QG3uXhhpkE_Uo6Me15mxdg

                 https://blog.csdn.net/adviceuser2014/article/details/102259145