RabbitMQ学习通配符模式(以C#代码为例)

通配符模式（Topics）

 

根据通配符（Topics）来接收消息，将路由key和某模式进行匹配，此时队列需要绑定在一个模式上，#匹配一个词或多个词，*只匹配一个词。

使用的Exchange类型为Topic

Topic类型的Exchange：

（1）Topic类型与Direct相比，都是可以根据RoutingKey将消息路由到不同的队列，只不过Topic类型可以让队列在绑定路由时使用通配符

（2）Topic类型的RoutingKey一般是由一个或多个单词构成，多个单词之间用“.”分割，如item.insert

（3）Topic类型通配符规则：#匹配一个或多个词，*恰好匹配一个词，例如item.#，可以匹配item.insert或item.insert.user，item.*只能匹配

         item.insert或item.user

Demo 

生产者类WeatherTopic.cs

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

namespace RabbitMQ.Producer.Producer
{
    public class WeatherTopic
    {
        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_TOPIC, item.Key, null, Encoding.UTF8.GetBytes(item.Value));
                    }
                    Console.WriteLine("气象信息发送成功！");
                    Console.WriteLine("Press [Enter] to exit");
                    Console.ReadLine();
                }
            }
        }
    }
}

消费者01 WeatherTopic.cs

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

namespace RabbitMQ.Consumer01.Consumer
{
    public class WeatherTopic
    {
        public static void ReceiveWeatherInfo()
        {
            using (var connection = RabbitUtils.GetConnection().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(RabbitConstant.EXCHANGE_WEATHER_TOPIC, ExchangeType.Topic);
                    channel.QueueDeclare(RabbitConstant.QUEUE_BAIDU, true, false, false, null);
                    /*
                     * QueueBind用于将队列与交换机绑定
                     * queue:队列名
                     * exchange:交换机名
                     * routingKey：路由key
                     */
                    channel.QueueBind(queue: RabbitConstant.QUEUE_BAIDU,
                                      exchange: RabbitConstant.EXCHANGE_WEATHER_TOPIC,
                                      routingKey: "china.hubei.*.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();
                }
            }
        }
    }
}

消费者02 WeatherTopic.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 WeatherTopic
    {
        public static void ReceiveWeatherInfo()
        {
            using (var connection = RabbitUtils.GetConnection().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(RabbitConstant.EXCHANGE_WEATHER_TOPIC, ExchangeType.Topic);
                    channel.QueueDeclare(RabbitConstant.QUEUE_SINA, true, false, false, null);
                    /*
                     * QueueBind用于将队列与交换机绑定
                     * queue:队列名
                     * exchange:交换机名
                     * routingKey：路由key
                     */
                    channel.QueueBind(queue: RabbitConstant.QUEUE_SINA,
                                      exchange: RabbitConstant.EXCHANGE_WEATHER_TOPIC,
                                      routingKey: "china.#");
                    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();
                }
            }
        }

    }
}

执行结果

  

运行前配置：删除原来的Exchange,具体操作如下

 

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

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