首页 > 代码库 > RabbitMQ消息队列（四）：分发到多Consumer（Publish/Subscribe）[转]

RabbitMQ消息队列（四）：分发到多Consumer（Publish/Subscribe）[转]

2024-10-25 15:38:02   211人阅读

上篇文章中，我们把每个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
 1 static void Main(string[] args) 2         { 3             var factory = new ConnectionFactory() { HostName = "localhost" }; 4             using (var connection = factory.CreateConnection()) 5             { 6                 using (var channel = connection.CreateModel()) 7                 { 8                     const string EXCHANGE_NAME = "logs"; 9                     const string ROUTING_KEY = "";10                     channel.ExchangeDeclare(EXCHANGE_NAME, "fanout");//广播11                     var message = GetMessage(args);12                     var body = Encoding.UTF8.GetBytes(message);13                     channel.BasicPublish(EXCHANGE_NAME, ROUTING_KEY, null, body);//不需要指定routing key，设置了fanout,指了也没有用.14                     Console.WriteLine(" [x] Sent {0}", message);15                 }16             }17         }

Consumer.cs

 1 static void Main(string[] args) 2         { 3             var factory = new ConnectionFactory() { HostName = "localhost" }; 4             using (var connection = factory.CreateConnection()) 5             { 6                 using (var channel = connection.CreateModel()) 7                 { 8                     const string EXCHANGE_NAME = "logs"; 9                     const string ROUTING_KEY = "";10                     channel.ExchangeDeclare(EXCHANGE_NAME, "fanout");//广播11                     QueueDeclareOk queueOk = channel.QueueDeclare();//每当Consumer连接时，我们需要一个新的，空的queue。因为我们不对老的log感兴趣。幸运的是，如果在声明queue时不指定名字，那么RabbitMQ会随机为我们选择这个名字。12                     ////现在我们已经创建了fanout类型的exchange和没有名字的queue（实际上是RabbitMQ帮我们取了名字）。13                     ////那exchange怎么样知道它的Message发送到哪个queue呢？答案就是通过bindings：绑定。14                     string queueName = queueOk.QueueName;//得到RabbitMQ帮我们取了名字15                     channel.QueueBind(queueName, EXCHANGE_NAME, ROUTING_KEY);//不需要指定routing key，设置了fanout,指了也没有用.16                     var consumer = new QueueingBasicConsumer(channel);17                     channel.BasicConsume(queueName, true, consumer);18                     Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C");19                     while (true)20                     {21                         var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//挂起的操作22                         var body = ea.Body;23                         var message = Encoding.UTF8.GetString(body);24                         Console.WriteLine(" [x] Received {0}", message);25                     }26                 }27             }28         }

 转：

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

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

RabbitMQ消息队列（四）：分发到多Consumer（Publish/Subscribe）[转]


联系
我们