RabbitMQ入门教程——发布/订阅

zlzyp

　　什么是发布订阅
　　发布订阅是一种设计模式定义了一对多的依赖关系，让多个订阅者对象同时监听某一个主题对象。这个主题对象在自身状态变化时，会通知所有的订阅者对象，使他们能够自动更新自己的状态。
　　为了描述这种模式，我们将会构建一个简单的日志系统。它包括两个程序——第一个程序负责发送日志消息，第二个程序负责获取消息并输出内容。在我们的这个日志系统中，所有正在运行的接收方程序都会接受消息。我们用其中一个接收者（receiver）把日志写入硬盘中，另外一个接受者（receiver）把日志输出到屏幕上。最终，日志消息被广播给所有的接受者（receivers）。
　　Exchanges
　　RabbitMQ消息模型的核心理念是生产者永远不会直接发送任何消息给队列，生产者只能发送消息给到exchange，exchange比较简单，一边从生产者就收消息，一边把消息推送到队列中。exchange必须清楚的知道消息应该按照什么规则路由到对应的队列中，而具体使用那种路由算法是由exchange type决定的。AMQP协议提供了四种交换机类型：

Name（交换机类型）	Default pre-declared names（预声明的默认名称）
Direct exchange（直连交换机）	(Empty string) and amq.direct
Fanout exchange（扇型交换机）	amq.fanout
Topic exchange（主题交换机）	amq.topic
Headers exchange（头交换机）	amq.match (and amq.headers in RabbitMQ)

　　除交换机类型外，在声明交换机时还可以附带许多其他的属性，其中最重要的几个分别是：

• 
  Name
  
• 
  Durability （消息代理重启后，交换机是否还存在）
  
• 
  Auto-delete （当所有与之绑定的消息队列都完成了对此交换机的使用后，删掉它）
  
