Springboot整合RabbitMQ基本使用

1、依赖

1.         <dependency>
2.                         <groupId>org.springframework.boot</groupId>
3.                         <artifactId>spring-boot-starter-amqp</artifactId>
4.                 </dependency>

复制代码

2、rabbitmq链接配置

1. spring:
2.   rabbitmq:
3.     host: 127.0.0.1
4.     port: 5672
5.     username: wq
6.     password: qifeng
7.     virtual-host: /
8.     #开启ack
9.     listener:
10.       direct:
11.         acknowledge-mode: manual
12.         prefetch: 1 # 限制一次拉取消息的数量
13.       simple:
14.         acknowledge-mode: manual #采取手动应答
15.         #concurrency: 1 # 指定最小的消费者数量
16.         #max-concurrency: 1 #指定最大的消费者数量
17.         retry:
18.           enabled: true # 是否支持重试

复制代码

3、生产者

• 声明交换机、队列、绑定关于
  

1. package com.wanqi.mq;
3. import org.springframework.amqp.core.*;
4. import org.springframework.beans.factory.annotation.Qualifier;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
8. /**
9. * @Description TODO
10. * @Version 1.0.0
11. * @Date 2022/11/19
12. * @Author wandaren
13. */
14. @Configuration
15. public class RabbitMQConfig {
16.     public static final String EXCHANGE_NAME = "boot_topic_exchange";
17.     public static final String QUEUE_NAME = "boot_queue";
18.     //交换机
19.     @Bean("bootExchange")
20.     public Exchange bootExchange(){
21.         return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true)
22.                 .build();
23.     }
24.     //
25.     @Bean("bootQueue")
26.     public Queue bootQueue(){
27.         return QueueBuilder.durable(QUEUE_NAME).build();
28.     }
29.     //队列和交换机绑定关系
30.     /*
31.         知道哪个队列
32.         知道哪个交换机
33.         知道routing key
34.      */
35.     @Bean
36.     public Binding bindingQueueExchange(@Qualifier("bootQueue") Queue queue,
37.                                         @Qualifier("bootExchange") Exchange exchange){
38.         return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
39.     }
40. }

复制代码

• 发送消息
  

1. package com.wanqi;
3. import com.wanqi.mq.RabbitMQConfig;
4. import org.junit.jupiter.api.Test;
5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
6. import org.springframework.beans.factory.annotation.Autowired;
7. import org.springframework.boot.test.context.SpringBootTest;
9. @SpringBootTest
10. class SpringbootMqProducersApplicationTests {
11.         @Autowired
12.         private RabbitTemplate rabbitTemplate;
14.         @Test
15.         void contextLoads() {
16.                 rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, "boot.test", "boot mq hello~~~");
17.         }
19. }

复制代码

4、消费者

1. package com.wanqi.listener;
3. import com.rabbitmq.client.Channel;
4. import org.springframework.amqp.core.Message;
5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
6. import org.springframework.stereotype.Component;
8. import java.io.IOException;
10. /**
11. * @Description TODO
12. * @Version 1.0.0
13. * @Date 2022/11/19
14. * @Author wandaren
15. */
16. @Component
17. public class RabbitMQListener {
19.     @RabbitListener(queues = {"boot_queue"})
20.     public void listenerQueue(Object msg, Message message, Channel channel) {
21.         final long deliveryTag = message.getMessageProperties().getDeliveryTag();
23.         try {
24.             System.out.println(msg.toString());
25.             System.out.println(new String(message.getBody()));
26. //            int x = 3/0;
27.             channel.basicAck(deliveryTag, true);
28.         } catch (Exception e) {
29.             try {
30.                 channel.basicNack(deliveryTag, true, true);
31.             } catch (IOException ex) {
32.                 throw new RuntimeException(ex);
33.             }
34.         }
35.     }
36. }

复制代码

5、死信队列

消息成为死信的三种情况
1、队列消息长度到达限制；
2、消费者拒接消费消息，并且不重回队列；
3、原队列存在消息过期设置，消息到达超时时间未被消费；
5.1、声明交换机、队列、绑定队列

• topic模式
  

