Spring Boot 中 RabbitMQ 的使用

八卦阵  金牌会员 | 17 小时前 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 854|帖子 854|积分 2562

目录
引入依靠
添加配置
Simple（简朴模式）
生产者代码
消费者代码
?编辑
Work Queue（工作队列）
生产者代码
消费者代码
Publish/Subscribe（发布/订阅）
生产者代码
消费者代码
Routing（路由模式）
生产者代码
消费者代码
Topics（通配符模式）
生产者代码
消费者代码
常见问题
交换机与队列的绑定
交换机类型错误
队列属性错误
在 RabbitMQ 的工作模式_发布订阅和工作模式的区别-CSDN博客中，我们学习了 RabbitMQ 的 7 种工作模式，接下来，在本篇文章中，我们就来在Spring Boot 中实现常见的工作模式（由于 RPC 模式 和 发布确认模式 使用较少，因此在这里就不举行先容了），进而学习在 Spring Boot 中如何使用 RabbitMQ
在编写代码之前，我们需要先创建项目，引入依靠，并配置基本信息
引入依靠

  1.         <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-amqp</artifactId>
  4.         </dependency>
  5.         <dependency>
  6.             <groupId>org.springframework.boot</groupId>
  7.             <artifactId>spring-boot-starter-web</artifactId>
  8.         </dependency>
  10.         <dependency>
  11.             <groupId>org.springframework.boot</groupId>
  12.             <artifactId>spring-boot-starter-test</artifactId>
  13.             <scope>test</scope>
  14.         </dependency>
复制代码
或是在创建项目时，添加依靠：

接着，需要添加配置
添加配置

  1. spring:
  2.   rabbitmq:
  3.     host: 49.232.238.62
  4.     port: 5672 #默认为5672
  5.     username: admin
  6.     password: 123456
  7.     virtual-host: test02
复制代码
或是：
  1. spring:
  2.     rabbitmq:
  3.       addresses: amqp://admin:123456@49.232.238.62:5672/test02
复制代码
在这里，使用的虚拟机为 test02，因此，需要创建虚拟机 test02：

创建乐成：

接下来，就可以开始编写代码了
Simple（简朴模式）

在简朴模式下，只有一个生产者和一个消费者，生产者生产的消息存储到队列后，由这个消费者消费

我们先来实现生产者代码
生产者代码

为了方便举行测试，我们通过接口来发送消息
创建 Constants 类，定义 简朴模式 下使用的队列名称：
  1. public class Constants {
  2.     // 简单模式
  3.     public static final String SIMPLE_QUEUE = "simple.queue";
  4. }
复制代码
接着，需要声明队列
创建 RabbitMQConfig 类，创建 简朴模式 下使用的队列：

注意不要导错包了，当前我们使用的队列位于org.springframework.amqp.core 包下
使用 QueueBuilder 声明队列：
  1. import com.example.springrabbitmq.constant.Constants;
  2. import org.springframework.amqp.core.*;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  6. @Configuration
  7. public class RabbitMQConfig {
  8.     // 简单模式
  9.     @Bean("simpleQueue")
  10.     public Queue simpleQueue() {
  11.         return QueueBuilder.durable(Constants.SIMPLE_QUEUE).build();
  12.     }
  13. }
复制代码
由于声明的队列来自于第三方的包，且后续 工作模式、发布定义模式等使用的队列也在 RabbitMQConfig 中定义，因此需要使用 @Bean 注解
此中durable 表示 持久化，调用 durable 方法，表示创建一个持久化队列
接着，调用 build 方法创建 Bean
不要忘记添加@Configuration 注解
在这里，我们直接使用内置的交换机来举行路由，因此，不需要声明交换机 以及 绑定交换机和队列
接着，我们创建 ProducerController 类，实现 简朴模式 下的生产者：
在 Spring 中，使用 RabbitTemplate 来操纵 RabbitMQ，因此，我们将它注入进来：
  1. @RestController
  2. @RequestMapping("/producer")
  3. public class ProducerController {
  4.     @Autowired
  5.     private RabbitTemplate rabbitTemplate;
  6. }
