其他分享
首页 > 其他分享> > NetCore RabbitMQ Topics 通配符模式

NetCore RabbitMQ Topics 通配符模式

作者:互联网

十年河东,十年河西,莫欺少年穷

学无止境,精益求精

上一节介绍了RabbitMQ定向模式,本篇介绍 Topics 通配符模式

我的系列博客:

NetCore RabbitMQ ,Routing定向模式

NetCore RabbitMQ 发布订阅模式,消息广播

RabbitMQ的六种工作模式

NetCore RabbitMQ 简介及兔子生产者、消费者 【简单模式,work工作模式,竞争消费】

windows环境下,RabbitMQ 安装教程

kafka、Rabbitmq、EasyNetQ NetCore 源码下载 仅自己可见

本篇介绍Topics通配符模式

1.5 topic 主题模式(路由模式的一种)

在这里插入图片描述

  1. 星号井号代表通配符
  2. 星号1个字符井号代表0个或多个字符
  3. 路由功能添加模糊匹配
  4. 消息产生者产生消息,把消息交给交换机
  5. 交换机根据key的规则模糊匹配到对应的队列,由队列的监听消费者接收消息消费

Topics通配符模式和定向模式类似,都是通过RoutingKey定向到不到的队列中,不同的是Topics通配符模式只是通配符,有了通配符,就可以做到模糊匹配。

Topics模式中支持两种通配符

* 代表1个字符

# 代表0个或多个字符

例如

路由Key定义为 *.Dog.* ,那么我们发送消息时中间为Dog且前后均为一个字符的路由key都可以匹配到该通配符,例如:C.Dog.D 【注意要带上 . 】

路由Key定义为 #.Dog ,那么发送消息时以Dog结尾的路由Key都可以匹配到该通配符,例如 chenwolong.Dog【注意要带上 . 】

路由Key定义为 Dog.# ,那么发送消息时以Dog开头的路由Key都可以匹配到该通配符 ,例如 Dog.Chenwolong【注意要带上 . 】

路由Key定义为 #.#  ,那么 Dog.Cat,Cat.Dog  . 等都可以匹配

下面由代码进行说明

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

namespace RabbitMqProducer
{
    class Program
    {
        /// <summary>
        /// 本示例红色预警为最高预警级别,蓝色预警次之,黄色预警等级最差
        /// 红色预警队列中只展示红色预警消息,
        /// 蓝色预警队列中展示红色预警和蓝色预警 
        /// 黄色预警队列展示红、蓝、黄三种预警
        /// </summary>
        /// <param name="args"></param>
        static void Main(string[] args)
        {
            ConnectionFactory factory = new ConnectionFactory();
            factory.HostName = "127.0.0.1"; //主机名
            factory.UserName = "guest";//使用的用户
            factory.Password = "guest";//用户密码
            factory.Port = 5672;//端口号
            factory.VirtualHost = "/"; //虚拟主机
            factory.MaxMessageSize = 1024; //消息最大字节数
            using (var connection = factory.CreateConnection()) 
            {
                //rabbitMQ 基于信道进行通信,因此,我们需要实例化信道Channel
                using (var channel = connection.CreateModel())
                {
                    //exchange 交换机名称
                    //type  交换机类型  ExchangeType.Direct【定向模式】  ExchangeType.Fanout【广播模式】  ExchangeType.Topic【通配符模式】  ExchangeType.Headers 【参数匹配模式】
                    //durable 是否持久化
                    //autoDelete 队列是否为临时队列
                    //arguments  其他配置 详见博客:https://www.cnblogs.com/chenwolong/p/RabbitMQ_S.html
                    //void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments);
                    //声明一个交换机  类型 : Direct
                    string Ename = "ExRabbitMQ_Topic";
                    channel.ExchangeDeclare(Ename, ExchangeType.Topic, false, false, null);
                    //声明广播的队列
                    string Qname_Red = "Queue_Red";
                    string Qname_Blue = "Queue_Blue"; 
                    channel.QueueDeclare(Qname_Red, false, false, false, null);
                    channel.QueueDeclare(Qname_Blue, false, false, false, null); 
                    //交换机 队列 绑定
                    //queue 队列名称
                    //exchange  交换机名称
                    //routingKey  路由规则  
                    //void QueueBind(string queue, string exchange, string routingKey, IDictionary<string, object> arguments);
                    string routingKey_Red = "Red.#"; // 红色预警
                    string routingKey_Blue = "*.Blue"; // 蓝色预警 
                    // 
                    channel.QueueBind(Qname_Red, Ename, routingKey_Red);
                    // 
                    channel.QueueBind(Qname_Blue, Ename, routingKey_Blue);
                 
                    //发送消息
                    for(int i = 0; i < 10 ; i++)
                    {
                        var messages_HotRed = "#号匹配0个或多个字符,我是高温红色预警(Red.Hot)"; //传递的消息内容
                        var messages_FloodRed = "#号匹配0个或多个字符,我是洪涝红色预警(Red.Floot)"; //传递的消息内容
                        var messages_Blue = "*号匹配一个字符,我是蓝色预警(A.Blue)"; //传递的消息内容 
                                                                    //exchange 交换机,如果使用默认的交换机,那么routingKey要和队列的名称一致
                                                                    //routingKey:路由  
                                                                    //basicProperties : 用于基础属性设置
                        ///BasicPublish(this IModel model, string exchange, string routingKey, IBasicProperties basicProperties, ReadOnlyMemory<byte> body);
                        //注意发送消息时,路由Key的匹配规则
                        channel.BasicPublish(Ename, "Red.Hot", null, Encoding.UTF8.GetBytes(messages_HotRed + "_" + i)); //生产高温红色预警消息
                        channel.BasicPublish(Ename, "Red.Flood", null, Encoding.UTF8.GetBytes(messages_FloodRed + "_" + i)); //生产洪涝红色预警消息
                        channel.BasicPublish(Ename, "A.Blue", null, Encoding.UTF8.GetBytes(messages_Blue + "_" + i)); //生产蓝色预警消息
                    }
                }
            }
            Console.Read();
        }
    }
}
View Code

