Spring Boot 集成 RabbitMQ

依赖与设置

在 pom.xml 中引入 RabbitMQ 相干依赖

1. <!-- AMQP 依赖, RabbitMq -->
2. <dependency>
3.     <groupId>org.springframework.boot</groupId>
4.     <artifactId>spring-boot-starter-amqp</artifactId>
5.     <version>3.2.7</version>
6. </dependency>

复制代码

在 application.yml 中添加设置

1. spring:
2.   rabbitmq:
3.     host: 127.0.0.1
4.     port: 5672
5.     username: guest
6.     password: guest

复制代码

RabbitMQ使用

扇形互换机

扇形互换机，顾名思义，就是像一把扇子一样，一个互换器可以绑定多个队列，只要互换机吸取到消息就会发送给全部和它绑定的队列。
假设扇形互换机 fanoutExchange 绑定了队列 fanoutQueue1 和 fanoutQueue2 ，那么我们往 fanoutExchange 发送一条消息，fanoutQueue1 和 fanoutQueue2 都会收到一条雷同的消息，如果消息未被斲丧我们可以在 RabbitMQ 管理端看到这两个队列和队列内积蓄的一条雷同的消息。

1. @Configuration
2. public class FanoutExchangeConfig {
4.     public static final String FANOUT_QUEUE1 = "fanout.queue1";
5.     public static final String FANOUT_QUEUE2 = "fanout.queue2";
6.     public static final String FANOUT_EXCHANGE = "fanout.exchange";
8.     //声明队列Q1
9.     @Bean("fanoutQ1")
10.     public Queue fanoutQ1() {
11.         return new Queue(FANOUT_QUEUE1);
12.     }
14.     //声明队列Q1
15.     @Bean("fanoutQ2")
16.     public Queue fanoutQ2() {
17.         return new Queue(FANOUT_QUEUE2);
18.     }
20.     //声明扇形交换机
21.     @Bean
22.     public FanoutExchange fanoutExchange() {
23.         return new FanoutExchange(FANOUT_EXCHANGE);
24.     }
26.     //队列Q1绑定扇形交换机
27.     @Bean("bindFanoutQ1")
28.     public Binding bindFanoutQ1() {
29.         return BindingBuilder.bind(fanoutQ1()).to(fanoutExchange());
30.     }
32.     //队列Q2绑定扇形交换机
33.     @Bean("bindFanoutQ2")
34.     public Binding bindFanoutQ2() {
35.         return BindingBuilder.bind(fanoutQ2()).to(fanoutExchange());
36.     }
38. }

复制代码

编写 junit 测试， 发送消息

1. @Test
2. public void fanoutTest() {
3.     rabbitTemplate.convertAndSend(FanoutExchangeConfig.FANOUT_EXCHANGE, "", "fantoutTest");
4. }

复制代码

发送消息后由于没有斲丧者，可以在管理端看到积蓄在队列中的消息。

直连互换机

一个直连互换机可以有多个队列，但每个队列都有一个路由逐一匹配，互换机根据路由将消息投递到对应队列中。当同一个队列有多个斲丧者时，消息不会被重复斲丧，直连互换机可以大概轮询公平的将消息分发给每个斲丧者。
假设直连互换机 directExchange 与队列 directQueue1 通过路由 directRoute1 绑定， 与directQueue2 通过路由 directRoute2 绑定。当生产者发送路由为 directRoute1  的消息给 directExchange 时，消息会被投递到 directQueue1 ，directQueue2 则吸取不到消息。