复制代码
接着，使用 convertAndSend 方法来发送消息：
   public void convertAndSend(String exchange, String routingKey, final Object object) throws AmqpException
  exchange：交换机名称，使用内置的交换机时，为 “”
routingKey：路由键，用内置交换机时，routingKey 为队列名称
object：要发送的消息
  1.     @RequestMapping("simple")
  2.     public String simple() {
  3.         for (int i = 0; i < 20; i++) {
  4.             String message = "simple... " + i;
  5.             rabbitTemplate.convertAndSend("", Constants.SIMPLE_QUEUE, message);
  6.         }
  7.         return "OK";
  8.     }
复制代码
完备代码：
  1. import com.example.springrabbitmq.constant.Constants;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("/producer")public class ProducerController {    @Autowired    private RabbitTemplate rabbitTemplate;    @RequestMapping("simple")
  2.     public String simple() {
  3.         for (int i = 0; i < 20; i++) {
  4.             String message = "simple... " + i;
  5.             rabbitTemplate.convertAndSend("", Constants.SIMPLE_QUEUE, message);
  6.         }
  7.         return "OK";
  8.     }
  9. }
复制代码
运行代码，观察 Queues and Streams：

此时并未创建队列
访问 127.0.0.1:8080/producer/simple，再次观察：

simple.queue 队列被创建，且队列中已有 20 条消息，也就是说，当我们运行步伐时，队列并不会立刻创建，而是当我们访问接口，要向队列中发送消息时，才会创建队列
我们检察此中一条消息：

消息精确发送
接下来，我们继续实现消费者代码
消费者代码

消费者需要监听队列，当队列中有消息时，从队列中获取消息并消费，因此，我们创建监听类SimpleListener
  1. public class SimpleListener {
  2.    
  3. }
复制代码
  接下来，如何监听队列中的消息呢？
  Spring 为我们提供了**@RabbitListener** 注解，用于监听 RabbitMQ 队列，通过这个注解，我们可以定义一个方法，以便从队列中吸收消息
@RabbitListener 支持多种参数类型，这些参数类型代表了从队列吸收到的消息和相关信息
常见参数类型：
   String：消息内容
  Message（org.springframework.amqp.core.Message）：Spring AMQP 的 Message 类，返回原始的消息体以及消息属性（如：消息ID、内容、队列信息等）
  Channel（com.rabbitmq.client.Channel）：RabbitMQ 的通道对象，可以用于举行更高级的操纵，如手动确认消息
  接下来，我们就来实现消费者监听队列：
使用**@RabbitListener** 时，需要指定监听的队列
  1. import com.example.springrabbitmq.constant.Constants;
  2. import org.springframework.amqp.core.Message;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class SimpleListener {
  8.     @RabbitListener(queues = Constants.SIMPLE_QUEUE)
  9.     public void simpleListener(Message message) {
  10.         System.out.println("listener 从队列 " + Constants.SIMPLE_QUEUE +" 中接收到消息：" + message);
  11.     }
  12. }
复制代码
不要忘记添加@Component，将其交给 Spring 管理
我们再次运行代码，观察打印的日志：


队列中的 20 条消息瞬间就被消费者消费掉了
可以看到，Message 中包罗了 交换机、RoutingKey、deliverTage、监听的队列等信息
且观察 Connection 和 Channel
此时有一个 Connection：

两个 Channel：

   哪一个是消费者的哪一个是生产者的呢？
  我们可以从 Message rates中确定，发布消息的是生产者，消费消息的是消费者

也可以根据consumer确定：

而当有多个消费者时，可以通过 ConsumerTag 来确定不同的消费者：

接下来，我们继续学习工作队列模式
Work Queue（工作队列）

工作队列模式下，有一个生产者和多个消费者，多个消费者共同消费消息

