RabbitMQ学习-订阅模式(以C#代码为例)

本文主要是介绍RabbitMQ学习-订阅模式(以C#代码为例)，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

订阅模式（Publish/Subscribe）

一次向许多消费者发送消息，一个生产者发送的消息会被多个消费者获取，也就是将消息将广播到所有的消费者中。

应用场景： 更新商品库存后需要通知多个缓存和多个数据库，这里的结构应该是：

• 一个fanout类型交换机扇出两个个消息队列，分别为缓存消息队列、数据库消息队列
• 一个缓存消息队列对应着多个缓存消费者
• 一个数据库消息队列对应着多个数据库消费者

在订阅模型中，多了一个Exchange角色，而且过程略有变化：

   Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交所有队列、或是将消息丢弃。如何操作，取决于               Exchange的类型。Exchange有常见的以下三种类型

        Fanout:广播，将消息交给所有绑定到交换机的队列，订阅模式使用此类型

        Direct:定向，将消息交给符合指定routing key的队列 路由模式使用此类型

        Topic:通配符，把消息交给符合routing pattern的队列 主题模式使用此类型

Exchange只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失

Demo 模拟百度和新浪订阅气象局发布的天气信息 

1.RabbitMQ.Common类库代码

  (1)RabbitConstant.cs代码

 1 using System;
 2 using System.Collections.Generic;
 3 using System.Text;
 4 
 5 namespace RabbitMQ.Common
 6 {
 7     public class RabbitConstant
 8     {
 9         public const string QUEUE_HELLO_WORLD = "helloworld.queue";
10         public const string QUEUE_SMS = "sms.queue";
11         public const string EXCHANGE_WEATHER = "weather.exchange";
12         public const string QUEUE_BAIDU = "baidu.queue";
13         public const string QUEUE_SINA= "sina.queue";
14         public const string EXCHANGE_WEATHER_ROUTING = "weather.routing.exchange";
15         public const string EXCHANGE_WEATHER_TOPIC = "weather.topic.exchange";
16     }
17 }

 (2)RabbitUtils.cs代码

 1 using System;
 2 using RabbitMQ.Client;
 3 
 4 namespace RabbitMQ.Common
 5 {
 6     public class RabbitUtils
 7     {
 8         public static ConnectionFactory GetConnection()
 9         {
10             var factory = new ConnectionFactory();
11             factory.HostName = "127.0.0.1";
12             factory.Port = 5672;//是服务端的端口号，与页面的端口号15672区分开
13             factory.UserName = "guest";
14             factory.Password = "guest";
15             //factory.VirtualHost = "/";
16             return factory;
17         }
18     }
19 }

2. RabbitMQ.Producer控制台项目代码

(1)WeatherFanout.cs代码

 1 using System;
 2 using System.Text;
 3 using RabbitMQ.Client;
 4 using RabbitMQ.Common;
 5 
 6 namespace RabbitMQ.Producer.Producer
 7 {
 8     public class WeatherFanout
 9     {
10         public static void SendWeatherInfo()
11         {
12             using (var connection = RabbitUtils.GetConnection().CreateConnection())
13             {
14                 using (var channel = connection.CreateModel())
15                 {
16                         string message = "零下10度";
17                         var body = Encoding.UTF8.GetBytes(message);
18               
19                         channel.BasicPublish(exchange: RabbitConstant.EXCHANGE_WEATHER,
20                                              routingKey: "",
21                                              basicProperties: null,
22                                              body: body);
23                     Console.WriteLine("天气信息发送成功");
24                     Console.WriteLine("Press [Enter] to exit");
25                     Console.ReadLine();
26                 }
27             }
28         }
29     }
30 }

(2)Program.cs代码

 1 using RabbitMQ.Producer.Producer;
 2 using System;
 3 
 4 namespace RabbitMQ.Producer
 5 {
 6     class Program
 7     {
 8         static void Main(string[] args)
 9         {
10             WeatherFanout.SendWeatherInfo();
11         }
12     }
13 }

3.RabbitMQ.Consumer01控制台项目代码

(1)WeatherFanout.cs

 1 using System;
 2 using System.Collections.Generic;
 3 using System.Text;
 4 using RabbitMQ.Common;
 5 using RabbitMQ.Client;
 6 using RabbitMQ.Client.Events;
 7 
 8 namespace RabbitMQ.Consumer01.Consumer
 9 {
10     public  class WeatherFanout
11     {
12         public static void ReceiveWeatherInfo()
13         {
14             using (var connection = RabbitUtils.GetConnection().CreateConnection())
15             {
16                 using (var channel = connection.CreateModel())
17                 {
18                     channel.ExchangeDeclare(RabbitConstant.EXCHANGE_WEATHER, ExchangeType.Fanout);
19                     channel.QueueDeclare(RabbitConstant.QUEUE_BAIDU, true, false, false, null);
20                     /*
21                      * QueueBind用于将队列与交换机绑定
22                      * queue:队列名
23                      * exchange:交换机名
24                      * routingKey：路由key
25                      */
26                     channel.QueueBind(queue: RabbitConstant.QUEUE_BAIDU,
27                                       exchange:RabbitConstant.EXCHANGE_WEATHER,
28                                       routingKey:"");
29                     channel.BasicQos(0, 1, false);
30                     var consumer = new EventingBasicConsumer(channel);
31 
32                     consumer.Received += (model, ea) =>
33                     {
34                         var message = Encoding.UTF8.GetString(ea.Body.ToArray());
35                         Console.WriteLine($"百度收到的气象信息：{message}");
36                         channel.BasicAck(ea.DeliveryTag, false);
37                     };
38                     channel.BasicConsume(queue: RabbitConstant.QUEUE_BAIDU,
39                                  autoAck: false,
40                                  consumer: consumer);
41                     Console.WriteLine(" Press [enter] to exit.");
42                     Console.ReadLine();
43                 }
44             }
45         }
46     }
47 }

(2)Program.cs代码

 1 using RabbitMQ.Consumer01.Consumer;
 2 using System;
 3 
 4 namespace RabbitMQ.Consumer01
 5 {
 6     class Program
 7     {
 8         static void Main(string[] args)
 9         {
10             WeatherFanout.ReceiveWeatherInfo();
11         }
12     }
13 }

4.RabbitMQ.Consumer02控制台项目代码

(1)WeatherFanout.cs

 1 using System;
 2 using System.Collections.Generic;
 3 using System.Text;
 4 using RabbitMQ.Common;
 5 using RabbitMQ.Client;
 6 using RabbitMQ.Client.Events;
 7 
 8 namespace RabbitMQ.Consumer02.Consumer
 9 {
10     public class WeatherFanout
11     {
12         public static void ReceiveWeatherInfo()
13         {
14             using (var connection = RabbitUtils.GetConnection().CreateConnection())
15             {
16                 using (var channel = connection.CreateModel())
17                 {
18                     channel.ExchangeDeclare(RabbitConstant.EXCHANGE_WEATHER, ExchangeType.Fanout);
19                     channel.QueueDeclare(RabbitConstant.QUEUE_SINA, true, false, false, null);
20                     /*
21                      * QueueBind用于将队列与交换机绑定
22                      * queue:队列名
23                      * exchange:交换机名
24                      * routingKey：路由key
25                      */
26                     channel.QueueBind(queue: RabbitConstant.QUEUE_SINA,
27                                       exchange: RabbitConstant.EXCHANGE_WEATHER,
28                                       routingKey: "");
29                     channel.BasicQos(0, 1, false);
30                     var consumer = new EventingBasicConsumer(channel);
31 
32                     consumer.Received += (model, ea) =>
33                     {
34                         var message = Encoding.UTF8.GetString(ea.Body.ToArray());
35                         Console.WriteLine($"新浪收到的气象信息：{message}");
36                         channel.BasicAck(ea.DeliveryTag, false);
37                     };
38                     channel.BasicConsume(queue: RabbitConstant.QUEUE_SINA,
39                                  autoAck: false,
40                                  consumer: consumer);
41                     Console.WriteLine(" Press [enter] to exit.");
42                     Console.ReadLine();
43                 }
44             }
45         }
46     }
47 }

(2)Program.cs代码

using RabbitMQ.Consumer02.Consumer;
using System;

namespace RabbitMQ.Consumer02
{
    class Program
    {
        static void Main(string[] args)
        {
            WeatherFanout.ReceiveWeatherInfo();
        }
    }
}

5.执行，使用powershell运行 

参考连接：https://mp.weixin.qq.com/s/QG3uXhhpkE_Uo6Me15mxdg 

                  https://www.bilibili.com/video/BV1GU4y1w7Yq?p=8

这篇关于RabbitMQ学习-订阅模式(以C#代码为例)的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！