1. @Configuration
2. public class DirectExchangeConfig {
4.     public static final String DIRECT_QUEUE1 = "direct.queue1";
5.     public static final String DIRECT_QUEUE2 = "direct.queue2";
6.     public static final String DIRECT_EXCHANGE = "direct.exchange";
8.     public static final String DIRECT_ROUTE_KEY1 = "direct.route.key1";
10.     //声明队列Q1
11.     @Bean("directQ1")
12.     public Queue directQ1() {
13.         return new Queue(DIRECT_QUEUE1);
14.     }
16.     //声明队列Q1
17.     @Bean("directQ2")
18.     public Queue directQ2() {
19.         return new Queue(DIRECT_QUEUE2);
20.     }
22.     //声明直连交换机
23.     @Bean
24.     public DirectExchange directExchange() {
25.         return new DirectExchange(DIRECT_EXCHANGE);
26.     }
28.     //队列Q1绑定直连交换机
29.     @Bean("bindDirectQ1")
30.     public Binding bindDirectQ1() {
31.         return BindingBuilder.bind(directQ1()).to(directExchange()).with(DIRECT_ROUTE_KEY1);
32.     }
34.     //队列Q2绑定直连交换机
35.     @Bean("bindDirectQ2")
36.     public Binding bindDirectQ2() {
37.         return BindingBuilder.bind(directQ2()).to(directExchange()).with("");
38.     }
40. }

复制代码

编写 junit 测试，投递消息给互换机。

1. @Test
2. public void directTest() {
3.     //消息被投递给 bindDirectQ2
4.     rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE, "", "directTest");
5.     //消息被投递给 bindDirectQ1
6.     rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE, DIRECT_ROUTE_KEY1, "directTest-key1");
7.     //没有匹配的路由，bindDirectQ1 和 bindDirectQ2 都无法接收
8.     rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE, "test", "directTest-test");
9. }

复制代码

主题互换机

一个主题互换机可以有多个绑定队列，支持路由暗昧匹配，可以使用星号（*）和井号（#）作为通配符举行匹配。此中，* 可以取代一个单词，# 可以取代恣意个单词。
假设主题互换机 topicExchange 通过路由 topic.route.* 绑定队列 topicQueue1 ， 通过路由 topic.route.# 绑定队列 topicQueue2。当生产者通过路由 topic.route.1 和 topic.route.1.1 投递消息给 topicExchange  时， topicQueue2 会吸取到两条差异路由的消息， 而 topicQueue1 仅能吸取到路由为 topic.route.1 的消息。

1. @Configuration
2. public class TopicExchangeConfig {
4.     public static final String TOPIC_QUEUE1 = "topic.queue1";
5.     public static final String TOPIC_QUEUE2 = "topic.queue2";
6.     public static final String TOPIC_EXCHANGE = "topic.exchange";
8.     public static final String TOPIC_ROUTE_KEY1 = "topic.route.*";
9.     public static final String TOPIC_ROUTE_KEY2 = "topic.route.#";
11.     //声明队列Q1
12.     @Bean("topicQ1")
13.     public Queue topicQ1() {
14.         return new Queue(TOPIC_QUEUE1);
15.     }
17.     //声明队列Q1
18.     @Bean("topicQ2")
19.     public Queue topicQ2() {
20.         return new Queue(TOPIC_QUEUE2);
21.     }
23.     //声明主题交换机
24.     @Bean
25.     public TopicExchange topicExchange() {
26.         return new TopicExchange(TOPIC_EXCHANGE);
27.     }
29.     //队列Q1绑定主题交换机
30.     @Bean("bindTopicQ1")
31.     public Binding bindTopicQ1() {
32.         return BindingBuilder.bind(topicQ1()).to(topicExchange()).with(TOPIC_ROUTE_KEY1);
33.     }
35.     //队列Q2绑定主题交换机
36.     @Bean("bindTopicQ2")
37.     public Binding bindTopicQ2() {
38.         return BindingBuilder.bind(topicQ2()).to(topicExchange()).with(TOPIC_ROUTE_KEY2);
39.     }
41. }

复制代码

编写 junit 测试，投递两条差异路由的消息给主题互换机

1. @Test
2. public void topicTest(){
3.     //消息被投递给 topicQ1 和 topicQ2
4.     rabbitTemplate.convertAndSend(TopicExchangeConfig.TOPIC_EXCHANGE, "topic.route.1", "topicTest-*");
5.     //消息仅投递给 topicQ2
6.     rabbitTemplate.convertAndSend(TopicExchangeConfig.TOPIC_EXCHANGE, "topic.route.1.1", "topicTest-#");
7. }

复制代码

可以从管理端看到 topicQueue2 由于绑定的是通配符#，故此两条消息都有被投递到队列中，topicQueue1 由于绑定的是通配符* 只匹配到一条消息，故此只有一条消息被投递到队列中。

