在上一章中,我们创建了一个工作队列,工作队列模式的设想是每一条消息只会被转发给一个消费者。本章将会讲解完全不一样的场景: 我们会把一个消息转发给多个消费者,这种模式称之为发布-订阅模式。

  为了阐述这个模式,我们将会搭建一个简单的日志系统,它包含两种程序:一种发送日志消息,另一种接收并打印日志消息。在这个日志系统里,每一个运行的消费者都可以获取到消息,在这种情况下,我们可以实现这种需求:一个消费者接收消息并写入磁盘,另一个消费者接收消息并打印在电脑屏幕上。简单来说,生产者发布的消息将会以广播的形式转发到所有的消费者。

1、交换器(Exchange)

  在前两章节我们,我们往队列中发布消息或获取消息,然而,前面的讲解其实并不完整,接下来,是时候介绍完整的RabbitMq消息模型了。

  回忆一下我们前两章指南中包含的内容:

    • 一个生产者用以发送消息;
    • 一个队列缓存消息;
    • 一个消费者用以消费队列中的消息。

  RabbitMq消息模式的核心思想是:一个生产者并不会直接往一个队列中发送消息,事实上,生产者根本不知道它发送的消息将被转发到哪些队列。

  实际上,生产者只能把消息发送给一个exchange,exchange只做一件简单的事情:一方面它们接收从生产者发送过来的消息,另一方面,它们把接收到的消息推送给队列。一个exchage必须清楚地知道如何处理一条消息。

  

  有四种类型的交换器,分别是:direct、topic、headers、fanout。本章主要讲解最后一种:fanous(广播模式)。下面创建一个fanout类型的交换器,我们称之为:logs:

1 channel.exchangeDeclare("logs", "fanout");

  广播模式交换器很简单,从字面意思也能理解,它其实就是把接收到的消息推送给所有它知道的队列。在我们的日志系统中正好需要这种模式。

  如果想查看当前系统中有多少个exchange,可以使用以下命令:

sudo rabbitmqctl list_exchanges

  

  或者通过控制台查看:

  

  可以看到有很多以amq.*开头的交换器,以及(AMQP default)默认交换器,这些是默认创建的交换器。

   在前面两章的指南中,我们并不知道交换器的存在,但是依然可以将消息发送到队列中,那其实并不是因为我们可以不使用交换器,实际上是我们使用了默认的交换器(我们通过指定交换器为字字符串:""),回顾一下我们之前是如何发送消息的:

1 channel.basicPublish("", "hello", null, message.getBytes());

  第一个参数是交换器的名字,空字符串表示它是一个默认或无命名的交换器,消息将会由指定的路由键(第二个参数,routingKey,后面会讲)转发到队列。

  你可能会有疑问:既然exchange可以指定为空字符串(""),那么可否指定为null?

  答案是:不能!

  通过跟踪发布消息的代码,在AMQImpl类中的Publish()方面中,可以看到,不光是exchange不能为null,同时routingKey路由键也不能为null,否则会抛出异常:

  接着上面的讲解,我们创建一个命名的交换器:

1 channel.basicPublish( "logs", "", null, message.getBytes());

2、临时队列

  在前两章的例子中,我们使用的队列都是有具体的队列名,创建命名队列是很必要的,因为我们需要将消费者指向同一名字的队列。因此,要想在生产者和消费者中间共享队列就必须要使用命名队列。

  但是,本章讲解的日志系统也可以使用非命名队列(可以不手动命名),我们希望收到所有日志消息,而不是部分。并且我们希望总是接收到新的日志消息而不是旧的日志消息。为了解决这个问题,需要分两步走。

  首先,无论何时我们的消费者连接到RabbitMq,我们都需要一个新的、空的队列来接收日志消息,因此,消费者在连接上RabbitMq之后需要创建一个任意名字的队列,或者让RabbitMq生成任意的队列名字。

  其次,一旦该消费者断开了与RabbitMq的连接,队列也被自动删除。

  通过JAVA客户端的无参方法:queueDeclare()来创建一个非持久化、专有的、自动删除的、名字随机生成的队列。