1. package com.wanqi.mq;
3. import org.springframework.amqp.core.*;
4. import org.springframework.beans.factory.annotation.Qualifier;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
8. /**
9. * @Description TODO
10. * @Version 1.0.0
11. * @Date 2022/11/19
12. * @Author wandaren
13. */
14. @Configuration
15. public class QDRabbitMQConfig {
16.     //交换机名称
17.     public static final String ITEM_EXCHANGE = "item_exchange";
18.     public static final String DEAD_EXCHANGE = "dead_exchange";
19.     //队列名称
20.     public static final String ITEM_QUEUE = "item_queue";
21.     public static final String DEAD_QUEUE = "dead_queue";
23.     //声明业务交换机
24.     @Bean("itemExchange")
25.     public Exchange itemExchange(){
26.         return ExchangeBuilder.topicExchange(ITEM_EXCHANGE).durable(true).build();
27.     }
28.     //声明死信交换机
29.     @Bean("deadExchange")
30.     public Exchange deadExchange(){
31.         return ExchangeBuilder.topicExchange(DEAD_EXCHANGE).durable(true).build();
32.     }
34.     /**
35.      * 声明普通队列，设置队列消息过期时间，队列长度，绑定的死信队列
36.      */
37.     @Bean("itemQueue")
38.     public Queue itemQueue(){
39.         return QueueBuilder
40.                 .durable(ITEM_QUEUE)
41.                 // 队列消息过期时间
42.                 .ttl(10000)
43.                 // 队列长度
44.                 .maxLength(15)
45.                 // 声明当前队列绑定的死信交换机
46.                 .deadLetterExchange(DEAD_EXCHANGE)
47.                 // 声明当前队列死信转发的路由key
48.                 .deadLetterRoutingKey("infoDead.haha")
49.                 .build();
50.     }
51.     /**
52.      * 死信队列，消费者需要监听的队列
53.      */
54.     @Bean("deadQueue")
55.     public Queue deadQueue(){
56.         return QueueBuilder
57.                 .durable(DEAD_QUEUE)
58.                 .build();
59.     }
61.     //绑定队列和交换机（业务）
62.     @Bean
63.     public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue,
64.                                      @Qualifier("itemExchange") Exchange exchange){
65.         return BindingBuilder.bind(queue).to(exchange).with("infoRouting.#").noargs();
66.     }
67.     //绑定队列和交换机（死信）
68.     @Bean
69.     public Binding deadQueueExchange(@Qualifier("deadQueue") Queue queue,
70.                                      @Qualifier("deadExchange") Exchange exchange){
71.         return BindingBuilder.bind(queue).to(exchange).with("infoDead.#").noargs();
72.     }
74. }

复制代码

5.2、发送测试消息

1. package com.wanqi;
3. import com.wanqi.mq.RabbitMQConfig;
4. import com.wanqi.mq.RabbitMQConfig2;
5. import org.junit.jupiter.api.Test;
6. import org.springframework.amqp.core.Message;
7. import org.springframework.amqp.core.MessageProperties;
8. import org.springframework.amqp.rabbit.core.RabbitTemplate;
9. import org.springframework.beans.factory.annotation.Autowired;
10. import org.springframework.boot.test.context.SpringBootTest;
12. @SpringBootTest
13. class SpringbootMqProducersApplicationTests {
14.         @Autowired
15.         private RabbitTemplate rabbitTemplate;
16.    
17.         // 原队列存在消息过期设置，消息到达超时时间未被消费；
18.         @Test
19.         void contextLoads3() {
20.                 MessageProperties messageProperties = new MessageProperties();
21.                 // 设置过期时间，单位：毫秒
22.                 messageProperties.setExpiration("5000");
23.                 byte[] msgBytes = "rabbitmq ttl message ...".getBytes();
24.                 Message message = new Message(msgBytes, messageProperties);
26.                 //发送消息
27.                 rabbitTemplate.convertAndSend(QDRabbitMQConfig.ITEM_EXCHANGE,"infoRouting.hehe",message);
28.                 System.out.println("发送消息成功");
29.         }
30.     // 模拟队列消息长度到达限制
31.         @Test
32.         void contextLoads4() {
33.                 for (int i = 0; i < 20; i++) {
34.                         rabbitTemplate.convertAndSend(QDRabbitMQConfig.ITEM_EXCHANGE,"infoRouting.hehe",i + "---message");
36.                 }
37.         }
38. }

复制代码

5.3、消费者

1. package com.wanqi.listener;
3. import com.rabbitmq.client.Channel;
4. import org.springframework.amqp.core.Message;
5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
6. import org.springframework.stereotype.Component;
8. import java.io.IOException;
10. /**
11. * @Description TODO
12. * @Version 1.0.0
13. * @Date 2022/11/19
14. * @Author wandaren
15. */
16. @Component
17. public class RabbitMQListener {
18.     @RabbitListener(queues = {"dead_queue"})
19.     public void listenerQueue2(Object msg, Message message, Channel channel) {
20.         final long deliveryTag = message.getMessageProperties().getDeliveryTag();
21.         try {
22.             System.out.println(msg.toString());
23.             channel.basicAck(deliveryTag, true);
24.         } catch (Exception e) {
25.             try {
26.                 channel.basicNack(deliveryTag, true, true);
27.             } catch (IOException ex) {
28.                 throw new RuntimeException(ex);
29.             }
30.         }
31.     }
32. }

复制代码

6、延迟队列（TTL + 死信队列）

6.1、声明交换机、队列、绑定队列