首先在 Constants 中声明工作队列模式下使用的队列：
  1.     // 工作模式
  2.     public static final String WORK_QUEUE = "work.queue";
复制代码
在 RabbitMQConfig 中声明队列：
  1.     // 工作队列模式
  2.     @Bean("workQueue")
  3.     public Queue workQueue() {
  4.         return QueueBuilder.durable(Constants.WORK_QUEUE).build();
  5.     }
复制代码
我们仍然使用内置的交换机举行路由，因此，也就不需要声明交换机和绑定交换机和队列
接下来， 我们就来实现生产者
生产者代码

工作队列模式与简朴模式的区别在于工作模式下有多个消费者，因此生产者的消费代码与简朴模式雷同
在 ProducerController 中发送消息：
  1.     @RequestMapping("/work")
  2.     public String work() {
  3.         for (int i = 0; i < 20; i++) {
  4.             String message = "work... " + i;
  5.             rabbitTemplate.convertAndSend("", Constants.WORK_QUEUE, message);
  6.         }
  7.         return "OK";
  8.     }
复制代码
运行，并访问 http://127.0.0.1:8080/producer/work，观察效果：

乐成创建队列，且 20 条消息乐成发送
接着，我们继续实现消费者代码
消费者代码

消费者的代码也与简朴模式下的代码雷同，只是在这里我们需要创建两个消费者：
我们可以定义WorkListener1 和WorkListener2，分别定义一个方法来监听：
  1. import com.example.springrabbitmq.constant.Constants;
  2. import org.springframework.amqp.core.Message;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class WorkListener1 {
  8.     @RabbitListener(queues = Constants.WORK_QUEUE)
  9.     public void queueListener1(Message message) {
  10.         System.out.println("listener1 从队列 " + Constants.SIMPLE_QUEUE +" 中接收到消息：" + message);
  11.     }
  12. }
  14. import com.example.springrabbitmq.constant.Constants;
  15. import org.springframework.amqp.core.Message;
  16. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  17. import org.springframework.stereotype.Component;
  19. @Component
  20. public class WorkListener2 {
  21.     @RabbitListener(queues = Constants.WORK_QUEUE)
  22.     public void queueListener2(Message message) {
  23.         System.out.println("listener2 从队列 " + Constants.SIMPLE_QUEUE +" 中接收到消息：" + message);
  24.     }
  25. }
复制代码
再次运行，观察效果：

可以看到两个消费者共同消费了这 20 条消息，且它们的 deliveryTag都是从 1 开始计数的
接下来，我们继续学习 发布定义模式
Publish/Subscribe（发布/订阅）

在发布/订阅模式中，每个消费者收到的消息都是雷同的，且多了Exchange脚色

在前面简朴模式和工作队列模式下，图中都没有出现交换机，但实际上，生产者生产的消息都是先发送到交换机，然后再路由到队列中的。在前两种模式下，直接使用 RabbitMQ 提供的内置交换机就可以实现，因此，并没有突出交换机的存在，但实际上生产者生产的消息不会直接投递到队列中
因此，在发布/订阅模式下，我们就需要声明交换机，并绑定队列和交换机
先在 Constants 类中声明 发布订阅模式 下使用的交换机和队列：
  1.     // 发布订阅模式
  2.     public static final String PUBLISH_CHANGE = "fanout";
  3.     public static final String PUBLISH_QUEUE_1 = "publish.queue.1";
  4.     public static final String PUBLISH_QUEUE_2 = "publish.queue.2";
复制代码
接着，在 RabbitMQConfig 中声明队列：
  1.     // 发布订阅模式
  2.     @Bean("fanoutQueue1")
  3.     public Queue fanoutQueue1() {
  4.         return QueueBuilder.durable(Constants.PUBLISH_QUEUE_1).build();
  5.     }
  7.     @Bean("fanoutQueue2")
  8.     public Queue fanoutQueue2() {
  9.         return QueueBuilder.durable(Constants.PUBLISH_QUEUE_2).build();
  10.     }
