RabbitMQ 消息队列之应用 (三)

1，路由模式 direct

direct 模式基础概念请参考 RabbitMQ消息队列之基础 (二)

direct类型的路由规则很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中；

也就是说：只要队列名、交换机、路由key完全一致，就可以匹配到该消息。

这里直接演示代码：

生产者：

1                 //direct类型   路由模式  1对1匹配
 2                 //生产者发送消息时需要指定一个路由键(routingKey),交换机只会把消息转发给包含该路由键的队列
 3                 //string exchange = "TestMq_Exchange";  //交换机
 4                 //string routingKey = "TestMq_RoutingKey";  //路由键
 5 
 6                 string queueName = "TestMq";  //队列名
 7                 for (int i = 0; i < 10; i++)
 8                 {
 9                     string message = "Hello World:" + i;
10                     RabbitMqConfig rabbitMqConfig = RabbitMqConfig.Init();  //初始化配置文件
11                     ConnectionFactory connFactory = new ConnectionFactory
12                     {
13                         Uri = new Uri(url),
14                         RequestedConnectionTimeout = rabbitMqConfig.RequestedConnectionTimeout,
15                         RequestedChannelMax = rabbitMqConfig.RequestedChannelMax,
16                         RequestedHeartbeat = rabbitMqConfig.RequestedHeartbeat,
17                         AutomaticRecoveryEnabled = false
18                     };
19                     string exchange = queueName + "_Exchange";     //交换机
20                     string routingKey = queueName + "_RoutingKey";    //路由键
21                     string exchangeType = ExchangeType.Direct;   //类型 direct
22 
23                     IModel model = this._conn.CreateModel();
24                     model.ExchangeDeclare(exchange, exchangeType, true, false, null); //声明交换机  
25                     model.QueueDeclare(queueName, true, false, false, null);  //声明队列
26                     model.QueueBind(queueName, exchange, routingKey, null);  //绑定
27 
28                     IBasicProperties basicProperties = model.CreateBasicProperties();
29                     basicProperties.Persistent = true;    //消息持久化
30                     basicProperties.DeliveryMode = 2;     //消息持久化, 默认为1 非持久化
31                     byte[] bytes = Encoding.UTF8.GetBytes(message);
32                     model.BasicPublish(exchange, routingKey, basicProperties, bytes);
33 
34                     Console.WriteLine(DateTime.Now);
35                 }

初始化配置文件，在config当中配置

1 ConfigurationManager.GetSection("RabbitMqConfig"); //读取配置文件
2 
3   //config配置文件
4   <configSections>
5     <section name="RabbitMqConfig" type="Rabbit.Common.RabbitMqConfig,Rabbit.Common" />
6   </configSections>
7   <RabbitMqConfig RequestedHeartbeat="60" RequestedConnectionTimeout="300" RequestedChannelMax="500" Uri="amqp://admin:[email protected]:5672//" />

消费者1：

1             //消费者1
 2             string queueName = "TestMq";  //队列名
 3             string exchange = queueName + "_Exchange"; //交换器 
 4             string routingKey = queueName + "_RoutingKey"; //路由关键字
 5 
 6             var rabbitMq = RabbitMqConfig.Init(); //获取Rabbit队列配置
 7             var rm = new ConnectionFactory()
 8             {
 9                 Uri = new Uri(rabbitMq.Uri),
10                 RequestedConnectionTimeout = rabbitMq.RequestedConnectionTimeout,
11                 RequestedChannelMax = rabbitMq.RequestedChannelMax,
12                 RequestedHeartbeat = rabbitMq.RequestedHeartbeat,
13             };
14 
15             var connection = rm.CreateConnection(); //创建连接
16             var channel = connection.CreateModel(); //创建通道
17             //开启队列持久化(durable = true),不自动删除(autoDelete = false),是否专属(exclusive: false)
18             channel.ExchangeDeclare(exchange, ExchangeType.Direct, true, false, null); //声明一个交换器
19             channel.QueueDeclare(queueName, true, false, false, null); //声明一个队列
20             channel.QueueBind(queueName, exchange, routingKey, null); //绑定交换器和路由
21             channel.BasicQos(0, 1, false);  //每次只接收1个，处理完后再接收下一个
22             var consumer = new EventingBasicConsumer(channel);
23 
24             consumer.Received += (model, ea) =>
25             {
26                 var message = Encoding.UTF8.GetString(ea.Body);  //消息主体
27 
28                 //处理消息逻辑,可以使用异步处理
29                 Console.WriteLine(DateTime.Now + " 收到消息：" + message);
30                 Thread.Sleep(3000);   //模拟消耗延时
31 
32                 channel.BasicAck(ea.DeliveryTag, false); //返回确认状态  该条消息将会从队列当中移除
33 
34             };
35             //监听队列，手动返回完成 第二个参数值为false代表关闭RabbitMQ的自动应答机制，改为手动应答。
36             channel.BasicConsume(queueName, false, consumer);

