ICode9

精准搜索请尝试: 精确搜索
首页 > 其他分享> 文章详细

NetCore RabbitMQ ,Routing定向模式

2022-07-22 18:14:53  阅读:207  来源: 互联网

标签:string NetCore 队列 factory RabbitMQ Routing using channel


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

学无止境,精益求精

上篇博客介绍了RabbitMQ的六种工作模式 RabbitMQ的六种工作模式 

RabbitMQ的简单模式和Work工作模式请参考:NetCore RabbitMQ 简介及兔子生产者、消费者 【简单模式,work工作模式,竞争消费】

RabbitMQ发布订阅模式之消息广播请参考:NetCore RabbitMQ 发布订阅模式,消息广播

本篇博客使用NetCore完成RabbitMQ发布订阅模式中的定向模式

何为定向模式?

1.4 routing路由模式

在这里插入图片描述

  1. 消息生产者将消息发送给交换机按照路由判断,路由是字符串(info) 当前产生的消息携带路由字符(对象的方法),交换机根据路由的key,只能匹配上路由key对应的消息队列,对应的消费者才能消费消息;
  2. 根据业务功能定义路由字符串
  3. 从系统的代码逻辑中获取对应的功能字符串,将消息任务扔到对应的队列中业务场景:error 通知;EXCEPTION;错误通知的功能;传统意义的错误通知;客户通知;利用key路由,可以将程序中的错误封装成消息传入到消息队列中,开发者可以自定义消费者,实时接收错误;

上图中X代表交换机,RabbitMQ中交换机的类型分为四种,分别为广播模式,定向模式,通配符模式,参数匹配模式

ExchangeType.Fanout【广播模式】 

ExchangeType.Direct【定向模式】 

ExchangeType.Topic【通配符模式】 

ExchangeType.Headers 【参数匹配模式】

广播模式中,我们通过设定相同的RoutingKey,完成消息的广播,定向模式中,我们只需在队列绑定交换机时赋值不同的RoutingKey且生产者生产消息时指定响应的RoutingKey即可完成消息定向到对应队列。

定向模式生产者

定向模式创建生产者,分为如下步骤,

1、声明一个交换机

2、声明广播的队列

3、交换机和队列进行绑定【创建不同的routingKey,用于区分绑定规则】

4、生产消息,生产消息时,指定生产消息的routingKey,已达到消息定向的作用

以上步骤用NetCore 实现如下: 

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_Direct";
                    channel.ExchangeDeclare(Ename, ExchangeType.Direct, false, false, null);
                    //声明广播的队列
                    string Qname_Red = "RabbitMQ_Queue_Red";
                    string Qname_Blue = "RabbitMQ_Queue_Blue";
                    string Qname_Yellow = "RabbitMQ_Queue_Yellow";
                    channel.QueueDeclare(Qname_Red, false, false, false, null);
                    channel.QueueDeclare(Qname_Blue, false, false, false, null);
                    channel.QueueDeclare(Qname_Yellow, 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"; // 蓝色预警
                    string routingKey_Yellow = "Yellow"; // 黄色预警
                    //只接受红色预警
                    channel.QueueBind(Qname_Red, Ename, routingKey_Red);
                    // Qname_Blue接受两种消息类型
                    channel.QueueBind(Qname_Blue, Ename, routingKey_Red);
                    channel.QueueBind(Qname_Blue, Ename, routingKey_Blue);
                    // Qname_Yellow 介绍三种消息类型
                    channel.QueueBind(Qname_Yellow, Ename, routingKey_Red);
                    channel.QueueBind(Qname_Yellow, Ename, routingKey_Blue);
                    channel.QueueBind(Qname_Yellow, Ename, routingKey_Yellow);
                    //发送消息
                    for(int i = 0; i < 10 ; i++)
                    {
                        var messages_Red = "I am Red Police"; //传递的消息内容
                        var messages_Blue = "I am Blue Police"; //传递的消息内容
                        var messages_Yellow = "I am Yellow Police"; //传递的消息内容
                                                                    //exchange 交换机,如果使用默认的交换机,那么routingKey要和队列的名称一致
                                                                    //routingKey:路由  
                                                                    //basicProperties : 用于基础属性设置
                        ///BasicPublish(this IModel model, string exchange, string routingKey, IBasicProperties basicProperties, ReadOnlyMemory<byte> body);
                        channel.BasicPublish(Ename, routingKey_Red, null, Encoding.UTF8.GetBytes(messages_Red + "_" + i)); //生产红色预警消息
                        channel.BasicPublish(Ename, routingKey_Blue, null, Encoding.UTF8.GetBytes(messages_Blue + "_" + i)); //生产蓝色预警消息
                        channel.BasicPublish(Ename, routingKey_Yellow, null, Encoding.UTF8.GetBytes(messages_Yellow + "_" + i)); //生产黄色预警消息
                    }
                }
            }
            Console.Read();
        }
    }
}
View Code

执行上述代码,得到如下信息

此时:三个队列中红色预警队列有10条消息,蓝色预警队列有20条消息,黄色预警队列有30条消息

 

 交换机信息,注意交换机的类型为定向模式

定向模式消费者

【红色预警队列消费者】

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 = "RabbitMQ_Queue_Red";
            channel.BasicConsume(Qname, false, consumer);
            Console.WriteLine("消费者已启动");
            Console.ReadKey();
            channel.Dispose();
            connection.Close();
        } 
    }
}
View Code

因为红色预警队列只绑定了红色预警的RoutingKey,因此,该队列只接受红色预警消息

 

 【蓝色预警队列消费者】

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 = "RabbitMQ_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 = "RabbitMQ_Queue_Yellow";
            channel.BasicConsume(Qname, false, consumer);
            Console.WriteLine("消费者已启动");
            Console.ReadKey();
            channel.Dispose();
            connection.Close();
        } 
    }
}
View Code

 

 定向模式中的核心为路由Key的声明、路由Key的绑定、发送带有路由Key的消息,这3点为实现定向发送的核心。

@陈大六的博客

标签:string,NetCore,队列,factory,RabbitMQ,Routing,using,channel
来源: https://www.cnblogs.com/chenwolong/p/Rabbit_Direct.html

本站声明: 1. iCode9 技术分享网(下文简称本站)提供的所有内容,仅供技术学习、探讨和分享;
2. 关于本站的所有留言、评论、转载及引用,纯属内容发起人的个人观点,与本站观点和立场无关;
3. 关于本站的所有言论和文字,纯属内容发起人的个人观点,与本站观点和立场无关;
4. 本站文章均是网友提供,不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属;如您发现该文章侵犯了您的权益,可联系我们第一时间进行删除;
5. 本站为非盈利性的个人网站,所有内容不会用来进行牟利,也不会利用任何形式的广告来间接获益,纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

专注分享技术,共同学习,共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有