复制代码
还需要声明交换机
在 发布订阅模式 下，交换机的类型为 Fanout（广播）
在 Spring 中，使用FanoutExchange 表示广播类型的交换机
  1.     // 声明交换机
  2.     @Bean("fanoutExchange")
  3.     public FanoutExchange fanoutExchange() {
  4.         return ExchangeBuilder.fanoutExchange(Constants.PUBLISH_CHANGE).durable(true).build();
  5.     }
复制代码
使用ExchangeBuilder 的fanoutExchange方法创建广播类型的交换机
接着，需要将队列和交换机举行绑定
在 Spring 中，使用 Binding（org.springframework.amqp.core.Binding）表示交换机与队列的绑定关系
使用 BindingBuilder 举行绑定：

但此时队列不能主动注入，因为此时有多个 Queue 类型的队列
我们可以使用 @Qualifier 注解指定我们要绑定的队列（fanoutExchange 由于只有一个，可以不指定）：
  1.     // 绑定交换机和队列
  2.     @Bean("fanoutQueueBinding1")
  3.     public Binding fanoutQueueBinding1(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,@Qualifier("fanoutQueue1") Queue queue) {
  4.         return BindingBuilder.bind(queue).to(fanoutExchange);
  5.     }
  7.     @Bean("fanoutQueueBinding2")
  8.     public Binding fanoutQueueBinding2(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange, @Qualifier("fanoutQueue2") Queue queue) {
  9.         return BindingBuilder.bind(queue).to(fanoutExchange);
  10.     }
复制代码
接着，我们就可以使用接口发送消息了
生产者代码

在 ProducerController 中使用rabbitTemplate 发送消息
  1.     @RequestMapping("/fanout")
  2.     public String fanout() {
  3.         for (int i = 0; i < 20; i++) {
  4.             String message = "fanout ... " + i;
  5.             rabbitTemplate.convertAndSend(Constants.PUBLISH_CHANGE, "", message);
  6.         }
  7.         return "OK";
  8.     }
复制代码
注意，发布订阅模式下，routingKey 为空，表示所有队列都可以收到消息
运行步伐，访问 http://127.0.0.1:8080/producer/fanout，并观察效果

两个队列中都有 20 条消息，消费发送乐成
接着，我们继续编写消费者代码
消费者代码

发布订阅模式下，消费者的代码与工作模式下的代码雷同
为了方便观察，在这里就直接将两个消费者写在一个类中：
  1. @Component
  2. public class FanoutListener {
  3.     @RabbitListener(queues = Constants.PUBLISH_QUEUE_1)
  4.     public void queueListener1(String message) {
  5.         System.out.println("listener1 从队列 " + Constants.PUBLISH_QUEUE_1 +" 中接收到消息：" + message);
  6.     }
  8.     @RabbitListener(queues = Constants.PUBLISH_QUEUE_2)
  9.     public void queueListener2(String message) {
  10.         System.out.println("listener2 从队列 " + Constants.PUBLISH_QUEUE_2 +" 中接收到消息：" + message);
  11.     }
  12. }
复制代码
由于我们只关心消息的内容，因此，可以只使用 String 类型来吸收消息
再次运行，观察打印的日志：

两个消费者都消费了 20 条雷同的消息
我们继续看路由模式
Routing（路由模式）

Routing 模式下，队列与交换机之间的绑定，不再是任意的绑定了，而是需要指定一个 BindingKey
生产者在向 交换机 发送消息时，也需要指定消息的 RoutingKey
交换机不会将消息发送给每一个绑定的 key，而是会根据消息的 RoutingKey 举行判断，只有队列绑定时的 BindingKey 和发送消息的 RoutingKey 完全一致时，才会吸收消息

先在 Constants 中声明路由模式下使用的队列和交换机：
  1.     // 路由模式
  2.     public static final String ROUTING_CHANGE = "routing";
  3.     public static final String ROUTINT_QUEUE_1 = "routing.queue.1";
  4.     public static final String ROUTINT_QUEUE_2 = "routing.queue.2";
