当前位置:   article > 正文

【RabbitMQ笔记04】消息队列RabbitMQ七种模式之发布订阅模式(Publish/Subscribe)_面试 消息队列 发布订阅模式

面试 消息队列 发布订阅模式

这篇文章,主要介绍消息队列RabbitMQ七种模式之发布订阅模式(Publish/Subscribe)。

目录

一、发布订阅模式

1.1、Exchange交换机

(1)什么是Exchange交换机呢???

(2)Exchange如何知道消息应该分发到哪个Queue里面?

(3)Exchange分类

1.2、发布订阅模式介绍

1.3、案例代码

(1)引入依赖

(2)编写生产者

(3)编写消费者


一、发布订阅模式

1.1、Exchange交换机

前面介绍的两个模式下,生产者都是直接和Queue消息队列交互的,这种方式存在一个问题,那就是如果要向不同的Queue消息队列里面发送数据,还得重新写一个生产者出来,为了解决这个问题,RabbitMQ引入了Exchange交换机。

(1)什么是Exchange交换机呢???

Exchange交换机,它主要作用就是接收生产者发送的消息,并且根据消息的RouteKey将消息映射到不同的Queue消息队列里面,通过这种方式,就可以实现不同的消息分发到不同的Queue消息队列里面,并且这种模式下,生产者是不会直接和Queue消息队列交互的,生产者也不需要知道消息应该放入哪个队列,它只要告诉Exchange即可,剩下的事情都是Exchange来处理。

(2)Exchange如何知道消息应该分发到哪个Queue里面?

为了能够让Exchange知道某一条消息应该分发到哪个消息队列里面,RabbitMQ需要给Exchange和Queue之间添加映射关系,这个关系叫做:Binding(绑定)。虽然绑定在一起了,但是还是区分不了哪个消息应该分发到哪个Queue队列里面,所以还需要一个唯一标识,这个标识就是RouteKey路由键,也可以认为是Queue的唯一标识。采用上面这种模式,大致的逻辑图就如下所示啦:

(3)Exchange分类

RabbitMQ中有四种常见的Exchange交换机类型,分别是:

  • direct:直接模式。
  • topic:主题模式,后面介绍的Topic模式就是利用这个Exchange类型。
  • fanout:广播模式,这是发布订阅模式要使用的Exchange类型。
  • headers:不常用。

1.2、发布订阅模式介绍

RabbitMQ中,发布订阅模式就是指定Exchange交换机的类型是【fanout】,然后RabbitMQ就会将消息分发到所有和这个Exchange交换机绑定的Queue消息队列里面,此时所有的消费者都可以接收到这一条消息。

注意:对于发布订阅模式来说,消息的路由键RouteKey是没有作用的,可以不写。

1.3、案例代码

(1)引入依赖

  1. <!-- 引入 RabbitMQ 依赖 -->
  2. <dependency>
  3. <groupId>com.rabbitmq</groupId>
  4. <artifactId>amqp-client</artifactId>
  5. <version>5.16.0</version>
  6. </dependency>

(2)编写生产者

  • 因为这里采用的Exchange交换机来分发消息,所以生产者中只需要将消息发送到Exchange交换机里面即可,而不需要指定具体的Queue队列。
  1. package com.rabbitmq.demo.pub;
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  5. /**
  6. * @version 1.0.0
  7. * @Date: 2023/2/25 16:23
  8. * @Copyright (C) ZhuYouBin
  9. * @Description: 消息生产者
  10. */
  11. public class Producer {
  12. public static void main(String[] args) {
  13. // 1、创建连接工厂
  14. ConnectionFactory factory = new ConnectionFactory();
  15. // 2、设置连接的 RabbitMQ 服务地址
  16. factory.setHost("127.0.0.1"); // 默认就是本机
  17. factory.setPort(5672); // 默认就是 5672 端口
  18. // 3、获取连接
  19. Connection connection = null; // 连接
  20. Channel channel = null; // 通道
  21. try {
  22. connection = factory.newConnection();
  23. // 4、获取通道
  24. channel = connection.createChannel();
  25. // 5、声明 Exchange,如果不存在,则会创建
  26. String exchangeName = "exchange_demo_2023";
  27. channel.exchangeDeclare(exchangeName, "fanout");
  28. // 6、发送消息
  29. String message = "这是发布订阅模式,发送的消息数据";
  30. channel.basicPublish(exchangeName, "", null, message.getBytes());
  31. } catch (Exception e) {
  32. e.printStackTrace();
  33. } finally {
  34. if (null != channel) {
  35. try {
  36. channel.close();
  37. } catch (Exception e) {}
  38. }
  39. if (null != connection) {
  40. try {
  41. connection.close();
  42. } catch (Exception e) {}
  43. }
  44. }
  45. }
  46. }

(3)编写消费者

  • 消费者中,需要创建Queue队列,并且将这个Queue绑定到对应的Exchange上面(不同的消费者可以创建不同的队列,队列不同照样可以接收到订阅的消息)。
  1. package com.rabbitmq.demo.pub;
  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. /**
  5. * @version 1.0.0
  6. * @Date: 2023/2/25 16:30
  7. * @Copyright (C) ZhuYouBin
  8. * @Description: 消息消费者
  9. */
  10. public class Consumer {
  11. public static void main(String[] args) {
  12. // 1、创建连接工厂
  13. ConnectionFactory factory = new ConnectionFactory();
  14. // 2、设置连接的 RabbitMQ 服务地址
  15. factory.setHost("127.0.0.1"); // 默认就是本机
  16. factory.setPort(5672); // 默认就是 5672 端口
  17. // 3、获取连接
  18. Connection connection = null; // 连接
  19. Channel channel = null; // 通道
  20. try {
  21. connection = factory.newConnection();
  22. // 4、获取通道
  23. channel = connection.createChannel();
  24. // 5、声明 Exchange,如果不存在,则会创建
  25. String exchangeName = "exchange_demo_2023";
  26. channel.exchangeDeclare(exchangeName, "fanout");
  27. // 6、指定需要操作的消息队列,如果队列不存在,则会创建
  28. String queueName = "queue_demo_2023";
  29. channel.queueDeclare(queueName, false, false, false, null);
  30. // 7、绑定 Exchange 和 Queue
  31. channel.queueBind(queueName, exchangeName, "");
  32. // 8、消费消息
  33. DeliverCallback callback = new DeliverCallback() {
  34. public void handle(String s, Delivery delivery) throws IOException {
  35. // 接收消息
  36. System.out.println("这是接收的消息:" + new String(delivery.getBody()));
  37. }
  38. };
  39. channel.basicConsume(queueName, true, callback, i->{});
  40. } catch (Exception e) {
  41. e.printStackTrace();
  42. }
  43. }
  44. }

运行消费者、生产者,查看控制台,消费者日志输出内容,可以发现两个绑定不同Queue队列的消费者,接收到了相同的消息,这就是发布订阅模式。

到此,RabbitMQ消息队列中的发布订阅模式就介绍完啦。

综上,这篇文章结束了,主要介绍消息队列RabbitMQ七种模式之发布订阅模式(Publish/Subscribe)。

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/小惠珠哦/article/detail/843978
推荐阅读
相关标签
  

闽ICP备14008679号