蓝色预警消费者端如下:

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

namespace RabbitMqConsumer
{
    class Program
    {
        static void Main(string[] args)
        {
            ConnectionFactory factory = new ConnectionFactory();
            factory.HostName = "127.0.0.1"; //主机名
            factory.UserName = "guest";//使用的用户
            factory.Password = "guest";//用户密码
            factory.Port = 5672;//端口号
            factory.VirtualHost = "/"; //虚拟主机
            factory.MaxMessageSize = 1024; //消息最大字节数
                                           //创建连接
            var connection = factory.CreateConnection();
            //创建通道
            var channel = connection.CreateModel();

            //事件基本消费者
            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

            //接收到消息事件
            consumer.Received += (ch, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                Console.WriteLine($"蓝色预警队列消费者收到消息: {message}");
                //确认该消息已被消费
                channel.BasicAck(ea.DeliveryTag, false); Thread.Sleep(100);
            };
            //启动消费者 
            string Qname = "Queue_Blue";
            channel.BasicConsume(Qname, false, consumer);
            Console.WriteLine("消费者已启动");
            Console.ReadKey();
            channel.Dispose();
            connection.Close();
        } 
    }
}
View Code

红色预警消费者端如下:

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

namespace RabbitMqConsumer
{
    class Program
    {
        static void Main(string[] args)
        {
            ConnectionFactory factory = new ConnectionFactory();
            factory.HostName = "127.0.0.1"; //主机名
            factory.UserName = "guest";//使用的用户
            factory.Password = "guest";//用户密码
            factory.Port = 5672;//端口号
            factory.VirtualHost = "/"; //虚拟主机
            factory.MaxMessageSize = 1024; //消息最大字节数
                                           //创建连接
            var connection = factory.CreateConnection();
            //创建通道
            var channel = connection.CreateModel();

            //事件基本消费者
            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

            //接收到消息事件
            consumer.Received += (ch, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                Console.WriteLine($"红色预警队列消费者收到消息: {message}");
                //确认该消息已被消费
                channel.BasicAck(ea.DeliveryTag, false); Thread.Sleep(100);
            };
            //启动消费者 
            string Qname = "Queue_Red";
            channel.BasicConsume(Qname, false, consumer);
            Console.WriteLine("消费者已启动");
            Console.ReadKey();
            channel.Dispose();
            connection.Close();
        } 
    }
}
View Code

 

上一节定向模式中,需要多次绑定,本节通配符模式可以模糊匹配

 @天才卧龙的博客

标签:string,NetCore,预警,Topics,通配符,factory,RabbitMQ,using,channel
来源: https://www.cnblogs.com/chenwolong/p/Rabbit_Topic.html