RabbitMQ的发布订阅模式（Publish/Subscribe）

一、发布/订阅（Publish/Subscribe）模式

发布订阅是我们经常会用到的一种模式，生产者生产消息后，所有订阅者都可以收到。RabbitMQ的发布/订阅模型图如下：

1、该模式下生产者并不是直接操作队列，而是将数据发送给交换机，由交换机将数据发送给与之绑定的队列；

2、该模式必须声明交换机，并且设置模式： channel.ExchangeDeclare(EXCHANGE_NAME, type: ExchangeType.Fanout);　　

fanout指分发模式（将每一条消息都发送到与交换机绑定的队)。

3、队列必须绑定交换机：channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

二、发布消息

消息生产者向交换机（exchange）发送消息，代码如下：

   // 定义交换机名称

        static string EXCHANGE_NAME = "ps_exchange_fanout";

        public static void PublishMessage()

        {

            try

            {

                var conn = RabbitMQHelper.GetConnection();

                var channel = conn.CreateModel();

                // 定义exchange

                channel.ExchangeDeclare(EXCHANGE_NAME, type: ExchangeType.Fanout);

                string msg = "hello ps!";

                var body = Encoding.UTF8.GetBytes(msg);

                channel.BasicPublish(EXCHANGE_NAME, "", null, body);

                Console.WriteLine("send msg:" + msg);

                channel.Close();

                conn.Close();

            }

            catch (Exception ex)

            {

                throw ex;

            }

        }

消息发送成功，截图如下：

三、订阅消息

在这里需要两个消费者，消息发送后，所有的订阅者都可以收到消息；

3.1 消费者1

和轮询分发以及公平分发不同的是，消费者需要将队列绑定到交换机，来订阅消息；实现代码如下：

        static string EXCHANGE_NAME = "ps_exchange_fanout";

        static string QUEUE_NAME = "ps_queue_sub1";

        /// <summary>

        /// 订阅消费者1

        /// </summary>

        static void SubscribeConsumer1()

        {

            var conn = RabbitMQHelper.GetConnection();

            var channel = conn.CreateModel();

            // 定义exchange

            channel.ExchangeDeclare(EXCHANGE_NAME, type: ExchangeType.Fanout);

            // 绑定queue

            channel.QueueDeclare(queue: QUEUE_NAME);

            channel.QueueBind(QUEUE_NAME, EXCHANGE_NAME, "");

            // 定义Consumer

            var consumer = new EventingBasicConsumer(channel);

            consumer.Received += (model,ea) =>

            {

                var body = ea.Body;

                var message = Encoding.UTF8.GetString(body.ToArray());

                Console.WriteLine($"SubscribeConsumer1 收到消息： {message}，时间：{DateTime.Now}");

            };

            //启动消费者 设置为手动应答消息

            channel.BasicConsume(queue: QUEUE_NAME, autoAck: true, consumer: consumer);

            Console.WriteLine("Subscribe Consumer1 消费者已启动");

            Console.ReadKey();

            channel.Dispose();

            conn.Close();

        }

3.2 消费者2

消费者2的代码和1的基本相同，大家可以将1的修改一下，就可以使用，在此就不重复贴出了；

3.3 接收消息结果

消费者1接收消息截图：

消费者2接收消息截图：

通过上图，我们可以看到，发布者发布消息后，订阅者1、2均受到了相同的消息，至此功能已经完成；

四、小结

4.1 订阅者代码的主要流程

根据消费者的代码，我们可以提炼流程如下

（1）创建连接

（2）声明exchange

（3）绑定队列到exchange

（4）声明消费者

（5）绑定消费者到channel，监听处理消息

（6）关闭连接

4.2 订阅成功后，我们打开mq的管理地址可以看到，有两个queue绑定到exchange上了：

注意：如果我们的exchange没有消费者订阅，发布的消息将不会被保存到任何队列，直接丢失了；

参考链接：https://www.rabbitmq.com/tutorials/tutorial-three-dotnet.html

巴特西