• 
  Arguments（依赖代理本身）　　交换机可以有两个状态：持久（durable）、暂存（transient）。持久化的交换机会在消息代理（broker）重启后依旧存在，而暂存的交换机则不会（它们需要在代理再次上线后重新被声明）。然而并不是所有的应用场景都需要持久化的交换机。
  　　本文中具体讲解下以下两种交换机：直连交换机（前面几个例子中使用的交换机类型），扇形交换机（本文中要使用的交换机类型）
  　　直连交换机
  　　直连交换机（direct exchange）可以使用消息携带的路由键（routing key）将消息投递给对应的队列中。用来处理消息的单播路由（unicast routing），也可以处理多播路由。
  　　那么它具体是如何工作的呢
  
  
  
  • 
    将一个队列绑定到某个交换机上，同时给该绑定指定一个路由键（routing key）
    
  • 
    当一个携带路由键为R的消息被发送到直连交换机时，交换机会把它路由给绑定值同样为R的队列。
    
  　　直连交换机经常用来循环分发任务给多个工作者，当这样做时，一定要明白，这时消息的负载均衡是发生在消费者（consumer）之间的，而不是队列（queue）中。
  　　直连交换机图例：
  
  
  
  　　扇形交换机
  　　扇形交换机（funout exchange）将消息路由给绑定到它身上的所有队列，不关心所绑定的路由键（routing key）。扇形交换机用来处理消息的广播路由（broadcast routing）。
  　　由于扇形交换机投递消息到所有绑定他的队列，以下几个场景比较适合使用扇形交换机：
  
  
  
  • 
    大规模多用户在线（MMO）游戏可以使用它来处理排行榜更新等全局事件
    
  • 
    体育新闻网站可以用它来近乎实时地将比分更新分发给移动客户端
    
  • 
    分发系统使用它来广播各种状态和配置更新
    
  • 
    在群聊的时候，它被用来分发消息给参与群聊的用户。（AMQP没有内置presence的概念，因此XMPP可能会是个更好的选择）
    
  　　扇形交换机图例
  
  
  
  　　创建exchange
  　　                    channel.ExchangeDeclare(exchange: "log_exchange", //exchange 名称
  　　                        type: ExchangeType.Fanout, //exchange 类型
  　　                        durable: false,
  　　                        autoDelete: false,
  　　                        arguments: null);
  　　临时队列
  　　之前的几个示例中我们在为每一个声名的队列都指定了一个名字，因为我们希望consumer指向正确的队列。当我们希望在生产者和消费者之间共享队列时，为队列命名就非常的重要了。
  　　不过我们要实现的日志系统只是想要得到所有的消息，而且只对当前正在传递的消息感兴趣，并不关心队列的名称，所以为了满足我们的需求，要做两件事情：
  　　无论什么时间连接到RabbitMQ我们都需要一个新的空的队列。为了达到目的我们可以使用随机数创建队列，或让服务器给我们提供一个随机的名称。
  　　一旦消费者与RabbitMQ断开，消费者所接受的队列都应该被自动删除。
  　　创建临时队列
  　　                    //创建一个未命名的新的消息队列，
  　　                    QueueDeclareOk queue = channel.QueueDeclare(queue: "", //队列名称，为空时有系统自动分配
  　　                        durable: false,
  　　                        exclusive: false,
  　　                        autoDelete: true,//自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。
  　　                        arguments: null);
  　　                    //或
  　　                    //queue = channel.QueueDeclare();
  　　绑定
  　　我们已经创建了一个扇型交换机（fanout）和一个队列。现在我们需要告诉交换机如何发送消息给我们的队列。交换器和队列之间的联系我们称之为绑定（binding）
  　　创建交换机与队列的关系
  　　//扇形交换机（funout exchange）将消息路由给绑定到它身上的所有队列，不关心所绑定的路由键（routing key）
  　　                    //fanout exchange不需要指定routing key 指定了也没用
  　　                    //通过绑定告诉exchange 需要发送消息到哪些消息队列
  　　                    channel.QueueBind(queue: queueName, exchange: EXCHANGE_NAME, routingKey: ROUTING_KEY, arguments: null);
  　　完整代码：
  　　生产者  Pub_SubProducer.cs
  　　using System;
  　　using System.Collections.Generic;
  　　using System.Linq;
  　　using System.Text;
  　　using System.Threading.Tasks;
  　　using RabbitMQ.Client;
  　　namespace RabbitMQProducer
  　　{
  　　    public class Pub_SubProducer
  　　    {
  　　        const string EXCHANGE_NAME = "log_exchange";
  　　        const string ROUTING_KEY = "";
  　　        //直接发送消息到交换机
  　　        public static void Publish()
  　　        {
  　　            var factory = new ConnectionFactory()
  　　            {
  　　                HostName = "127.0.0.1"
  　　            };
  　　            using (var connection = factory.CreateConnection())
  　　            {
  　　                using (IModel channel = connection.CreateModel())
  　　                {
  　　                    channel.ExchangeDeclare(exchange: EXCHANGE_NAME, //exchange 名称
  　　                        type: ExchangeType.Fanout, //exchange 类型
  　　                        durable: false,
  　　                        autoDelete: false,
  　　                        arguments: null);
  　　                    Parallel.For(1, 100, item =>
  　　                    {
  　　                        string message = $"日志内容{DateTime.Now.ToString()}";
  　　                        channel.BasicPublish(exchange: EXCHANGE_NAME, routingKey: ROUTING_KEY, basicProperties: null, body: Encoding.UTF8.GetBytes(message));
  　　                        Console.WriteLine(message);
  　　                    });
  　　                    Console.WriteLine(" Press [enter] to exit.");
  　　                    Console.ReadLine();
  　　                }
  　　            }
  　　        }
  　　    }
  　　}
  　　消费者 Pub_SubConsumer.cs
  　　using RabbitMQ.Client;
  　　using System;
  　　using System.Collections.Generic;
  　　using System.Linq;
  　　using System.Text;
  　　using System.Threading.Tasks;
  　　using RabbitMQ.Client.Events;
  　　using System.IO;
  　　namespace RabbitMQConsumer
  　　{
  　　    public class Pub_SubConsumer
  　　    {
  　　        const string EXCHANGE_NAME = "log_exchange";
  　　        const string ROUTING_KEY = "";
  　　        //输出到屏幕
  　　        public static void Subscribe()
  　　        {
  　　            var factory = new ConnectionFactory()
  　　            {
  　　                HostName = "127.0.0.1"
  　　            };
  　　            using (var connection = factory.CreateConnection())
  　　            {
  　　                using (IModel channel = connection.CreateModel())
  　　                {
  　　                    channel.ExchangeDeclare(exchange: EXCHANGE_NAME, //exchange 名称
  　　                        type: ExchangeType.Fanout, //exchange 类型
  　　                        durable: false,
  　　                        autoDelete: false,
  　　                        arguments: null);
  　　                    //创建一个未命名的新的消息队列，
  　　                    QueueDeclareOk queue = channel.QueueDeclare(queue: "", //队列名称，为空时有系统自动分配
  　　                        durable: false,
  　　                        exclusive: false,
  　　                        autoDelete: true,//自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。
  　　                        arguments: null);
  　　                    //或
  　　                    //queue = channel.QueueDeclare();
  　　                    string queueName = queue.QueueName;
  　　                    //扇形交换机（funout exchange）将消息路由给绑定到它身上的所有队列，不关心所绑定的路由键（routing key）
  　　                    //fanout exchange不需要指定routing key 指定了也没用
  　　                    //通过绑定告诉exchange 需要发送消息到哪些消息队列
  　　                    channel.QueueBind(queue: queueName, exchange: EXCHANGE_NAME, routingKey: ROUTING_KEY, arguments: null);
  　　                    EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
  　　                    consumer.Received += (sender, args) =>
  　　                    {
  　　                        string message = Encoding.UTF8.GetString(args.Body);
  　　                        Console.WriteLine(message);
  　　                    };
  　　                    channel.BasicConsume(queue: queueName, noAck: true, consumer: consumer);
  　　                    Console.WriteLine(" Press [enter] to exit.");
  　　                    Console.ReadLine();
  　　                }
  　　            }
  　　        }
  　　        /// <summary>
  　　        /// 输出到文件
  　　        /// </summary>
  　　        public static void SubscribeFile()
  　　        {
  　　            var factory = new ConnectionFactory()
  　　            {
  　　                HostName = "127.0.0.1"
  　　            };
  　　            using (var connection = factory.CreateConnection())
  　　            {
  　　                using (IModel channel = connection.CreateModel())
  　　                {
  　　                    channel.ExchangeDeclare(exchange: EXCHANGE_NAME, //exchange 名称
  　　                        type: ExchangeType.Fanout, //exchange 类型
  　　                        durable: false,
  　　                        autoDelete: false,
  　　                        arguments: null);
  　　                    //创建一个未命名的新的消息队列，
  　　                    QueueDeclareOk queue = channel.QueueDeclare(queue: "", //队列名称，为空时有系统自动分配
  　　                        durable: false,
  　　                        exclusive: false,
  　　                        autoDelete: true,//自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。
  　　                        arguments: null);
  　　                    //或
  　　                    //queue = channel.QueueDeclare();
  　　                    string queueName = queue.QueueName;
  　　                    //扇形交换机（funout exchange）将消息路由给绑定到它身上的所有队列，不关心所绑定的路由键（routing key）
  　　                    //fanout exchange不需要指定routing key 指定了也没用
  　　                    //通过绑定告诉exchange 需要发送消息到哪些消息队列
  　　                    channel.QueueBind(queue: queueName, exchange: EXCHANGE_NAME, routingKey: ROUTING_KEY, arguments: null);
  　　                    EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
  　　                    consumer.Received += (sender, args) =>
  　　                    {
  　　                        string message = Encoding.UTF8.GetString(args.Body);
  　　                        //写入日志到txt文件
  　　                        using (StreamWriter writer = new StreamWriter(@"c:\log\log.txt", true, Encoding.UTF8))
  　　                        {
  　　                            writer.WriteLine(message);
  　　                            writer.Close();
  　　                        }
  　　                        Console.WriteLine(message);
  　　                    };
  　　                    channel.BasicConsume(queue: queueName, noAck: true, consumer: consumer);
  　　                    Console.WriteLine(" Press [enter] to exit.");
  　　                    Console.ReadLine();
  　　                }
  　　            }
  　　        }
  　　    }
  　　}
  　　运行以上实例代码发现，每个订阅者实例 都能得到相同的内容。