RabbitMQ消息隊列（四）：分發到多Consumer（Publish/Subscribe）[轉]

上篇文章中，我們把每個Message都是deliver（提供）到某個Consumer。在這篇文章中，我們將會將同一個Message deliver（提供）到多個Consumer中。這個模式也被成為 "publish / subscribe"。
    這篇文章中，我們將創建一個日志系統，它包含兩個部分：第一個部分是發出log（Producer），第二個部分接收到並打印（Consumer）。 我們將構建兩個Consumer，第一個將log寫到物理磁盤上；第二個將log輸出的屏幕。

1. Exchanges

   關於exchange的概念在《RabbitMQ消息隊列（一）: Detailed Introduction 詳細介紹》中有詳細介紹。現在做一下簡單的回顧。

   RabbitMQ 的Messaging Model就是Producer並不會直接發送Message到queue。實際上，Producer並不知道它發送的Message是否已經到達queue。

   Producer發送的Message實際上是發到了Exchange中。Exchange的功能也很簡單：從Producer接收Message，然后投遞到queue中。Exchange需要知道如何處理Message，是把它放到一個queue中，還是放到多個queue中？這個rule是通過 Exchange 的類型定義的。

 

我們知道有三種類型的Exchange： direct, topic 和fanout。fanout就是廣播模式，會將所有的Message都放到它所知道的queue中。創建一個名字為logs，類型為fanout的Exchange。

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

2. Temporary queues

    截至現在，我們用的queue都是有名字的：第一個是hello，第二個是task_queue。使用有名字的queue，使得在Producer和Consumer之前共享queue成為可能。

    但是對於我們將要構建的日志系統，並不需要有名字的queue。我們希望得到所有的log，而不是它們中間的一部分。而且我們只對當前的log感興趣。為了實現這個目標，我們需要兩件事情：
    1） 每當Consumer連接時，我們需要一個新的，空的queue。因為我們不對老的log感興趣。幸運的是，如果在聲明queue時不指定名字，那么RabbitMQ會隨機為我們選擇這個名字。基本上都是這個樣子： amq.gen-JzTY20BRgKO-HjmUJj0wLg。

     2）當Consumer關閉連接時，這個queue要被deleted。可以加個exclusive的參數。方法：NET好像不用設置exclusive，Consumer關閉之后會自動刪掉Queue。

channel.QueueDeclare(queueName, durable, exclusive, autoDelete, null);

3. Bindings綁定

現在我們已經創建了fanout類型的exchange和沒有名字的queue（實際上是RabbitMQ幫我們取了名字）。那exchange怎么樣知道它的Message發送到哪個queue呢？答案就是通過bindings：綁定。

方法：

取名字：

channel.ExchangeDeclare(EXCHANGE_NAME, "fanout");//廣播
QueueDeclareOk queueOk = channel.QueueDeclare();
var queueName = queueOk.QueueName;

綁定：

channel.QueueBind(queueName, EXCHANGE_NAME, ROUTING_KEY);//不需要指定routing key，設置了fanout,指了也沒有用.

現在logs的exchange就將它的Message附加到我們創建的queue了。

完整的fanout例子：

Producer.cs

static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    const string EXCHANGE_NAME = "logs";
                    const string ROUTING_KEY = "";
                    channel.ExchangeDeclare(EXCHANGE_NAME, "fanout");//廣播
                    var message = GetMessage(args);
                    var body = Encoding.UTF8.GetBytes(message);
                    channel.BasicPublish(EXCHANGE_NAME, ROUTING_KEY, null, body);//不需要指定routing key，設置了fanout,指了也沒有用.
                    Console.WriteLine(" [x] Sent {0}", message);
                }
            }
        }

Consumer.cs

static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    const string EXCHANGE_NAME = "logs";
                    const string ROUTING_KEY = "";
                    channel.ExchangeDeclare(EXCHANGE_NAME, "fanout");//廣播
                    QueueDeclareOk queueOk = channel.QueueDeclare();//每當Consumer連接時，我們需要一個新的，空的queue。因為我們不對老的log感興趣。幸運的是，如果在聲明queue時不指定名字，那么RabbitMQ會隨機為我們選擇這個名字。
                    ////現在我們已經創建了fanout類型的exchange和沒有名字的queue（實際上是RabbitMQ幫我們取了名字）。
                    ////那exchange怎么樣知道它的Message發送到哪個queue呢？答案就是通過bindings：綁定。
                    string queueName = queueOk.QueueName;//得到RabbitMQ幫我們取了名字
                    channel.QueueBind(queueName, EXCHANGE_NAME, ROUTING_KEY);//不需要指定routing key，設置了fanout,指了也沒有用.
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume(queueName, true, consumer);
                    Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C");
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//掛起的操作
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine(" [x] Received {0}", message);
                    }
                }
            }
        }

 轉：

http://www.rabbitmq.com/tutorials/tutorial-three-dotnet.html（官網）

http://blog.csdn.net/anzhsoft/article/details/19617305（翻譯）