RabbitMQ交換機與Springboot整合的簡單實現

Posted on 2021-07-24 by WalkonNet

RabbitMQ-交換機

1、交換機是幹什麼的？
消息(Message)由Client發送，RabbitMQ接收到消息之後通過交換機轉發到對應的隊列上面。Worker會從隊列中獲取未被讀取的數據處理。

在這裡插入圖片描述

1、交換機的種類
RabbitMQ包含四種不同的交換機類型：

Direct exchange：直連交換機，轉發消息到routigKey指定的隊列
Fanout exchange：扇形交換機，轉發消息到所有綁定隊列（速度最快）
Topic exchange：主題交換機，按規則轉發消息（最靈活）
Headers exchange：首部交換機（不常用）

1、Direct exchange（直連交換機）

直連交換機是一種帶路由功能的交換機，根據消息攜帶的路由鍵將消息投遞給對應隊列。
一個隊列通過routing_key（路由鍵）與一個交換機綁定，當消息被發送的時候，需要指定一個routing_key，這個消息被送達交換機的時候，就會被交換機送到指定的隊列裡面去。
類似一對一的關系！

在這裡插入圖片描述

1.1 Springboot的簡單實現

創建2個springboot項目，一個 rabbitmq-provider （生產者），一個rabbitmq-consumer（消費者）。

1、首先創建 rabbitmq-provider

pom.xml添加依賴：

	<!--rabbitmq-->
 	 <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-amqp</artifactId>
     </dependency>

配置application.yml：

server:
  port: 8090
spring:
  application:
    name: rabbitmq-provider
  rabbitmq:
    host: 192.168.152.173
    port: 5672     #註意15672是web頁面端口，消息發送端口為5672
    username: guest
    password: guest

編寫配置文件

@Configuration
public class DirectRabbitConfig {

    /**
     * 創建消息隊列 起名：TestDirectQueue
     * durable:是否持久化,默認是false,持久化隊列：會被存儲在磁盤上，當消息代理重啟時仍然存在，暫存隊列：當前連接有效
     * exclusive:默認也是false，隻能被當前創建的連接使用，而且當連接關閉後隊列即被刪除。此參考優先級高於durable
     * autoDelete:是否自動刪除，當沒有生產者或者消費者使用此隊列，該隊列會自動刪除。
     * 一般設置一下隊列的持久化就好,其餘兩個就是默認false
     * @return new Queue(name,durable,exclusive,autoDelete)
     */
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue",true);
    }

    /**
     * Direct交換機 起名：TestDirectExchange
     */
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange",true,false);
    }

    /**
     * 綁定 
     * 將隊列和交換機綁定, 並設置用於匹配鍵（路由鍵）：TestDirectRouting
     */
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }   
}

編寫controller層

@RestController
@RequestMapping("/rabbit")
public class SendMessageController {

    @Autowired
    private RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,這提供瞭接收/發送等等方法

    @RequestMapping("/send")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "League of Legends never dies！";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        //將消息攜帶綁定鍵值：TestDirectRouting 發送到交換機TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";
    }
}

把rabbitmq-provider項目運行，調用下接口:

在這裡插入圖片描述

因為我們目前還沒弄消費者 rabbitmq-consumer，消息沒有被消費的，我們去rabbitMq管理頁面看看，是否推送成功：

在這裡插入圖片描述

消息已經推送到rabbitMq服務器上面瞭。

2、創建rabbitmq-consumer項目：
pom.xml和application.yml跟上邊一樣（把端口號和應用名換瞭即可）

創建消息接收監聽類，DirectReceiver01.java：

@Component
@RabbitListener(queues = "TestDirectQueue")//監聽的隊列名稱 TestDirectQueue
public class DirectReceiver01 {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("DirectReceiver01消費者收到的消息: " + testMessage.toString());
    }
}

啟動消費者項目：
運行結果如下

在這裡插入圖片描述

