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代码

using System;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQ.Common
{
    public class RabbitConstant
    {
        public const string QUEUE_HELLO_WORLD = "helloworld.queue";
        public const string QUEUE_SMS = "sms.queue";
        public const string EXCHANGE_WEATHER = "weather.exchange";
        public const string QUEUE_BAIDU = "baidu.queue";
        public const string QUEUE_SINA= "sina.queue";
        public const string EXCHANGE_WEATHER_ROUTING = "weather.routing.exchange";
        public const string EXCHANGE_WEATHER_TOPIC = "weather.topic.exchange";
    }
}

 (2)RabbitUtils.cs代码

using System;
using RabbitMQ.Client;

namespace RabbitMQ.Common
{
    public class RabbitUtils
    {
        public static ConnectionFactory GetConnection()
        {
            var factory = new ConnectionFactory();
            factory.HostName = "127.0.0.1";
            factory.Port = 5672;//是服务端的端口号，与页面的端口号15672区分开
            factory.UserName = "guest";
            factory.Password = "guest";
            //factory.VirtualHost = "/";
            return factory;
        }
    }
}

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

(1)WeatherFanout.cs代码

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Common;

namespace RabbitMQ.Producer.Producer
{
    public class WeatherFanout
    {
        public static void SendWeatherInfo()
        {
            using (var connection = RabbitUtils.GetConnection().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                        string message = "零下10度";
                        var body = Encoding.UTF8.GetBytes(message);
              
                        channel.BasicPublish(exchange: RabbitConstant.EXCHANGE_WEATHER,
                                             routingKey: "",
                                             basicProperties: null,
                                             body: body);
                    Console.WriteLine("天气信息发送成功");
                    Console.WriteLine("Press [Enter] to exit");
                    Console.ReadLine();
                }
            }
        }
    }
}

(2)Program.cs代码

using RabbitMQ.Producer.Producer;
using System;

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

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

(1)WeatherFanout.cs

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

namespace RabbitMQ.Consumer01.Consumer
{
    public  class WeatherFanout
    {
        public static void ReceiveWeatherInfo()
        {
            using (var connection = RabbitUtils.GetConnection().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(RabbitConstant.EXCHANGE_WEATHER, ExchangeType.Fanout);
                    channel.QueueDeclare(RabbitConstant.QUEUE_BAIDU, true, false, false, null);
                    /*
                     * QueueBind用于将队列与交换机绑定
                     * queue:队列名
                     * exchange:交换机名
                     * routingKey：路由key
                     */
                    channel.QueueBind(queue: RabbitConstant.QUEUE_BAIDU,
                                      exchange:RabbitConstant.EXCHANGE_WEATHER,
                                      routingKey:"");
                    channel.BasicQos(0, 1, false);
                    var consumer = new EventingBasicConsumer(channel);

                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"百度收到的气象信息：{message}");
                        channel.BasicAck(ea.DeliveryTag, false);
                    };
                    channel.BasicConsume(queue: RabbitConstant.QUEUE_BAIDU,
                                 autoAck: false,
                                 consumer: consumer);
                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }
    }
}

(2)Program.cs代码

using RabbitMQ.Consumer01.Consumer;
using System;

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

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

(1)WeatherFanout.cs

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

namespace RabbitMQ.Consumer02.Consumer
{
    public class WeatherFanout
    {
        public static void ReceiveWeatherInfo()
        {
            using (var connection = RabbitUtils.GetConnection().CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(RabbitConstant.EXCHANGE_WEATHER, ExchangeType.Fanout);
                    channel.QueueDeclare(RabbitConstant.QUEUE_SINA, true, false, false, null);
                    /*
                     * QueueBind用于将队列与交换机绑定
                     * queue:队列名
                     * exchange:交换机名
                     * routingKey：路由key
                     */
                    channel.QueueBind(queue: RabbitConstant.QUEUE_SINA,
                                      exchange: RabbitConstant.EXCHANGE_WEATHER,
                                      routingKey: "");
                    channel.BasicQos(0, 1, false);
                    var consumer = new EventingBasicConsumer(channel);

                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"新浪收到的气象信息：{message}");
                        channel.BasicAck(ea.DeliveryTag, false);
                    };
                    channel.BasicConsume(queue: RabbitConstant.QUEUE_SINA,
                                 autoAck: false,
                                 consumer: consumer);
                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }
    }
}

(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