RabbitMQ入门教程——发布/订阅
Posted 一个小码农
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ入门教程——发布/订阅相关的知识,希望对你有一定的参考价值。
发布订阅是一种设计模式定义了一对多的依赖关系,让多个订阅者对象同时监听某一个主题对象。这个主题对象在自身状态变化时,会通知所有的订阅者对象,使他们能够自动更新自己的状态。
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) |
除交换机类型外,在声明交换机时还可以附带许多其他的属性,其中最重要的几个分别是:
本文中具体讲解下以下两种交换机:直连交换机(前面几个例子中使用的交换机类型),扇形交换机(本文中要使用的交换机类型)
直连交换机(direct exchange)可以使用消息携带的路由键(routing key)将消息投递给对应的队列中。用来处理消息的单播路由(unicast routing),也可以处理多播路由。
直连交换机经常用来循环分发任务给多个工作者,当这样做时,一定要明白,这时消息的负载均衡是发生在消费者(consumer)之间的,而不是队列(queue)中。
扇形交换机(funout exchange)将消息路由给绑定到它身上的所有队列,不关心所绑定的路由键(routing key)。扇形交换机用来处理消息的广播路由(broadcast routing)。
由于扇形交换机投递消息到所有绑定他的队列,以下几个场景比较适合使用扇形交换机:
channel.ExchangeDeclare(exchange: "log_exchange", //exchange 名称
type: ExchangeType.Fanout, //exchange 类型
之前的几个示例中我们在为每一个声名的队列都指定了一个名字,因为我们希望consumer指向正确的队列。当我们希望在生产者和消费者之间共享队列时,为队列命名就非常的重要了。
不过我们要实现的日志系统只是想要得到所有的消息,而且只对当前正在传递的消息感兴趣,并不关心队列的名称,所以为了满足我们的需求,要做两件事情:
无论什么时间连接到RabbitMQ我们都需要一个新的空的队列。为了达到目的我们可以使用随机数创建队列,或让服务器给我们提供一个随机的名称。
一旦消费者与RabbitMQ断开,消费者所接受的队列都应该被自动删除。
QueueDeclareOk queue = channel.QueueDeclare(queue: "", //队列名称,为空时有系统自动分配
autoDelete: true,//自动删除,如果该队列没有任何订阅的消费者的话,该队列会被自动删除。这种队列适用于临时队列。
//queue = channel.QueueDeclare();
我们已经创建了一个扇型交换机(fanout)和一个队列。现在我们需要告诉交换机如何发送消息给我们的队列。交换器和队列之间的联系我们称之为绑定(binding)
//扇形交换机(funout exchange)将消息路由给绑定到它身上的所有队列,不关心所绑定的路由键(routing key)
//fanout exchange不需要指定routing key 指定了也没用
//通过绑定告诉exchange 需要发送消息到哪些消息队列
using System.Collections.Generic;
const string EXCHANGE_NAME = "log_exchange";
const string ROUTING_KEY = "";
var factory = new ConnectionFactory()
using (var connection = factory.CreateConnection())
using (IModel channel = connection.CreateModel())
channel.ExchangeDeclare(exchange: EXCHANGE_NAME, //exchange 名称
type: ExchangeType.Fanout, //exchange 类型
string message = $"日志内容{DateTime.Now.ToString()}";
Console.WriteLine(" Press [enter] to exit.");
using System.Collections.Generic;
const string EXCHANGE_NAME = "log_exchange";
const string ROUTING_KEY = "";
public static void Subscribe()
var factory = new ConnectionFactory()
using (var connection = factory.CreateConnection())
using (IModel channel = connection.CreateModel())
channel.ExchangeDeclare(exchange: EXCHANGE_NAME, //exchange 名称
type: ExchangeType.Fanout, //exchange 类型
QueueDeclareOk queue = channel.QueueDeclare(queue: "", //队列名称,为空时有系统自动分配
autoDelete: true,//自动删除,如果该队列没有任何订阅的消费者的话,该队列会被自动删除。这种队列适用于临时队列。
//queue = channel.QueueDeclare();
string queueName = queue.QueueName;
//扇形交换机(funout exchange)将消息路由给绑定到它身上的所有队列,不关心所绑定的路由键(routing key)
//fanout exchange不需要指定routing key 指定了也没用
//通过绑定告诉exchange 需要发送消息到哪些消息队列
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
consumer.Received += (sender, args) =>
string message = Encoding.UTF8.GetString(args.Body);
channel.BasicConsume(queue: queueName, noAck: true, consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
public static void SubscribeFile()
var factory = new ConnectionFactory()
using (var connection = factory.CreateConnection())
using (IModel channel = connection.CreateModel())
channel.ExchangeDeclare(exchange: EXCHANGE_NAME, //exchange 名称
type: ExchangeType.Fanout, //exchange 类型
QueueDeclareOk queue = channel.QueueDeclare(queue: "", //队列名称,为空时有系统自动分配
autoDelete: true,//自动删除,如果该队列没有任何订阅的消费者的话,该队列会被自动删除。这种队列适用于临时队列。
//queue = channel.QueueDeclare();
string queueName = queue.QueueName;
//扇形交换机(funout exchange)将消息路由给绑定到它身上的所有队列,不关心所绑定的路由键(routing key)
//fanout exchange不需要指定routing key 指定了也没用
//通过绑定告诉exchange 需要发送消息到哪些消息队列
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
consumer.Received += (sender, args) =>
string message = Encoding.UTF8.GetString(args.Body);
以上是关于RabbitMQ入门教程——发布/订阅的主要内容,如果未能解决你的问题,请参考以下文章