发布与订阅
在我们使用手机发送消息的时候，即可以选择给单个手机号码发送消息，也可以选择多个手机号码，群发消息。
前面学习工作队列的时候，我们使用的场景是一个消息只能被一个消费者程序实例接收并处理，但是如果想要群发消息，仅凭之前学到的东西是实现不了的。
所以这里需要引入RabbitMQ的发布与订阅模式。
Exchange
什么是Exchange?
RabbitMQ通信模型的核心思想是消息生产者不会直接发送消息到消息队列，生产者程序也不知道他产生的消息是否发送到了一个消息队列中。
实际上消息生产者只会发送消息到一个Exchange对象。Exchange是英语中交换的意思，这RabbitMQ中它实际就是一个消息的中转站，一方面它会接收所有与它绑定的生产者程序实例产生的消息，一方面它有会根据自身的定义将消息发送到不同的消息队列中。不同的Exchange类型决定了它会将消息发送到一个特定的消息队列，还是多个消息队列。
Exchange有四种类型
- direct
- topic
- headers
- fanout
RabbitMQ的发布和订阅模式会使用fanout类型的Exchange。
“fanout”在英语中是扇出、展开的意思，这种类型的Exchange会将收到的一条消息，发送到多个消息队列中，而不同的消息消费者实例监听不同的消息队列，从而实现消息广播的功能。
如何设置Exchange
我们可以使用channel对象中的ExchangeDeclare方法来声明Exchange
channel.ExchangeDeclare("logs", "fanout");
这个方法的第一个参数是Exchange的名称，第二个参数是Exchange的类型。
当我们调用channel对象的BasicPublish方法发送消息的时候，我们可以设置Exchange参数
public static void BasicPublish(this IModel model, string exchange, string routingKey, IBasicProperties basicProperties, byte[] body);
例：
channel.BasicPublish(exchange: "logs", routingKey: "", basicProperties: null, body: body);
默认Exchange
回想上一章工作队列的例子，会发现之前才发布消息的时候，我们有传递一个空 exchange
channel.BasicPublish(exchange: "", routingKey: "work_queue", basicProperties: properties, body: body );
那为什么消息会正确的发送到消息消费者程序实例呢？
这个其实是RabbitMQ的一个默认设置，当发送消息时，如果将exchange设置为空字符串，系统会自动启用一个默认的Exchange, 这个Exchange会把消息自动添加到第二个参数routeingKey指定的消息队列中（在上一章的例子中，即work_queue消息队列），所以消息消费者程序实例才能正确的接收消息。
临时消息队列
当我们群发消息的时候，我们需要为每个消息消费者实例创建一个名字独特的消息队列。
但是这里我们只是关注群发消息，而不想去关注为消息队列起名字这件事情。
RabbitMQ提供了一种临时消息队列，来帮助我们简化这个操作。
当我们在声明消息队列的时候，不传递任何参数，生成的就是一个临时消息队列
channel.QueueDeclare();
而如果想获得这个消息队列的名字，我们可以从它的QueueName属性中获取。
var queueName = channel.QueueDeclare().QueueName;
这个queueName是RabbitMQ帮我们动态随机生成，例：amq.gen-JzTY20BRgKO-HjmUJj0wLg.
Exchange和消息队列绑定
当我们创建了一个 Exchange之后，我们需要设置Exchange给那些消息队列发送消息，Exchange和消息队列之间的关系叫做Binding(绑定)
在channel对象中有一个QueueBind的方法来帮助我们完成这一操作
channel.QueueBind(queue: queueName, exchange: "logs", routingKey: "");
修改程序
现在我们修改一下我们之前的发送消息程序，将它改造成一个消息群发器。
Send
- 声明一个名为broadcast的Exchange
- 修改发送消息部分的代码，现在直接发送到创建的Exchange中
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
HostName = "localhost"
};
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare("broadcast", "fanout");
string message = args[0];
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "broadcast",
routingKey: "",
basicProperties: null,
body: body
);
Console.WriteLine("[x] Sent {0}", message);
}
}
}
Receive
- 使用临时消息队列
- 将Exchange和临时消息队列绑定
static void Main(string[] args)
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare("broadcast", "fanout");
var queueName = channel.QueueDeclare().QueueName;
channel.QueueBind(queue: queueName, exchange: "broadcast", routingKey: "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine("[x] Received {0}", message);
};
channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
Console.Read();
}
}
}
最终效果
我们启动1个Send程序，2个Receive程序。