兩個消息直接被消費瞭。然後可以再繼續調用rabbitmq-provider項目的推送消息接口，可以看到消費者即時消費消息！

擴展
直連交換機既然是一對一，那如果咱們配置多臺監聽綁定到同一個直連交互的同一個隊列，會怎麼樣？
直接復制兩份DirectReceiver.java，改下打印內容！

在這裡插入圖片描述

重新啟動項目！在生產者多發送幾次消息。運行結果如下：

在這裡插入圖片描述

可以看到是實現瞭輪詢的方式對消息進行消費，而且不存在重復消費。

2、Topic Exchange（主題交換機）

直連交換機的routing_key方案非常簡單，但是它是一對一的關系，那麼我們需要一對多呢？希望一條消息發送給多個隊列。所以RabbitMQ提供瞭一種主題交換機，發送到主題交換機上的消息需要攜帶指定規則的routing_key，主題交換機會根據這個規則將數據發送到對應的(多個)隊列上。

主題交換機的routing_key需要有一定的規則，交換機和隊列的binding_key需要采用 *.# 的格式，每個部分用.分開，其中：
*表示一個單詞
#表示任意數量（零個或多個）單詞。（例如：topic.man）

在這裡插入圖片描述

1、在rabbitmq-provider項目裡面創建TopicRabbitConfig.java：

@Configuration
public class TopicRabbitConfig {
    //綁定鍵
      public final static String man = "topic.man";
      public final static String woman = "topic.woman";

      @Bean
      public Queue firstQueue() {
          return new Queue(TopicRabbitConfig.man);
      }

      @Bean
      public Queue secondQueue() {
          return new Queue(TopicRabbitConfig.woman);
      }

      @Bean
      TopicExchange exchange() {
          return new TopicExchange("topicExchange");
      }


      //將firstQueue和topicExchange綁定,而且綁定的鍵值為topic.man
      //這樣隻要是消息攜帶的路由鍵是topic.man,才會分發到該隊列
      @Bean
      Binding bindingExchangeMessage() {
          return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
      }

      //將secondQueue和topicExchange綁定,而且綁定的鍵值為用上通配路由鍵規則topic.#
      // 這樣隻要是消息攜帶的路由鍵是以topic.開頭,都會分發到該隊列
      @Bean
      Binding bindingExchangeMessage2() {
          return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
      }
}

編寫controller層

@RequestMapping("rabbit")
@RestController
public class TopicSendController {
    @Autowired
    private RabbitTemplate template;

    @GetMapping("/sendTopicMessage1")
    public String sendTopicMessage1() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: M A N ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

        Map<String, Object> manMap = new HashMap<>();
        manMap.put("messageId", messageId);
        manMap.put("messageData", messageData);
        manMap.put("createTime", createTime);
        template.convertAndSend("topicExchange", "topic.man", manMap);
        return "ok";
    }

    @GetMapping("/sendTopicMessage2")
    public String sendTopicMessage2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: woman is all ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

        Map<String, Object> womanMap = new HashMap<>();
        womanMap.put("messageId", messageId);
        womanMap.put("messageData", messageData);
        womanMap.put("createTime", createTime);
        template.convertAndSend("topicExchange", "topic.woman", womanMap);
        return "ok";
    }
}

不要急著去運行，先去編寫消費者,創建兩個接口

TopicManReceiver :

@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicManReceiver消費者收到消息  : " + testMessage.toString());
    }
}

TopicWoManReceiver :

@Component
@RabbitListener(queues = "topic.woman")
public class TopicWoManReceiver {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicWoManReceiver消費者收到消息  : " + testMessage.toString());
    }
}

然後把rabbitmq-provider，rabbitmq-consumer兩個項目都跑起來，先調用/sendTopicMessage1 接口：

在這裡插入圖片描述

然後看消費者rabbitmq-consumer的控制臺輸出情況：
TopicManReceiver監聽隊列1，綁定鍵為：topic.man
TopicTotalReceiver監聽隊列2，綁定鍵為：topic.#
而當前推送的消息，攜帶的路由鍵為：topic.man

