RabbitMQ(五)——发布订阅模式
前言
上一章的工作队列模式中,生产者发布的一堆消息进入队列,消费者接收队列中的消息,每条消息只能发给一个消费者。
本章要做的是吧一条消息发送给多个消费者,这种模式就是Fanout Exchange(扇形交换机)“发布/订阅模式”,它会将消息路由给绑定到它身上的所有队列。
注意:该模式没有存储消息功能,如果消息发送到没有绑定消费队列的交换器,消息丢失,也就是说,当消息生产者发送消息时,消费者还没有绑定此交换器,则没有接收到消息,且接收不到了。
交换器:
在RabbitMQ中完整的消息传递并不是生产者直接将消息发送到队列,生产者甚至不不知道消息是否会进入队列。正确的模型是生产者把消息发送给交换器(Exchange),交换器会将接受到的消息转发到队列中,交换器必须确定将消息发送到哪些队列。

交换器类型(详情):
- Fanout
- Direct
- Topic
- Header
fanout类型即发布订阅模式,它会把收到的消息广播到它绑定的队列中。
channel.ExchangeDeclare("exchange", "fanout");
之前我们发布的做法是这样的:
channel.BasicPublish("", "simple", null, Encoding.UTF8.GetBytes(msg));前面我们并没有声明交换器,之所以还能把消息发送到队列是因为用了“”空的字符串标识了默认或匿名交换器。BasicPublish方法第一个参数就是交换器,第二个参数是routekey,消息通过本交换器进入到routekey队列。
发送消息到指定交换器:
channel.BasicPublish("exchange", "simple", null, Encoding.UTF8.GetBytes(msg));
创建好fanout交换器后,定义多个队列,然后将交换器和队列绑定:
channel.QueueBind("simple", "exchange", ""); 
代码
生产者:
static void Main(string[] args)
{
Console.WriteLine("FanoutServer发布服务器启动...");
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "guest",
Password = "guest"
};
//2.创建连接
using (var connection = factory.CreateConnection())
//3.创建通道
using (var channel = connection.CreateModel())
{
//4.创建交换器
channel.ExchangeDeclare("exchange", "fanout");
string msg = "";
for (int i = 0; i < 100; i++)
{
msg = $"发布消息{i}";
var body = Encoding.UTF8.GetBytes(msg);
channel.BasicPublish("exchange", "", null, body);
Console.WriteLine($"发布成功:{msg}");
Thread.Sleep(600);
}
Console.ReadKey();
}
}消费者:
static void Main(string[] args)
{
Console.WriteLine("FanoutClient接收客户端启动...");
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "guest",
Password = "guest"
};
//2.创建连接
using (var connection = factory.CreateConnection())
{
//3.创建通道
using (var channel = connection.CreateModel())
{
//4.定义交换器
channel.ExchangeDeclare("exchange", "fanout");
//5.创建匿名队列,绑定交换器
//var queueName = channel.QueueDeclare("simple");
var queueName = channel.QueueDeclare().QueueName;
channel.QueueBind(queueName, "exchange", "");
//6.创建消费者
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
//接收消息
var body = Encoding.UTF8.GetString(ea.Body.ToArray());
Console.WriteLine($"接收消息:{body.ToString()}");
};
//7.消费消息
channel.BasicConsume(queueName, true, consumer);
Console.ReadKey();
}
}
}生产者创建完通道后,创建一个交换器,给它命名“exchange”以及设定“fanout”类型,最后发布消息到此交换器,与前面的模式不同的是不用设定队列参数。
消费者创建完通道后,创建一个交换器,与生产者的交换器一致,定义一个队列(可以指定,或者使用匿名),然后将此队列与交换器绑定,消息会在此队列中消费,最后创建消费者进行消费。
效果
先启动2个消费者,再启动生产者,消息完全接受到。

先启动生产者,再分别启动消费者,只能接受到消费者启动之后生产发送的消息

发布订阅模式就先到这,有什么不对的地方望斧正~