• direct模式
  

1. package com.wanqi.mq;
3. import org.springframework.amqp.core.*;
4. import org.springframework.beans.factory.annotation.Qualifier;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
8. /**
9. * @Description TODO
10. * @Version 1.0.0
11. * @Date 2022/11/19
12. * @Author wandaren
13. */
14. @Configuration
15. public class TtlQueueConfig {
16.     //普通交换机名称
17.     public static final String X_CHANGE = "X_Exchange";
18.     //死信交换机名称
19.     public static final String Y_DEAD_CHANGE = "Y_Exchange";
20.     //普通队列
21.     public static final String QUEUE_A = "QA_QUEUE";
22.     public static final String QUEUE_B = "QB_QUEUE";
23.     //死信队列
24.     public static final String DEAD_QUEUE_D = "QD_QUEUE";
26.     //声明普通交换机
27.     @Bean("xExchange")
28.     public DirectExchange xExchange() {
29.         return ExchangeBuilder.directExchange(X_CHANGE).durable(true)
30.                 .build();
31.     }
33.     //声明死信交换机
34.     @Bean("yExchange")
35.     public DirectExchange yExchange() {
36.         return ExchangeBuilder.directExchange(Y_DEAD_CHANGE).durable(true)
37.                 .build();
38.     }
40.     /**
41.      * 声明队列,延迟10秒
42.      */
44.     @Bean("queueA")
45.     public Queue queueA() {
47.         return QueueBuilder.durable(QUEUE_A)
48.                 .deadLetterExchange(Y_DEAD_CHANGE) //死信交换机
49.                 .deadLetterRoutingKey("YD")  //死信RoutingKey
50.                 .ttl(10000)  //消息过期时间
51.                 .build();
52.     }
54.     /**
55.      * 声明队列，延迟40秒
56.      *
57.      * @return
58.      */
59.     @Bean("queueB")
60.     public Queue queueB() {
61.         return QueueBuilder.durable(QUEUE_B)
62.                 .deadLetterExchange(Y_DEAD_CHANGE) //死信交换机
63.                 .deadLetterRoutingKey("YD")  //死信RoutingKey
64.                 .ttl(40000)  //消息过期时间
65.                 .build();
66.     }
68.     /**
69.      * 死信队列，消费者需要监听的队列
70.      */
71.     @Bean("queueD")
72.     public Queue queueD() {
73.         return QueueBuilder.durable(DEAD_QUEUE_D).build();
74.     }
76.     //绑定  X_CHANGE绑定queueA
77.     @Bean
78.     public Binding queueABindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange") DirectExchange xExchange) {
79.         return BindingBuilder.bind(queueA).to(xExchange).with("XA");
80.     }
82.     //绑定  X_CHANGE绑定queueB
83.     @Bean
84.     public Binding queueBBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") DirectExchange xExchange) {
85.         return BindingBuilder.bind(queueB).to(xExchange).with("XB");
86.     }
88.     //绑定  Y_CHANGE绑定queueD
89.     @Bean
90.     public Binding queueDBindingY(@Qualifier("queueD") Queue queueD, @Qualifier("yExchange") DirectExchange yExchange) {
91.         return BindingBuilder.bind(queueD).to(yExchange).with("YD");
92.     }
93. }

复制代码

6.2、发送消息

1.     @Test
2.     void contextLoads5() {
3.         String message = "延迟队列消息";
4.         rabbitTemplate.convertAndSend(TtlQueueConfig.X_CHANGE, "XA", "TTL=10s的队列：" + message);
5.         rabbitTemplate.convertAndSend(TtlQueueConfig.X_CHANGE, "XB", "TTL=40s的队列：" + message);
6.     }

复制代码

6.3、消费

1. package com.wanqi.listener;
3. import com.rabbitmq.client.Channel;
4. import org.springframework.amqp.core.Message;
5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
6. import org.springframework.stereotype.Component;
8. import java.io.IOException;
9. import java.util.Date;
11. /**
12. * @Description TODO
13. * @Version 1.0.0
14. * @Date 2022/11/19
15. * @Author wandaren
16. */
17. @Component
18. public class RabbitMQListener {
19.     @RabbitListener(queues = "QD_QUEUE")
20.     public void receiveMessage(Object msg,Message message, Channel channel){
21.         final long deliveryTag = message.getMessageProperties().getDeliveryTag();
22.         try {
23.             System.out.println(msg.toString());
24.             System.out.print("当前时间: " + new Date());
25.             System.out.println("  收到死信队列的消息：" + msg);
26.             channel.basicAck(deliveryTag, true);
27.         } catch (Exception e) {
28.             try {
29.                 channel.basicNack(deliveryTag, true, true);
30.             } catch (IOException ex) {
31.                 throw new RuntimeException(ex);
32.             }
33.         }
34.     }
35. }

复制代码