RabbitMQ 消息隊列之應用 (三)

 1,路由模式  direct html

direct 模式基礎概念請參考  RabbitMQ消息隊列之基礎 (二)異步

direct類型的路由規則很簡單,它會把消息路由到那些binding key與routing key徹底匹配的Queue中;post

也就是說:只要隊列名、交換機、路由key徹底一致,就能夠匹配到該消息。ui

這裏直接演示代碼:this

生產者:url

 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當中配置spa

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:123456@127.0.0.1:5672//" />

消費者1:code

 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:htm

            //消費者2
            string queueName = "TestMq2";  //隊列名2
            string exchange = "TestMq_Exchange"; //交換器           TestMq_Exchange
            string routingKey = "TestMq_RoutingKey"; //路由關鍵字    TestMq_RoutingKey
            。。。
            。。。

下圖所示: 消費者1 徹底匹配, 消費者2的隊列名稱爲:TestMq2,與生產者不匹配blog

 

 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

相關文章
相關標籤/搜索