在這裡插入圖片描述

所以可以看到兩個監聽消費者receiver都成功消費到瞭消息，因為這兩個recevier監聽的隊列的綁定鍵都能與這條消息攜帶的路由鍵匹配上。

在這裡插入圖片描述

用/sendTopicMessage2接口：

在這裡插入圖片描述

然後看消費者rabbitmq-consumer的控制臺輸出情況：
TopicManReceiver監聽隊列1，綁定鍵為：topic.man
TopicTotalReceiver監聽隊列2，綁定鍵為：topic.#
而當前推送的消息，攜帶的路由鍵為：topic.woman

在這裡插入圖片描述

所以可以看到兩個監聽消費者，隻有TopicWoManReceiver 成功消費到瞭消息，

在這裡插入圖片描述

3、Fanout exchange（扇形交換機）

扇形交換機是最基本的交換機類型，它做的事情很簡單–廣播信息。Fanout交換機會把接收到的消息全部轉發到綁定的隊列上。因為廣播不需要“思考”，所以Fanout交換機是四種交換機中速度最快的。

在這裡插入圖片描述

同樣地，先在rabbitmq-provider項目上創建FanoutRabbitConfig.java：

@Configuration
public class FanoutRabbitConfig {

    /**
     * 創建三個隊列 ：fanout.A   fanout.B  fanout.C
     * 將三個隊列都綁定在交換機 fanoutExchange 上
     * 因為是扇型交換機, 路由鍵無需配置,配置也不起作用
     */

    @Bean
    public Queue queueA() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue queueB() {
        return new Queue("fanout.B");
    }
    @Bean
    public Queue queueC() {
        return new Queue("fanout.C");
    }
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchangeC() {
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

編寫controller層：

@RestController
@RequestMapping("rabbit")
public class FanoutController {

    @Autowired
    private RabbitTemplate template;

    @GetMapping("/sendFanoutMessage")
    public String sendFanoutMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: testFanoutMessage ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        template.convertAndSend("fanoutExchange", null, map);
        return "ok";
    }
}

接著在rabbitmq-consumer項目裡加上消息消費類，

FanoutReceiverA:

@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費者收到消息  : " +testMessage.toString());
    }
}

FanoutReceiverB

@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB{
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費者收到消息  : " +testMessage.toString());
    }
}

FanoutReceiverC :

@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費者收到消息  : " +testMessage.toString());
    }
}

最後將rabbitmq-provider和rabbitmq-consumer項目都跑起來，調用下接口/sendFanoutMessage，運行結果如下：

在這裡插入圖片描述

可以看到隻要發送到 fanoutExchange 這個扇型交換機的消息，三個隊列都綁定這個交換機，所以三個消息接收類都監聽到瞭這條消息。

4、消息確認（發送確認與接收確認） 4.1發送確認

在rabbitmq-provider項目中
編寫配置文件：

server:
  port: 8090
spring:
  application:
    name: rabbitmq-consumer
  rabbitmq:
    host: 192.168.152.173
    port: 5672
    username: guest
    password: guest
    #確認消息已發送到交換機(Exchange)
    publisher-confirm-type: correlated
    #確認消息已發送到隊列(Queue)
    publisher-returns: true

配置相關的消息確認回調函數，RabbitConfig.java：

/**
 * 配置相關的消息確認回調函數
 */
@Configuration
public class RabbitConfig {

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //設置開啟Mandatory,才能觸發回調函數,無論消息推送結果怎麼樣都強制調用回調函數
        rabbitTemplate.setMandatory(true);

        //確認回調
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean condition, String cause) {
                System.out.println("ConfirmCallback:     " + "相關數據：" + correlationData);
                System.out.println("ConfirmCallback:     " + "確認情況：" + condition);
                System.out.println("ConfirmCallback:     " + "原因：" + cause);
            }
        });

        //設置返回消息的回調
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     " + "消息：" + message);
                System.out.println("ReturnCallback:     " + "回應碼：" + replyCode);
                System.out.println("ReturnCallback:     " + "回應信息：" + replyText);
                System.out.println("ReturnCallback:     " + "交換機：" + exchange);
                System.out.println("ReturnCallback:     " + "路由鍵：" + routingKey);
            }
        });
        return rabbitTemplate;
    }
}