复制代码
接着，在RabbitMQConfig 中声明队列：
  1.     // 路由模式
  2.     @Bean("routingQueue1")
  3.     public Queue routingQueue1() {
  4.         return QueueBuilder.durable(Constants.ROUTINT_QUEUE_1).build();
  5.     }
  7.     @Bean("routingQueue2")
  8.     public Queue routingQueue2() {
  9.         return QueueBuilder.durable(Constants.ROUTINT_QUEUE_2).build();
  10.     }
复制代码
声明交换机：
注意，路由模式下，交换机类型为Direct（定向）
在 Spring 中，使用DirectExchange表示广播类型的交换机
  1.     // 声明交换机
  2.     @Bean("routingExchange")
  3.     public DirectExchange routingExchange() {
  4.         return ExchangeBuilder.directExchange(Constants.ROUTING_CHANGE).durable(true).build();
  5.     }
复制代码
使用ExchangeBuilder 的directExchange方法创建通配符类型的交换机
绑定队列和交换机：
  1.     // 绑定交换机和队列
  2.     @Bean("routingQueueBinding1")
  3.     public Binding routingQueueBinding1(@Qualifier("routingExchange") DirectExchange directExchange, @Qualifier("routingQueue1") Queue queue) {
  4.         return BindingBuilder.bind(queue).to(directExchange).with("a");
  5.     }
  7.     @Bean("routingQueueBinding2")
  8.     public Binding routingQueueBinding2(@Qualifier("routingExchange") DirectExchange directExchange, @Qualifier("routingQueue2") Queue queue) {
  9.         return BindingBuilder.bind(queue).to(directExchange).with("a");
  10.     }
  12.     @Bean("routingQueueBinding3")
  13.     public Binding routingQueueBinding3(@Qualifier("routingExchange") DirectExchange directExchange, @Qualifier("routingQueue2") Queue queue) {
  14.         return BindingBuilder.bind(queue).to(directExchange).with("b");
  15.     }
  17.     @Bean("routingQueueBinding4")
  18.     public Binding routingQueueBinding4(@Qualifier("routingExchange") DirectExchange directExchange, @Qualifier("routingQueue2") Queue queue) {
  19.         return BindingBuilder.bind(queue).to(directExchange).with("c");
  20.     }
复制代码
在绑定交换机时，需要调用 with 方法指定 BindingKey
之后，就可以发送消息了
生产者代码

在 ProducerController 中使用rabbitTemplate 发送消息
在 路由模式 下，需要指定 RoutingKey，为了测试不同 RoutingKey 发送消息的情况，我们在路径中指定RoutingKey
  1.     @RequestMapping("/routing/{routingKey}")
  2.     public String direct(@PathVariable("routingKey") String routingKey) {
  3.         rabbitTemplate.convertAndSend(Constants.ROUTING_CHANGE, routingKey, "routing test... " +routingKey);
  4.         return "OK";
  5.     }
复制代码
运行步伐，访问http://127.0.0.1:8080/producer/routing/a、http://127.0.0.1:8080/producer/routing/b和http://127.0.0.1:8080/producer/routing/c
观察效果：

routing.queue.1 中只有 1 条消息，routing.queue.2 中有 3 条消息
只有当 routingKey 为 a 时，才能与routing.queue.1 的 BindingKey 匹配，而 routingKey 为 a、b、c 时，都能与routing.queue.2 的 BindingKey 相匹配，因此routing.queue.1 中只有 1 条消息，routing.queue.2 中有 3 条消息
我们检察routing.queue.2 中的消息：

接着，我们继续实现消费者代码
消费者代码