首部互换机

首部互换机通过设置消息的头部信息来举行绑定队列的分发，它不依赖于路由键的匹配规则来分发消息，而是根据发送的消息内容中的headers属性举行匹配。当消息投递到首部互换器时，RabbitMQ会获取到该消息的headers（一个键值对的情势），而且对比此中的键值对是否完全匹配队列和互换器绑定时指定的键值对。如果完全匹配，则消息会路由到该队列，否则不会路由到该队列。

1. @Configuration
2. public class HeadExchangeConfig {
4.     public static final String HEADER_QUEUE1 = "header.queue1";
5.     public static final String HEADER_QUEUE2 = "header.queue2";
6.     public static final String HEADER_QUEUE3 = "header.queue3";
7.     public static final String HEADER_EXCHANGE = "header.exchange";
9.     public static final String HEADER_KEY1 = "headerKey1";
10.     public static final String HEADER_KEY2 = "headerKey2";
12.     //声明queue
13.     @Bean("headerQueue1")
14.     public Queue headerQueue1() {
15.         return new Queue(HEADER_QUEUE1);
16.     }
18.     @Bean("headerQueue2")
19.     public Queue headerQueue2() {
20.         return new Queue(HEADER_QUEUE2);
21.     }
23.     @Bean("headerQueue3")
24.     public Queue headerQueue3() {
25.         return new Queue(HEADER_QUEUE3);
26.     }
28.     //声明首部交换机
29.     @Bean
30.     public HeadersExchange headerExchange() {
31.         return new HeadersExchange(HEADER_EXCHANGE);
32.     }
34.     //声明Binding,绑定Header(消息头部参数)中 HEADER_KEY1 = a的队列。header的队列匹配可以用mathces和exisits
35.     @Bean
36.     public Binding bindHeaderQueue1() {
37.         return BindingBuilder.bind(headerQueue1()).to(headerExchange()).where(HEADER_KEY1).matches("a");
38.     }
40.     //绑定Header中 HEADER_KEY2 =1的队列。
41.     @Bean
42.     public Binding bindHeaderBusTyp1() {
43.         return BindingBuilder.bind(headerQueue2()).to(headerExchange()).where(HEADER_KEY2).matches("b");
44.     }
46.     //绑定Header中 HEADER_KEY1 = a 或者 HEADER_KEY2 = b 的队列。
47.     @Bean
48.     public Binding bindHeaderTxBusTyp1() {
49.         Map<String, Object> condMap = new HashMap<>();
50.         condMap.put(HEADER_KEY1, "a");
51.         condMap.put(HEADER_KEY2, "b");
52.         return BindingBuilder.bind(headerQueue3()).to(headerExchange()).whereAny(condMap).match();
53.     }
55. }

复制代码

编写 junit 测试， 往首部互换机投递信息

1. @Test
2. public void headerTest(){
3.     MessageProperties properties = new MessageProperties();
4.     properties.setHeader(HeadExchangeConfig.HEADER_KEY1, "a");
5.    
6.     //消息被投递到 headerQueue1 和 headerQueue3
7.     rabbitTemplate.convertAndSend(HeadExchangeConfig.HEADER_EXCHANGE, "", new Message("headerTest-a".getBytes(), properties));
8.     properties.setHeader(HeadExchangeConfig.HEADER_KEY1, "");
9.     properties.setHeader(HeadExchangeConfig.HEADER_KEY2, "b");
10.    
11.     //消息被投递到 headerQueue2 和 headerQueue3
12.     rabbitTemplate.convertAndSend(HeadExchangeConfig.HEADER_EXCHANGE, "", new Message("headerTest-b".getBytes(), properties));
14. }

复制代码

备份互换机

通过设置互换机的alternate-exchange的参数设置备份互换机，当消息路由无法在当前互换机匹配到符合的队列投递时，将消息转到备份互换机，分发到其绑定的备份队列中。