ConfirmCallback
ConfirmCallback是一個回調接口，就是隻確認是否正確到達 Exchange 中。
ReturnCallback
通過實現 ReturnCallback 接口，啟動消息失敗返回，此接口是在交換器路由不到隊列時觸發回調，該方法可以不使用，因為交換器和隊列是在代碼裡綁定的，如果存在綁定隊列失敗，那就是你代碼寫錯瞭。

從總體的情況分析，推送消息存在四種情況：

1、消息推送到server，但是在server裡找不到交換機
2、消息推送到server，找到交換機瞭，但是沒找到隊列
3、消息推送到sever，交換機和隊列啥都沒找到
4、消息推送成功

1、消息推送到server，但是在server裡找不到交換機
寫個測試接口，把消息推送到名為‘non-existent-exchange’的交換機上（這個交換機是沒有創建沒有配置的）：

 @GetMapping("/TestMessageAck")
    public String TestMessageAck() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        //non-existent-exchange這個交換機沒有創建和配置
        template.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
        return "ok";
    }

調用接口，查看rabbitmq-provuder項目的控制臺輸出情況

在這裡插入圖片描述

因為這個交換機沒有創建，消息不能到達到達 Exchange 中。所以會觸發ConfirmCallback這個回調函數

結論： 隻會觸發ConfirmCallback。

2、消息推送到server，找到交換機瞭，但是沒找到隊列
新增一個交換機，但是不給這個交換機綁定隊列，在DirectRabitConfig裡面新增一個直連交換機，不做任何綁定配置操作：

 @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }

同樣編寫測試接口

 @GetMapping("/TestMessageAck2")
    public String TestMessageAck2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
        return "ok";
    }

調用方法,查看運行結果：

在這裡插入圖片描述

可以看到兩個回調函數都被調用瞭；
這種情況下，消息是推送成功到服務器瞭的，所以ConfirmCallback對消息確認情況是true；
而在RetrunCallback回調函數的打印參數裡面可以看到，消息是推送到瞭交換機成功瞭，但是在路由分發給隊列的時候，找不到隊列，所以報瞭錯誤 NO_ROUTE。
結論： 兩個回調函數都會觸發！

3、消息推送到sever，交換機和隊列啥都沒找到
結論： 觸發的是 ConfirmCallback 回調函數。

4、消息推送成功
結論： 觸發的是 ConfirmCallback 回調函數。

總結： ConfirmCallback回調函數總是會被觸發，確認情況true和false是有沒有找到交換機，false的話打印原因。ReturnCallback是交換機路由不到隊列是回調，也就是你交換機沒有綁定隊列，這個回調函數可以不寫，綁定失敗的話就好好看看代碼哪裡寫錯瞭吧。

以上是生產者推送消息的消息確認回調函數的使用介紹！

4.2 接受確認

消費者確認發生在監聽隊列的消費者處理業務失敗，如，發生瞭異常，不符合要求的數據……，這些場景我們就需要手動處理，比如重新發送或者丟棄。

消息確認模式有：

AcknowledgeMode.NONE：自動確認
AcknowledgeMode.AUTO：根據情況確認
AcknowledgeMode.MANUAL：手動確認

參考：https://blog.csdn.net/qq_35387940/article/details/100514134

4.1.1 自動確認

修改消費者配置文件

rabbitmq:
  host: 192.168.152.173
  port: 5672
  username: guest
  password: guest
  listener:
    direct:
      acknowledge-mode: auto   # manual：手動

在消費者項目裡，
新建MessageListenerConfig.java上添加代碼相關的配置代碼：

package com.zsn.demo.config;

