zoukankan      html  css  js  c++  java
  • 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(翻译)

  • 相关阅读:
    各大互联网公司架构演进之路汇总
    Java工程师成神之路~(2018修订版)
    Java的并发编程中的多线程问题到底是怎么回事儿?
    深入理解多线程(五)—— Java虚拟机的锁优化技术
    Java中线程池,你真的会用吗?
    深入理解多线程(四)—— Moniter的实现原理
    深入理解多线程(三)—— Java的对象头
    深入理解多线程(二)—— Java的对象模型
    深入理解多线程(一)——Synchronized的实现原理
    css
  • 原文地址:https://www.cnblogs.com/qiyebao/p/4205737.html
Copyright © 2011-2022 走看看