.Net下RabbitMQ的使用(4) 订阅和发布

消息的订阅和发布是使用消息队列的常用场景。在上一篇文章中，虽然有多个消费者，但是一个消息只会有一个消费者来处理。而订阅和发布则是每个订阅该消息的消费者都会收到这个消息。RabbitMQ的路由机制让我们实现这个功能轻而易举。

要了解RabbitMQ的路由机制，exchange是一个关键。exchange可以叫做交换机，也似乎可以叫做路由器，反正它是用来选择路由的。前文说到，RabbitMQ的核心思想就是消息的发布者不是直接把消息发送到目标队列中的，事实上，通常它并不知道消息要发到哪个队列中，它只知道把消息队列发送到exchange中。exchange一边接收发送者发过来的消息，而另一边则把消息发送到目标队列中去。exchange一定知道哪些队列需要接收这个消息，是加到一个队列里还是加到好几个队列里，还是直接扔掉。下图中的X就是exchange。

RabbitMQ的exchange有一些类型，这些类型决定了exchange的行为。分别是 direct, topic, headers 和 fanout四种类型。在这一篇文章中介绍的最简单的发送和接收例子中使用的如下代码中

//指定发送的路由，通过默认的exchange直接发送到指定的队列中。
channel.BasicPublish("", "esbtest.rmq.consoleserver", null, bytes);

第一个参数我们输入了空字符串来代表一个exchange。空字符串的exchange在RabbitMQ中时默认的exchange，类型是direct。在这个例子中它会直接将消息发送到第二个参数route_key定义的同名的队列中。

而我们这篇介绍的订阅和发布是用了fanout这个类型。由于默认类型是direct的，所以需要使用fanout就需要额外定义。如下代码就是定义了一个名字叫publish的fanout类型的exchange。

channel.ExchangeDeclare("publish", "fanout");

这种exchange有分发的意思。那分发到哪些队列中呢？因为发布者不需要知道，所以这段代码页就在订阅者那边来实现。来看订阅者得代码片段：

using (IConnection connection = factory.CreateConnection())
            {
                using (IModel channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare("publish", "fanout");
                    //随机创建一个队列
                    string queue_name = channel.QueueDeclare("subscriber1", true, false, false, null);
                    //绑定到名字叫publish的exchange上
                    channel.QueueBind(queue_name, "publish", "");
                    //定义这个队列的消费者
                    QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume(queue_name, true, consumer);

                    while (true)
                    {
                        BasicDeliverEventArgs ea =
                            (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                        byte[] bytes = ea.Body;

                        XmlSerializer xs = new XmlSerializer(typeof(RequestMessage));
                        using (MemoryStream ms = new MemoryStream(bytes))
                        {
                            RequestMessage message = (RequestMessage)xs.Deserialize(ms);
                            Console.WriteLine("Receive a Message, Id:" + message.MessageId + " Message:" + message.Message);
                        }
                    }
                }
            }

同样先定义一个fanout的exchange。对于为什么要在发布者和订阅者都要定义同名的exchange，我的理解是如果没有定义的一方先启动的话则会报错说找不到那个exchange。我测试过如果有定义的先启动，没定义的后启动也是没有问题的。

因为每个订阅者都需要一个队列来存放发给自己的消息，所以需要创建一个队列。通过QueueBind来和exchange关联了。所有发送给名字为publish的exchange的消息，都会被它分发给所有与之绑定的队列中，这样，每个对应的消费者都会收到一个副本。

在浏览器的管理界面上我们可以看见，RabbitMQ为每一个消费者（订阅者）创建了一个队列，而没有为发送者创建队列。

从这里下载示例代码