1. @Configuration
2. public class BackupExchangeConfig {
4.     public static final String BACKUP_QUEUE = "backup.queue";
5.     public static final String BACKUP_EXCHANGE = "backup.exchange";
6.     public static final String BACKUP_ROUTE_KEY = "backup.key";
8.     public static final String NON_BACKUP_QUEUE = "nonbackup.queue";
9.     public static final String NON_BACKUP_EXCHANGE = "nonbackup.exchange";
10.     public static final String NON_BACKUP_ROUTE_KEY = "nonbackup.key";
12.     @Bean("backupQueue")
13.     public Queue backupQueue(){
14.         return new Queue(BACKUP_QUEUE, true, false, false);
15.     }
17.     @Bean("nonBackupQueue")
18.     public Queue nonBackupQueue(){
19.         return new Queue(NON_BACKUP_QUEUE, true, false, false);
20.     }
22.     @Bean("nonBackupExchange")
23.     public DirectExchange nonBackupExchange(){
24.         Map<String, Object> args = new HashMap<>(2);
25.         args.put("alternate-exchange", BACKUP_EXCHANGE);
26.         return new DirectExchange(NON_BACKUP_EXCHANGE,true,false, args);
27.     }
29.     @Bean("backupExchange")
30.     public FanoutExchange backupExchange(){
31.         return new FanoutExchange(BACKUP_EXCHANGE,true,false);
32.     }
34.     @Bean("bindNonBackupQueue")
35.     public Binding bindNonBackupQueue(){
36.         return BindingBuilder.bind(nonBackupQueue()).to(nonBackupExchange()).with(NON_BACKUP_ROUTE_KEY);
37.     }
39.     @Bean("bindBackupQueue")
40.     public Binding bindBackupQueue(){
41.         return BindingBuilder.bind(backupQueue()).to(backupExchange());
42.     }
44. }

复制代码

编写测试用例， 投递消息给备份互换机

1. @Test
2. public void backupTest() {
3.     //路由正确匹配，消息投递到非备份队列中
4.     rabbitTemplate.convertAndSend(BackupExchangeConfig.NON_BACKUP_EXCHANGE, BackupExchangeConfig.NON_BACKUP_ROUTE_KEY, "nonBackupTest");
5.     //路由无法匹配，消息投递到备份队列中
6.     rabbitTemplate.convertAndSend(BackupExchangeConfig.NON_BACKUP_EXCHANGE, BackupExchangeConfig.NON_BACKUP_ROUTE_KEY + "123", "backupTest");
7. }

复制代码

可以看到备份队列和非备份队列中各有一条消息。

死信互换机

死信互换机着实可以明确成一个拥有特殊意义的直连互换机，正常队列通过设置队列中的x-dead-letter-exchange 和 x-dead-letter-routing-key 参数来设置绑定死信互换机，当斲丧者拒绝斲丧、消息积蓄队列到达最大长度大概消息逾期时，消息从正常队列转到死信队列。
死信在转移到死信队列时，它的路由也会生存下来。但是如果设置了x-dead-letter-routing-key参数的话，路由就会被更换为设置的这个值。别的，死信在转移到死信队列的过程中，是没有颠末消息发送者确认的，以是并不能包管消息的安全性。
消息被作为死信转移到死信队列后，会在Header当中增长一些消息。比如时间、缘故因由(rejected,expired,maxlen)、队列等。别的header中还会加上第一次成为死信的三个属性(x-first-death-reason，  x-first-death-queue， x-first-death-exchange)，而且这三个属性在以后的转达过程中都不会更改。
死信队列也可以向别的队列一样被斲丧者正常订阅斲丧。