1 String queueName = channel.queueDeclare().getQueue();

3、绑定(Binding)

  

  前面广播模式的交换器和队列已经创建好了,接下来就是要告诉交换器向队列里发送消息。交换器与队列之间的关系称之为绑定关系。

1 channel.queueBind(queueName, "logs", "");

  至此,交换器已经可以往队列中发送消息了。

  可以通过下列命令来查看队列的绑定关系:

4、完整的代码

  EmitLog.java

 1 import com.rabbitmq.client.BuiltinExchangeType; 
 2 import com.rabbitmq.client.Channel; 
 3 import com.rabbitmq.client.Connection; 
 4 import com.rabbitmq.client.ConnectionFactory; 
 5  
 6 public class EmitLog { 
 7  
 8     private static final String EXCHANGE_NAME = "logs"; 
 9  
10     public static void main(String[] args) throws Exception { 
11  
12         ConnectionFactory factory = new ConnectionFactory(); 
13         factory.setHost("192.168.92.130"); 
14  
15         try (Connection connection = factory.newConnection(); 
16              Channel channel = connection.createChannel();) { 
17  
18             channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); 
19  
20             String message = "RabbitMq fanout。。。。。。"; 
21             channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("utf-8")); 
22  
23             System.out.println(" [x] Sent '" + message + "'"); 
24         } 
25     } 
26 }

  正好你所看到的,Connection创建完成之后,定义了exchange,这一步是必要的,因为如果没有交换器将无法发送消息。

  如此没有队列绑定到该交换器上,那么,交换器收到的消息将会丢失,但是对我们本章的日志系统来说没问题的,当没有消费者时,我们可以安全地放弃掉数据,我们只接收最新的日志消息。

  ReceiveLogs.java

 1 public class ReceiveLogs { 
 2  
 3     private static final String EXCHANGE_NAME = "logs"; 
 4  
 5     public static void main(String[] args) throws Exception { 
 6  
 7         ConnectionFactory factory = new ConnectionFactory(); 
 8         factory.setHost("192.168.92.130"); 
 9  
10         Connection connection = factory.newConnection(); 
11         Channel channel = connection.createChannel(); 
12  
13         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); 
14  
15         final String queue = channel.queueDeclare().getQueue(); 
16         channel.queueBind(queue,EXCHANGE_NAME,""); 
17  
18         System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); 
19  
20         DeliverCallback deliverCallback = (consumerTa,delivery) -> { 
21  
22             String message = new String(delivery.getBody(), "UTF-8"); 
23             System.out.println(" [x] Received '" + message + "'"); 
24  
25         }; 
26  
27         channel.basicConsume(queue,true,deliverCallback,consumerTag -> {}); 
28     } 
29 }

  这里的autoAck设置为true,因为我们这里是广播模式,每个消费者都会收到一样的消息,并且这里给消费者生产的随机名称的队列相当于是独有的,所以在接收到消息之后立即发送确认回执是OK的。

  但是这里先提出一个疑问:在这种模式下,每个队列收到的消息是否也会有Ready和Unacked状态?

5、测试结果

  一、首先启动生产者,再启动两个消费者

  

       

  可以看到,生产者启动后发送的消息丢失了,两个消费者并没有消费到,此时再看控制台:

  

  可见RabbitMq为我们创建了两个随机命名的队列,其Exclusive是Owner,表示是专有的,Parameters为AD(auto delete),拥有该队列的消费者一占断开连接,队列将会被自动删除。

  二、其次启动生产者发送一次消息

   

  两个消费都都收到了消息。

  三、关闭所有消费者,观察控制台变化

  

  两个专有随机队列自动删除了。

6、SpringBoot的实现

  工程结构图:

 一、配置文件application.properties:

  生产者:

#RabbitMq 
spring.rabbitmq.host=192.168.92.130 
spring.rabbitmq.exchange=logs

  消费者:

#RabbitMq 
spring.rabbitmq.host=192.168.92.130 
spring.rabbitmq.exchange=logs 
 