消费者的代码与 发布订阅模式 的雷同，因此我们只需要修改消费的队列即可：
  1. @Component
  2. public class RoutingListener {
  3.     @RabbitListener(queues = Constants.ROUTINT_QUEUE_1)
  4.     public void queueListener1(String message) {
  5.         System.out.println("listener1 从队列 " + Constants.ROUTINT_QUEUE_1 +" 中接收到消息：" + message);
  6.     }
  8.     @RabbitListener(queues = Constants.ROUTINT_QUEUE_2)
  9.     public void queueListener2(String message) {
  10.         System.out.println("listener2 从队列 " + Constants.ROUTINT_QUEUE_2 +" 中接收到消息：" + message);
  11.     }
  12. }
复制代码
再次运行，观察日志：

我们继续看 通配符模式
Topics（通配符模式）


相比于 routing 模式，topics 类型的交换机在匹配规则上举行了扩展，BindingKey 支持通配符匹配
RoutingKey 是一系列由 . 分割的单词，如 user.name、work.abc等
BindingKey 也和 RoutingKey 一样，由 . 分割的字符串
在 BindingKey 中可以存在两种特殊的字符串，用于模糊匹配：
   * ：表示可以或许匹配任意一个单词
  #：表示可以或许匹配任意多个单词（可以为 0 个）
  先在 Constants 中声明通配符模式下使用的队列和交换机：
  1.     // 通配符模式
  2.     public static final String TOPICS_CHANGE = "topics";
  3.     public static final String TOPICS_QUEUE_1 = "topics.queue.1";
  4.     public static final String TOPICS_QUEUE_2 = "topics.queue.2";
复制代码
接着，在RabbitMQConfig 中声明队列：
  1.     // 通配符模式
  2.     @Bean("topicsQueue1")
  3.     public Queue topicsQueue1() {
  4.         return QueueBuilder.durable(Constants.TOPICS_QUEUE_1).build();
  5.     }
  6.     @Bean("topicsQueue2")
  7.     public Queue topicsQueue2() {
  8.         return QueueBuilder.durable(Constants.TOPICS_QUEUE_2).build();
  9.     }
复制代码
声明交换机：
注意，通配符模式下，交换机类型为Topics（通配符）
在 Spring 中，使用TopicExchange 表示通配符类型的交换机
  1.     @Bean("topicsExchange")
  2.     public TopicExchange topicExchange() {
  3.         return ExchangeBuilder.topicExchange(Constants.TOPICS_CHANGE).durable(true).build();
  4.     }
复制代码
使用ExchangeBuilder 的topicExchange 方法创建通配符类型的交换机
绑定队列和交换机：
  1.     // 绑定交换机和队列
  2.     @Bean("topicsQueueBinding1")
  3.     public Binding topicsQueueBinding1(@Qualifier("topicsExchange") TopicExchange topicExchange, @Qualifier("topicsQueue1") Queue queue) {
  4.         return BindingBuilder.bind(queue).to(topicExchange).with("*.*.a");
  5.     }
  7.     @Bean("topicsQueueBinding2")
  8.     public Binding topicsQueueBinding2(@Qualifier("topicsExchange") TopicExchange topicExchange, @Qualifier("topicsQueue2") Queue queue) {
  9.         return BindingBuilder.bind(queue).to(topicExchange).with("*.b.*");
  10.     }
  11.     @Bean("topicsQueueBinding3")
  12.     public Binding topicsQueueBinding3(@Qualifier("topicsExchange") TopicExchange topicExchange, @Qualifier("topicsQueue2") Queue queue) {
  13.         return BindingBuilder.bind(queue).to(topicExchange).with("c.#");
  14.     }
复制代码
调用 with 方法指定 BindingKey
接下来，就可以发送消息了
生产者代码

在 ProducerController 中使用rabbitTemplate 发送消息
同样的，在 通配符模式 下，需要指定 RoutingKey，为了测试不同 RoutingKey 发送消息的情况，我们在路径中指定RoutingKey
  1.     @RequestMapping("/topics/{topicsKey}")
  2.     public String topics(@PathVariable("topicsKey") String topicsKey) {
  3.         rabbitTemplate.convertAndSend(Constants.TOPICS_CHANGE, topicsKey, "topics test... " +topicsKey);
  4.         return "OK";
  5.     }