1. @Configuration
2. public class DeadLetterExchangeConfig {
4.     public static final String DEAD_QUEUE = "dead.queue";
5.     public static final String DEAD_EXCHANGE = "dead.exchange";
6.     public static final String DEAD_ROUTE_KEY = "dead.key";
8.     public static final String NON_DEAD_QUEUE = "nondead.queue";
9.     public static final String NON_DEAD_EXCHANGE = "nondead.exchange";
10.     public static final String NON_DEAD_ROUTE_KEY = "nondead.key";
12.     @Bean("deadQueue")
13.     public Queue deadQueue(){
14.         return new Queue(DEAD_QUEUE, true, false, false);
15.     }
17.     @Bean("nonDeadQueue")
18.     public Queue nonDeadQueue(){
19.         Map<String, Object> args = new HashMap<>(2);
20.         args.put("x-message-ttl",10000);
21.         args.put("x-dead-letter-exchange",DEAD_EXCHANGE);
22.         args.put("x-dead-letter-routing-key",DEAD_ROUTE_KEY);
23.         return new Queue(NON_DEAD_QUEUE, true, false, false, args);
24.     }
26.     @Bean("deadExchange")
27.     public DirectExchange deadExchange(){
28.         return new DirectExchange(DEAD_EXCHANGE,false,false);
29.     }
31.     @Bean("nonDeadExchange")
32.     public DirectExchange nonDeadExchange(){
33.         return new DirectExchange(NON_DEAD_EXCHANGE,true,false);
34.     }
36.     @Bean("bindDeadQueue")
37.     public Binding bindDeadQueue(){
38.         return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(DEAD_ROUTE_KEY);
39.     }
41.     @Bean("bindNonDeadQueue")
42.     public Binding bindNonDeadQueue(){
43.         return BindingBuilder.bind(nonDeadQueue()).to(nonDeadExchange()).with(NON_DEAD_ROUTE_KEY);
44.     }
46. }

复制代码

编写 junit 测试， 投递一条10秒逾期的消息，刚投递时消息存在于正常队列，10秒逾期后转到死信队列。 投递消息和队列同时设置逾期时间时，以时间更短的为准。

1. @Test
2. public void deadTest() {
3.     rabbitTemplate.convertAndSend(DeadLetterExchangeConfig.NON_DEAD_EXCHANGE, DeadLetterExchangeConfig.NON_DEAD_ROUTE_KEY, "deadTest 1");
4. }

复制代码

延时队列

通过TTL+死信队列实现延时队列，与上述死信互换机使用大同小异，焦点就是创建队列的时间设置如下三个参数：

• x-message-ttl (须要) ：当前队列消息多久未斲丧进入死信队列
  
• x-dead-letter-exchange (须要)：消息逾期后进入的死信队列互换机
  
• x-dead-letter-routing-key (非须要)：消息的路由， 未设置时生存原队列的路由
  

TTL 消息可以通过以下方式创建
方式一：在队列中设置 x-message-ttl 参数

1. @Bean("nonDeadQueue")
2. public Queue nonDeadQueue(){
3.     Map<String, Object> args = new HashMap<>(2);
4.     args.put("x-message-ttl",10000);
5.     args.put("x-dead-letter-exchange",DEAD_EXCHANGE);
6.     args.put("x-dead-letter-routing-key",DEAD_ROUTE_KEY);
7.     return new Queue(NON_DEAD_QUEUE, true, false, false, args);
8. }

复制代码

方式二： 在投递消息时设置消息的逾期时间

1. MessageProperties properties = new MessageProperties();
2. properties.setExpiration("10000");
3. rabbitTemplate.convertAndSend(DeadLetterExchangeConfig.NON_DEAD_EXCHANGE, DeadLetterExchangeConfig.NON_DEAD_ROUTE_KEY, new Message("ttl test".getBytes(), properties));

复制代码

两种方式同时都有设置时，时间短的设置见效。
动态创建队列、互换机及绑定关系

Spring Boot 封装了一些类用于对 RabbitMQ 的管理

• AmqpAdmin
  用于管理队列、互换机及绑定关系 。
  
• RabbitTemplate
  对消息操纵的一些封装。
  

1. @Autowired
2. private AmqpAdmin amqpAdmin;
4. public void createComponents(){
6.         String queueName = "amqp.queue";
7.         String exchangeName = "amqp.exchange";
8.        
9.         //声明(创建)队列
10.         Queue queue = new Queue(queueName, false, false, false, null);
11.         amqpAdmin.declareQueue(queue);
12.        
13.         //声明交换机
14.         FanoutExchange fanoutExchange = new FanoutExchange(exchangeName, false, false, null);
15.         amqpAdmin.declareExchange(fanoutExchange);
16.        
17.         //声明绑定
18.         amqpAdmin.declareBinding(new Binding(queueName, Binding.DestinationType.QUEUE, exchangeName, "", null));
19. }

复制代码

消息监听