消费者2：

//消费者2
            string queueName = "TestMq2";  //队列名2
            string exchange = "TestMq_Exchange"; //交换器           TestMq_Exchange
            string routingKey = "TestMq_RoutingKey"; //路由关键字    TestMq_RoutingKey
            。。。
            。。。

下图所示：消费者1 完全匹配，消费者2的队列名称为：TestMq2，与生产者不匹配

2，广播模式 / 分发模式 ( fanout )

这种模式下,消息会被所有消费者消费.也就是说,只要是"绑定"到某个交换机的队列,都会收到生产者发送到该交换机的消息.

fanout 类型的发送规则非常简单，它会把所有发送到该交换机Exchange的消息路由到所有与它绑定的Queue中；

也就是说：在fanout模式下，只跟交换机Exchange有关系，跟路由key无关。

1             string queueName = "TestMq";
 2             string exchange = "TestMq_Exchange";  //只需要声明交换机 即可
 3             string routingKey = "";  //路由键
 4             string exchangeType = ExchangeType.Fanout;   //类型 Fanout
 5             for (int i = 0; i < 10; i++)
 6             {
 7                 string message = "Hello World:" + i;
 8                 RabbitMqConfig rabbitMqConfig = RabbitMqConfig.Init();  //初始化配置文件
 9                 ConnectionFactory connFactory = new ConnectionFactory
10                 {
11                     Uri = new Uri(url),
12                     RequestedConnectionTimeout = rabbitMqConfig.RequestedConnectionTimeout,
13                     RequestedChannelMax = rabbitMqConfig.RequestedChannelMax,
14                     RequestedHeartbeat = rabbitMqConfig.RequestedHeartbeat,
15                     AutomaticRecoveryEnabled = false
16                 };
17 
18                 IModel model = this._conn.CreateModel();
19                 model.ExchangeDeclare(exchange, exchangeType, true, false, null); //声明交换机  
20                 model.QueueDeclare(queueName, true, false, false, null);  //声明队列
21                 model.QueueBind(queueName, exchange, routingKey, null);  //绑定
22 
23                 IBasicProperties basicProperties = model.CreateBasicProperties();
24                 basicProperties.Persistent = true;    //消息持久化
25                 basicProperties.DeliveryMode = 2;     //消息持久化, 默认为1 非持久化
26                 byte[] bytes = Encoding.UTF8.GetBytes(message);
27                 model.BasicPublish(exchange, routingKey, basicProperties, bytes);
28 
29                 Console.WriteLine(DateTime.Now + ":" + message);
30             }

消费者1：这里设置： queueName = TestMq

string queueName = "TestMq";  //队列名
        string exchange = "TestMq_Exchange";  //交换机
        string routingKey = "";  //路由键

        ....
        ....
        
        channel.ExchangeDeclare(exchange, ExchangeType.Fanout, true, false, null); //声明一个交换器

消费者2：这里设置： queueName = TestMq

