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

zqllove 2020-05-12 原文

一、发布/订阅（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();
        }