##队列--我们可以自己指定队列名称,也可以由RabbitMq自动生成,这里为了方便,我们自己命名(如果需要,我也可以写一个自动生成名称的方法) 
rqbbitmq.log.fanout.info=info 
rqbbitmq.log.fanout.error=error
server.port=8090

二、生产者代码

  这里为了让系统生产者启动时就自动发送一条消息,我加了一个EmitLogRunner类。

  EmitLog.java

 1 import org.springframework.amqp.core.AmqpTemplate; 
 2 import org.springframework.beans.factory.annotation.Autowired; 
 3 import org.springframework.beans.factory.annotation.Value; 
 4 import org.springframework.stereotype.Component; 
 5  
 6 @Component 
 7 public class EmitLog { 
 8  
 9     @Value("${spring.rabbitmq.exchange}") 
10     private String exchange; 
11  
12     @Autowired 
13     private AmqpTemplate amqpTemplate; 
14  
15     public void send(String msg) { 
16         amqpTemplate.convertAndSend(exchange,"",msg); 
17     } 
18 }

  EmitLogRunner.java

 1 import org.springframework.beans.factory.annotation.Autowired; 
 2 import org.springframework.boot.ApplicationArguments; 
 3 import org.springframework.boot.ApplicationRunner; 
 4 import org.springframework.stereotype.Component; 
 5  
 6 @Component 
 7 public class EmitLogRunner implements ApplicationRunner { 
 8  
 9     @Autowired 
10     private EmitLog emitLog; 
11  
12     @Override 
13     public void run(ApplicationArguments args) throws Exception { 
14         System.out.println("生产者发布消息:" + msg); 
15         emitLog.send("RabbitMq fanout test message"); 
16     } 
17 }

二、消费者代码  

  ReceiveInfoLogs.java

 1 @Component 
 2 @RabbitListener( 
 3         bindings = @QueueBinding( 
 4                 value = @Queue(value = "${rqbbitmq.log.fanout.info}",autoDelete = "true"), 
 5                 exchange = @Exchange(value = "${spring.rabbitmq.exchange}",type = ExchangeTypes.FANOUT) 
 6         ) 
 7 ) 
 8 public class ReceiveInfoLogs { 
 9  
10     @Autowired 
11     private AmqpTemplate amqpTemplate; 
12  
13     @RabbitHandler 
14     public void receiveInfoLog (Object message) { 
15   
16         System.out.println("接收到info级别的日志:" + message); 
17     } 
18 }

  ReceiveErrorLogs.java

 1 import org.springframework.amqp.core.AmqpTemplate; 
 2 import org.springframework.amqp.core.ExchangeTypes; 
 3 import org.springframework.amqp.rabbit.annotation.*; 
 4 import org.springframework.beans.factory.annotation.Autowired; 
 5 import org.springframework.stereotype.Component; 
 6  
 7 @Component 
 8 @RabbitListener( 
 9         bindings = @QueueBinding( 
10                 value = @Queue(value = "${rqbbitmq.log.fanout.error}",autoDelete = "true"), 
11                 exchange = @Exchange(value = "${spring.rabbitmq.exchange}",type = ExchangeTypes.FANOUT) 
12         ) 
13 ) 
14 public class ReceiveErrorLogs { 
15  
16     @Autowired 
17     private AmqpTemplate amqpTemplate; 
18  
19     @RabbitHandler 
20     public void receiveErrorLog(Object message) { 
21         System.out.println("接收到的error级别日志:" + message); 
22     } 
23 }

  注意看一下注解方式bindings里面都是以@开头并加上对应的要绑定的项,琢磨一下应该都能理解。

三、验证

  启动消费者和生产者,查看控制台:

    

  至此,发布订阅模式讲解完了,在下一章中将会讲解Routing(路由)的概念。


发布评论
IT序号网

微信公众号号:IT虾米 (左侧二维码扫一扫)欢迎添加!

RabbitMQ指南之二:工作队列(Work Queues)知识解答
你是第一个吃螃蟹的人
发表评论

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。