启动类添加@EnableRabbit注解启用RabbitMQ ，通过注解 @RabbitListener 和 @RabbitHandler 举行消息的斲丧

1. @Component
2. @RabbitListener(queues = "fanout.queue1")
3. public class FanoutQueueRecever {
5.     @RabbitHandler
6.     public void handle(String msg){
7.         System.out.println("收到来自 fanout.queue1 的消息 ：" + msg);
8.     }
9. }

复制代码

消息确认

包管发送方消息不丢失

开启生产端确认， 消息发送乐成后回调，得到预期效果后才以为消息发送乐成。

• 互换机收到消息举行回调，ConfirmCallback
  spring.rabbitmq.publisher-confirm-type: correlated （高版本Spring使用）
  spring.rabbitmq.publisher-confirms: true（低版本Spring使用）
  
• 消息精确抵达队摆列行回调，ReturnsCallback
  spring.rabbitmq.publisher-returns: true
  spring.rabbitmq.template.mandatory: true， 只要抵达队列，以异步情势优先发送回调 ReturnCallback
  

包管斲丧者消息不丢失

开启斲丧端确认(包管每个消息都被精确斲丧，此时才可以删除这个消息)

• 手动ack消息
  spring.rabbitmq.listener.simple.acknowledge-mode: manual
  
  
  
  • AcknowledgeMode.NONE：自动确认
    自动确认会在消息发送给斲丧者后立即确认，但存在丢失消息的大概，如果斲丧端斲丧逻辑抛出非常，也就是斲丧端没有处理处罚乐成这条消息，那么就相当于丢失了消息。
    如果消息已经被处理处罚，但后续出现非常导致事件回滚,也同样造成了实际意义的消息丢失。
    
  • AcknowledgeMode.AUTO：根据情况确认
    
  • AcknowledgeMode.MANUAL：手动确认
    如果手动确认则当斲丧者调用 ack、nack、reject 几种方法举行确认，手动确认可以在业务失败后举行一些操纵，如果消息未被 ACK 则会发送到下一个斲丧者。
    如果某个服务忘记 ACK 了，则 RabbitMQ 不会再发送数据给它，由于 RabbitMQ 以为该服务的处理处罚本事有限，不敷以处理处罚后续投递的消息。
    
  
  
• ACK的几种方法
  
  
  
  • channel.basicNack(deliveryTag, multiple, requeue); 拒绝斲丧。
    deliveryTag（唯一标识 ID）：当一个斲丧者向 RabbitMQ 注册后，会创建起一个 Channel ，RabbitMQ 会用 basic.deliver 方法向斲丧者推送消息，这个方法携带了一个 delivery tag， 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID，是一个单调递增的正整数，delivery tag 的范围仅限于 Channel。
    multiple：为了镌汰网络流量，手动确认可以被批处理处罚，当该参数为 true 时，则可以一次性确认 delivery_tag 小于便是传入值的全部消息。
    requeue： 是否重新入队
    
  • channel.basicAck(deliveryTag, multiple); 确认斲丧，参数表明同上。
    
  • channel.basicReject(deliveryTag, requeue); 拒绝斲丧，不支持批量操纵，用法与basicNack()类似。
    
  
  

代码实现

yaml 文件设置

1. spring:
2.   rabbitmq:
3.     host: 127.0.0.1
4.     port: 5672
5.     username: guest
6.     password: guest
7.     # 开启发送端确认    #publisher-confirms: true    publisher-confirm-type: correlated    #开启发送端消息抵达确认    publisher-returns: true    #只要抵达队列。以异步发送优先回调returnconfirm    template:      mandatory: true    # 手动ack消息    listener:      simple:        acknowledge-mode: manual

复制代码

设置RabbitMQ， 设置发送者消息确认逻辑

