发布与订阅

在我们使用手机发送消息的时候，即可以选择给单个手机号码发送消息，也可以选择多个手机号码，群发消息。

前面学习工作队列的时候，我们使用的场景是一个消息只能被一个消费者程序实例接收并处理，但是如果想要群发消息，仅凭之前学到的东西是实现不了的。

所以这里需要引入RabbitMQ的发布与订阅模式。

Exchange

什么是Exchange?

RabbitMQ通信模型的核心思想是消息生产者不会直接发送消息到消息队列，生产者程序也不知道他产生的消息是否发送到了一个消息队列中。

实际上消息生产者只会发送消息到一个Exchange对象。Exchange是英语中交换的意思，这RabbitMQ中它实际就是一个消息的中转站，一方面它会接收所有与它绑定的生产者程序实例产生的消息，一方面它有会根据自身的定义将消息发送到不同的消息队列中。不同的Exchange类型决定了它会将消息发送到一个特定的消息队列，还是多个消息队列。

 

 

Exchange有四种类型

directtopicheadersfanout

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程序。

转载于:https://www.cnblogs.com/lwqlun/p/9095122.html