1         string queueName = "TestMq";  //队列名
2         string exchange = "TestMq_Exchange";  //交换机
3         string routingKey = "";  //路由键
4 
5         ....
6         ....
7         
8         channel.ExchangeDeclare(exchange, ExchangeType.Fanout, true, false, null); //声明一个交换器

上边示例当中使用相同的 queueName = "TestMq"; 会出现如下图所示的情况，各消费一半的情况，因为系统认为是一个队列

下边我们修改一下队列名称：

1            //消费者1
2             string queueName = "TestMq1";  //队列名1
3             string exchange = "TestMq_Exchange";  //交换机
4             string routingKey = "";  //路由键

1             //消费者2
2             string queueName = "TestMq2";  //队列名2
3             string exchange = "TestMq_Exchange";  //交换机
4             string routingKey = "";  //路由键

运行结果如下：分别由队列1和队列2 接收

3，主题模式/模糊匹配 ( fanout )

对于topic类型，

生产者：的消息的路由键routing key 一般不会任意给定。它一般是一些单词的集合，中间用点号.分割。这些单词可以任意，但通常要体现出消息的特征。

一些有效的路由键示例：stock.usd.nyse，nyse.vmw，quick.orange.rabbit。这些路由键可以包含很多单词，但路由键总长度不能超过255个字节；

消费者：binding key也一般是这种形式。topic类型的逻辑与direct类似：以特定路由键发送的消息将会发送到所有绑定键与之匹配的队列中。

但绑定键有两种特殊的情况：

　　*（星号）仅代表一个单词

　　#（井号）代表任意个单词

也就是说：

当消费者队列以绑定键“#”绑定，它将会接收到所有的消息，而无视路由键（实际是绑定键

匹配了任意的路由键）。----这和

fanout模式

一样了。

当消费者队列中，

和

这两个特殊的字符不出现在绑定键中，

Topic类型

就会和

direct类型一样

了。

特别重要的提示：匹配的最小单位是 “单词”，且中间用点号.分割

下图很好的说明了 topic模式的匹配规则：

1             //生产者
2             string queueName = "TestMq";
3             string exchange = "TestMq_Exchange";  //交换机
4             string routingKey = "test.key";  //路由键
5             string exchangeType = ExchangeType.Topic;   //类型 Topic
6 
7             。。。
8             。。。

1             //消费者1
2             string queueName = "TestMq1";  //队列名1
3             string exchange = "TestMq_Exchange";  //交换机
4             string routingKey = "test";  //路由键为test
5 
6             。。。
7             。。。
8             //开启队列持久化(durable = true),不自动删除(autoDelete = false),是否专属(exclusive: false)
9             channel.ExchangeDeclare(exchange, ExchangeType.Topic, true, false, null); //声明一个交换器

1             //消费者2
2             string queueName = "TestMq2";  //队列名2
3             string exchange = "TestMq_Exchange";  //交换机
4             string routingKey = "#";  //路由键
5 
6              。。。

下图所示：消费者1 的路由key没有匹配到消息，而消费者2 使用 # 作为路由key,匹配到消息

接下来，我们修改一下消费者的路由key

1   //消费者1
2   string queueName = "TestMq1";  //队列名1
3   string exchange = "TestMq_Exchange";  //交换机
4   string routingKey = "test.*";  //路由键 修改为 test.*    效果同：*.key

1    //消费者2
2    string queueName = "TestMq2";  //队列名2
3    string exchange = "TestMq_Exchange";  //交换机
4    string routingKey = "#.key";  //路由键 修改为 #.key     效果同：test.#

下图表示：上边两个消费者均匹配到了生产者发布的消息

参考文档： https://www.cnblogs.com/zhangbLearn/p/9559336.html

作者：PeterZhang

出处：https://www.cnblogs.com/peterzhang123

本文版权归作者和博客园共有，欢迎转载，但必须给出原文链接，并保留此段声明，否则保留追究法律责任的权利。