1. @Configuration
2. public class RabbitMqConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback{
4.     @Autowired
5.     RabbitTemplate rabbitTemplate;
7.     @PostConstruct
8.     public void init(){
9.         rabbitTemplate.setReturnsCallback(this);
10.         rabbitTemplate.setConfirmCallback(this);
11.     }
13.     /**
14.      * 发送消息触发confirmCallback回调, 无论是否到达队列，只要有到达交换机都会触发这个回调
15.      * @param correlationData：当前消息的唯一关联数据（如果发送消息时未指定此值，则回调时返回null）
16.      * @param ack：消息是否成功收到（ack=true，消息抵达Broker）
17.      * @param cause：失败的原因
18.      */
19.     @Override
20.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
21.         System.out.println("发送消息触发confirmCallback回调");
22.         System.out.println(String.format("correlationData：%s\nack：%s\ncause：%s", correlationData, ack, cause));
23.     }
25.     /**
26.      * 消息未到达队列触发returnCallback回调,只要消息没有投递给指定的队列，就触发这个失败回调
27.      * @param returnedMessage 返回的消息，包含
28.      *                        message：投递失败的消息详细信息
29.      *                        replyCode：回复的状态码
30.      *                        replyText：回复的文本内容
31.      *                        exchange：接收消息的交换机
32.      *                        routingKey：接收消息的路由键
33.      */
34.     @Override
35.     public void returnedMessage(ReturnedMessage returnedMessage) {
36.         System.out.println("消息未到达队列触发returnCallback回调");
37.         System.out.println(returnedMessage.toString());
38.     }
39. }

复制代码

实现斲丧者确认

1. @Component
2. @RabbitListener(queues = "fanout.queue1")
3. public class FanoutQueueRecever {
5.     @RabbitHandler
6.     public void handle(String msg, Channel channel, @Headers Map<String, Object> headers){
7.         System.out.println("收到来自 fanout.queue1 的消息 ：" + msg);
8.         System.out.println("tag：" + headers.get(AmqpHeaders.DELIVERY_TAG));
9.         try {
10.             if("ack".equalsIgnoreCase(msg)){
11.                 channel.basicAck((long)headers.get(AmqpHeaders.DELIVERY_TAG), false);
12.             }
13.             if("nack".equalsIgnoreCase(msg)){
14.                 channel.basicNack((long)headers.get(AmqpHeaders.DELIVERY_TAG), false, true);
15.             }
16.             if("reject".equalsIgnoreCase(msg)){
17.                 channel.basicReject((long)headers.get(AmqpHeaders.DELIVERY_TAG), false);
18.             }
20.         } catch (IOException e) {
21.             e.printStackTrace();
22.         }
23.     }
25.     /**
26.      * @Payload 注解的对象需要实现序列化
27.      * @Headers 获取所有头部信息
28.      * @Header  获取单个头部信息
29.      */
30.     @RabbitHandler(isDefault = true)
31.     public void handleMap(@Payload MyMessage message, Channel channel, @Headers Map<String, Object> headers){
32.         System.out.println("收到来自 fanout.queue1 的消息 ：" + message.toString());
33.         System.out.println("tag：" + headers.get(AmqpHeaders.DELIVERY_TAG));
34.         try {
35.             channel.basicAck((long)headers.get(AmqpHeaders.DELIVERY_TAG), false);
36.         } catch (IOException e) {
37.             e.printStackTrace();
38.         }
39.     }
41. }

复制代码

编写 junit 测试， 发送测试消息

1. @Data
2. @AllArgsConstructor
3. @NoArgsConstructor
4. @EqualsAndHashCode
5. public class MyMessage implements Serializable {
7.     private String id;
9.     private String name;
11. }

复制代码

1. @Test
2. public void confirmTest(){
3.     //未知交换机， 触发 confirmCallback 回调
4.     //rabbitTemplate.convertAndSend("unknow", "unknow", "confirmTest");
5.     //未知路由， 消息到达交换机但是无法到达队列， 触发 confirmCallback 回调和 returnCallback 回调
6.     //rabbitTemplate.convertAndSend(TopicExchangeConfig.TOPIC_EXCHANGE, "topic", "topicTest-*");
8.     //正常到达队列，触发confirmCallback回调
9.     rabbitTemplate.convertAndSend(FanoutExchangeConfig.FANOUT_EXCHANGE, "", "fantoutTest", new CorrelationData("1"));
10.     rabbitTemplate.convertAndSend(FanoutExchangeConfig.FANOUT_EXCHANGE, "", new MyMessage("1", "张三"), new CorrelationData("2"));
12. }

复制代码