复制代码
运行步伐，并访问http://127.0.0.1:8080/producer/topics/a.b.a、http://127.0.0.1:8080/producer/topics/c.work和http://127.0.0.1:8080/producer/topics/a.a.a
观察效果：

topics.queue.1 和topics.queue.2 中都有两条消息，我们先来看topics.queue.1：

topics.queue.2 中的消息：

当 topicsKey为 a.b.a 时，能与topics.queue.1的 BindingKey（*.*.a） 匹配，也能与 topics.queue.2的 BindingKey（*.b.*） 匹配，因此，消息会被路由到两个队列中
topicsKey 为 c.work时，只能与topics.queue.2 的 BindingKey（c.#）相匹配
topicsKey 为 a.a.a时，只能与topics.queue.1的 BindingKey（*.*.a） 相匹配
我们继续实现消费者代码
消费者代码

消费者的代码与 路由模式 的雷同，因此我们只需要修改消费的队列即可：
  1. @Component
  2. public class TopicsListener {
  3.     @RabbitListener(queues = Constants.TOPICS_QUEUE_1)
  4.     public void queueListener1(String message) {
  5.         System.out.println("listener1 [" + Constants.TOPICS_QUEUE_1 +"] 接收到消息: " + message);
  6.     }
  8.     @RabbitListener(queues = Constants.TOPICS_QUEUE_2)
  9.     public void queueListener2(String message) {
  10.         System.out.println("listener2 [" + Constants.TOPICS_QUEUE_2 +"] 接收到消息: " + message);
  11.     }
  12. }
复制代码
再次运行，观察日志：

常见问题

在我们编写代码的过程中，可能会出现一些问题，接下来，我们就一起来看看常见的错误类型
交换机与队列的绑定

比方，在 通配符模式 下，交换机类型绑定为 Direct（定向）模式：

此时，若使用了@Qualifier 注解，则会直接报错，表现类型不匹配
但是，若未使用@Qualifier 注解，则不会报错，步伐也能正常运行

我们观察交换机 routing：

可以看到 topics.queue.1 与 routing 交换机举行了绑定，我们再访问， 消息乐成发送
但 topics.queue.1 队列中始终没有消息，这是因为topics.queue.1 此时并未与 topics 交换机举行绑定，topics 交换机在吸收到消息后，发现没有匹配的 BindingKey，就直接将消息丢弃了
当消息发送乐成，但队列中却没有消息时，就需要检查队列和交换机的绑定关系了
交换机类型错误

在我们定义交换机时，可能会一不小心将交换机的名称写错：

我们运行步伐，观察日志：

我们来看 ERROR 信息表现在 虚拟机 test02 上的交换机 fanout 吸收到的类型为 topic，但当前 fanout 是 fanout 类型的

fanout交换机 是 fanout 类型的，但是此时将其当做 topic 类型来使用，也就报错了
若一个交换机已经存在，则不能再修改其类型了，若我们需要举行修改，则需要将之前已经存在的交换机删除
队列属性错误

比方，topics.queue.1 是一个持久化的队列，但此时我们将其声明为 非持久化 的：

再次运行：

也表现了 ERROR 信息，提示 在虚拟机 test02 上的队列 topics.queue.1 是持久化的，吸收到的参数是 false，但当前为 true
也就是说，我们尝试修改队列的持久化参数，此时是不被允许的
topics.queue.1 是持久化的：

D 表示 durable 为 true
若我们需要将其修改为非持久化的，需要先将已经存在的 topics.queue.1 队列删除：

此时再次运行步伐，访问http://127.0.0.1:8080/producer/topics/a.a.a
观察效果：

此时，topics.queue.1 就黑白持久化的了
当交换机、队列创建完成时，其属性是不能发生变革的，若需要修改，则需要将当前交换机或队列删除，然后重新声明

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

八卦阵

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表