import com.zsn.demo.resceiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author： zhouzhou
 * @date：2021/7/22 9:49
 */
@Configuration
public class MessageListenerConfig {
    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private MyAckReceiver myAckReceiver;//消息接收處理類

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默認是自動確認，這裡改為手動確認消息
        //設置一個隊列
        container.setQueueNames("TestDirectQueue");
        //如果同時設置多個如下： 前提是隊列都是必須已經創建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");


        //另一種設置隊列的方法,如果使用這種情況,那麼要設置多個,就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
        //container.addQueues(new Queue("TestDirectQueue2",true));
        //container.addQueues(new Queue("TestDirectQueue3",true));
        container.setMessageListener(myAckReceiver);

        return container;
    }
}

對應的手動確認消息監聽類，MyAckReceiver.java（手動確認模式需要實現 ChannelAwareMessageListener）：

package com.zsn.demo.resceiver;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @author： zhouzhou
 * @date：2021/7/22 9:50
 */
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
    /**
     * 用於處理接收到的 Rabbit 消息的回調。
     * 實現者應該處理指定的信息
     * 通常通過給定的會話發送回復消息
     * @param message the received AMQP message (never <code>null</code>)
     * @param channel the underlying Rabbit Channel (never <code>null</code>)
     * @throws Exception Any.
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因為傳遞消息的時候用的map傳遞,所以將Map從Message內取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點進Message裡面看源碼,單引號直接的數據就是我們的map消息數據
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            System.out.println("  MyAckReceiver  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
            System.out.println("消費的主題消息來自："+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二個參數，手動確認可以被批處理，當該參數為 true 時，則可以一次性確認 delivery_tag 小於等於傳入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二個參數，true會重新放回隊列，所以需要自己根據業務邏輯判斷什麼時候使用拒絕
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

    //{key=value,key=value,key=value} 格式轉換成map
    private Map<String, String> mapStringToMap(String str,int entryNum ) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",entryNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}

需要註意的 basicAck 方法需要傳遞兩個參數

deliveryTag（唯一標識 ID）：當一個消費者向 RabbitMQ 註冊後，會建立起一個 Channel ，RabbitMQ 會用 basic.deliver 方法向消費者推送消息，這個方法攜帶瞭一個 delivery tag，它代表瞭 RabbitMQ 向該 Channel 投遞的這條消息的唯一標識 ID，是一個單調遞增的正整數，delivery tag 的范圍僅限於 Channel
multiple：為瞭減少網絡流量，手動確認可以被批處理，當該參數為 true 時，則可以一次性確認 delivery_tag 小於等於傳入值的所有消息

basicNack方法需要傳遞三個參數

deliveryTag（唯一標識 ID）：上面已經解釋瞭。
multiple：上面已經解釋瞭。
requeue： true ：重回隊列，false ：丟棄，我們在nack方法中必須設置 false，否則重發沒有意義。

basicReject方法需要傳遞兩個參數

deliveryTag（唯一標識 ID）：上面已經解釋瞭。
requeue：上面已經解釋瞭，在reject方法裡必須設置true。

總結

basic.ack用於肯定確認
basic.nack用於否定確認（註意：這是AMQP 0-9-1的RabbitMQ擴展）
basic.reject用於否定確認，但與basic.nack相比有一個限制:一次隻能拒絕單條消息

消費者端以上的3個方法都表示消息已經被正確投遞，但是basic.ack表示消息已經被正確處理。
而basic.nack,basic.reject表示沒有被正確處理：

這時，先調用接口，給直連交換機TestDirectExchange 的隊列TestDirectQueue 推送一條消息，可以看到監聽器正常消費瞭下來：

在這裡插入圖片描述

擴展1
除瞭直連交換機的隊列TestDirectQueue需要變成手動確認以外，我們還需要將一個其他的隊列或者多個隊列也變成手動確認，而且不同隊列實現不同的業務處理。
那麼我們需要做的第一步，往SimpleMessageListenerContainer裡添加多個隊列

在這裡插入圖片描述
MyAckReceiver.java 就可以同時將上面設置到的隊列的消息都消費下來。

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
   /**
     * 用於處理接收到的 Rabbit 消息的回調。
     * 實現者應該處理指定的信息
     * 通常通過給定的會話發送回復消息
     * @param message the received AMQP message (never <code>null</code>)
     * @param channel the underlying Rabbit Channel (never <code>null</code>)
     * @throws Exception Any.
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因為傳遞消息的時候用的map傳遞,所以將Map從Message內取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點進Message裡面看源碼,單引號直接的數據就是我們的map消息數據
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            
            if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消費的消息來自的隊列名為："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消費到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("執行TestDirectQueue中的消息的業務處理流程......");
                
            }
 
            if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消費的消息來自的隊列名為："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消費到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("執行fanout.A中的消息的業務處理流程......");
 
            }
            
            channel.basicAck(deliveryTag, true);
//			channel.basicReject(deliveryTag, true);//為true會重新放回隊列
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
 
    //{key=value,key=value,key=value} 格式轉換成map
    private Map<String, String> mapStringToMap(String str,int enNum) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",enNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}

分別調用這兩個隊列相關接口，運行結果如下：

在這裡插入圖片描述

擴展2： 手動確認的另一種寫法：
我們在實際工作中可能不會這麼寫，一般會寫在server層，自己定義一個接口：
RabbitConsumerService

public interface RabbitConsumerService {
    void process(Channel channel, Message message) throws IOException;
}

@Component
public class RabbitConsumerServiceImpl implements RabbitConsumerService{
    @RabbitHandler
    @RabbitListener(queues = "TestDirectQueue")
    @Override
    public void process(Channel channel, Message message) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因為傳遞消息的時候用的map傳遞,所以將Map從Message內取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點進Message裡面看源碼,單引號直接的數據就是我們的map消息數據
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            System.out.println("  MyAckReceiver  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
            System.out.println("消費的主題消息來自："+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二個參數，手動確認可以被批處理，當該參數為 true 時，則可以一次性確認 delivery_tag 小於等於傳入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二個參數，true會重新放回隊列，所以需要自己根據業務邏輯判斷什麼時候使用拒絕
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
    //{key=value,key=value,key=value} 格式轉換成map
    private Map<String, String> mapStringToMap(String str,int entryNum ) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",entryNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}

@Component這個理論上要換成@service，但是由於我沒有添加日志，所以換成@service會報錯，這裡就是提供一個思路，運行結果如下：

在這裡插入圖片描述

附：
rabbitmq在application.yml文件中的其他配置

rabbitmq:
    addresses: 192.168.152.193:5672
    username: guest
    password: guest  
    #虛擬host 可以不設置,使用server默認host
    virtual-host: /
    listener:
      simple:
        prefetch: 1               #設置一次處理一個消息
        acknowledge-mode: manual  #設置消費端手動 ack
        concurrency: 3            #設置同時有3個消費者消費
    connection-timeout: 500
    # 確認消息是否正確到達queue,如果沒有則觸發，如果有則不觸發
    publisher-returns: true
    #確認消息已發送到交換機(Exchange)
    publisher-confirm-type: correlated
    template:
      #設置為 true後 消費者在消息沒有被路由到合適隊列情況下會被return監聽，而不會自動刪除
      mandatory: true

解釋：Mandatory 設置為 true 則會監聽器會接受到路由不可達的消息，然後處理。如果設置為 false，broker 將會自動刪除該消息。

參考https://blog.csdn.net/qq_35387940/article/details/100514134

到此這篇關於RabbitMQ交換機與Springboot整合的簡單實現的文章就介紹到這瞭,更多相關RabbitMQ Springboot整合內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

RabbitMQ交換機與Springboot整合的簡單實現

RabbitMQ-交換機

推薦閱讀：

發佈留言取消回覆

近期文章

RabbitMQ-交換機

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