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

 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(翻译)

时间: 2024-10-13 23:13:09

RabbitMQ消息队列(四):分发到多Consumer(Publish/Subscribe)[转]的相关文章

消息队列 RabbitMQ系列 第四篇:发布/订阅 Publish/Subscribe

上篇中我们实现了Work Queue的创建,在Work Queue背后,其实是rabbitMQ把每条任务消息只发给一个消费者.本篇中我们将要研究如何把一条消息推送给多个消费者,这种模式被称为publish/subscribe(发布/订阅). 为了说明这个模式,我们将会构建一个简单的日志系统.这将会包含两部分程序,第一个是发送日志信息,第二个将会接收并打印它们. 在我们的日志系统里,每个运行的消费者程序都能接收到消息.这样我就运行一个receiver并把日志写到磁盘上,同时我们再运行另外一个消费者

RabbitMQ消息队列(四):分发到多Consumer(Publish/Subscribe)

目录(?)[-] Exchanges Temporary queues Bindings绑定 最终版本 上篇文章中,我们把每个Message都是deliver到某个Consumer.在这篇文章中,我们将会将同一个Message deliver到多个Consumer中.这个模式也被成为 "publish / subscribe".    这篇文章中,我们将创建一个日志系统,它包含两个部分:第一个部分是发出log(Producer),第二个部分接收到并打印(Consumer). 我们将构建

RabbitMQ消息队列(六):使用主题进行消息分发[转]

在上篇文章RabbitMQ消息队列(五):Routing 消息路由 中,我们实现了一个简单的日志系统.Consumer可以监听不同severity(严重级别)的log.但是,这也是它之所以叫做简单日志系统的原因,因为是仅仅能够通过severity(严重级别)设定.不支持更多的标准. 比如syslog unix的日志工具,它可以通过severity (info/warn/crit...) 和模块(auth/cron/kern...).这可能更是我们想要的:我们可以仅仅需要cron模块的log. 为

(转)RabbitMQ消息队列(六):使用主题进行消息分发

在上篇文章RabbitMQ消息队列(五):Routing 消息路由 中,我们实现了一个简单的日志系统.Consumer可以监听不同severity的log.但是,这也是它之所以叫做简单日志系统的原因,因为是仅仅能够通过severity设定.不支持更多的标准. 比如syslog unix的日志工具,它可以通过severity (info/warn/crit...) 和模块(auth/cron/kern...).这可能更是我们想要的:我们可以仅仅需要cron模块的log. 为了实现类似的功能,我们需

RabbitMQ消息队列(六):使用主题进行消息分发

在上篇文章RabbitMQ消息队列(五):Routing 消息路由 中,我们实现了一个简单的日志系统.Consumer可以监听不同severity的log.但是,这也是它之所以叫做简单日志系统的原因,因为是仅仅能够通过severity设定.不支持更多的标准. 比如syslog unix的日志工具,它可以通过severity (info/warn/crit...) 和模块(auth/cron/kern...).这可能更是我们想要的:我们可以仅仅需要cron模块的log. 为了实现类似的功能,我们需

RabbitMQ消息队列应用

RabbitMQ消息队列应用 消息通信组件Net分布式系统的核心中间件之一,应用与系统高并发,各个组件之间解耦的依赖的场景.本框架采用消息队列中间件主要应用于两方面:一是解决部分高并发的业务处理:二是通过消息队列传输系统日志.目前业界使用较多的消息队列组件有RabbitMQ.ActiveMQ.MSMQ.kafka.zeroMQ等,本文对系统架构之MQ Component诠释,并采用RabbitMQ作为消息队列中间件. 图1- 消息队列组件示意图 一.RabbitMQ介绍 RabbitMQ是一款基

(转)RabbitMQ消息队列(九):Publisher的消息确认机制

在前面的文章中提到了queue和consumer之间的消息确认机制:通过设置ack.那么Publisher能不到知道他post的Message有没有到达queue,甚至更近一步,是否被某个Consumer处理呢?毕竟对于一些非常重要的数据,可能Publisher需要确认某个消息已经被正确处理. 在我们的系统中,我们没有是实现这种确认,也就是说,不管Message是否被Consume了,Publisher不会去care.他只是将自己的状态publish给上层,由上层的逻辑去处理.如果Message

(转)(二)RabbitMQ消息队列-RabbitMQ消息队列架构与基本概念

http://blog.csdn.net/super_rd/article/details/70238869 没错我还是没有讲怎么安装和写一个HelloWord,不过快了,这一章我们先了解下RabbitMQ的基本概念. RabbitMQ架构 说是架构其实更像是应用场景下的架构(自己画的有点丑,勿嫌弃) 从图中可以看出RabbitMQ主要由Exchange和Queue两部分组成,然后通过RoutingKey关联起来,消息投递到Exchange然后通过Queue接收. RabbitMQ消息队列基本概

RabbitMQ消息队列(九):Publisher的消息确认机制

在前面的文章中提到了queue和consumer之间的消息确认机制:通过设置ack.那么Publisher能不到知道他post的Message有没有到达queue,甚至更近一步,是否被某个Consumer处理呢?毕竟对于一些非常重要的数据,可能Publisher需要确认某个消息已经被正确处理. 在我们的系统中,我们没有是实现这种确认,也就是说,不管Message是否被Consume了,Publisher不会去care.他只是将自己的状态publish给上层,由上层的逻辑去处理.如果Message