Springboot 整合RabbitMq ，用心看完這一篇就夠了

該篇文章内容較多，包括有rabbitMq相關的一些簡單理論介紹，provider消息推送執行個體，consumer消息消費執行個體，Direct、Topic、Fanout的使用，消息回調、手動确認等。（但是關于rabbitMq的安裝，就不介紹了）

在安裝完rabbitMq後，輸入http://ip:15672/ ，是可以看到一個簡單背景管理界面的。

在這個界面裡面我們可以做些什麼？

可以手動建立虛拟host，建立使用者，配置設定權限，建立交換機，建立隊列等等，還有檢視隊列消息，消費效率，推送效率等等。

以上這些管理界面的操作在這篇暫時不做擴充描述，我想着重介紹後面執行個體裡會使用到的。

首先先介紹一個簡單的一個消息推送到接收的流程，提供一個簡單的圖：

RabbitMq -JCccc

黃色的圈圈就是我們的消息推送服務，将消息推送到中間方框裡面也就是 rabbitMq的伺服器，然後經過伺服器裡面的交換機、隊列等各種關系（後面會詳細講）将資料處理入列後，最終右邊的藍色圈圈消費者擷取對應監聽的消息。

常用的交換機有以下三種，因為消費者是從隊列擷取資訊的，隊列是綁定交換機的（一般），是以對應的消息推送/接收模式也會有以下幾種：

Direct Exchange

直連型交換機，根據消息攜帶的路由鍵将消息投遞給對應隊列。

大緻流程，有一個隊列綁定到一個直連交換機上，同時賦予一個路由鍵 routing key 。

然後當一個消息攜帶着路由值為X，這個消息通過生産者發送給交換機時，交換機就會根據這個路由值X去尋找綁定值也是X的隊列。

Fanout Exchange

扇型交換機，這個交換機沒有路由鍵概念，就算你綁了路由鍵也是無視的。這個交換機在接收到消息後，會直接轉發到綁定到它上面的所有隊列。

Topic Exchange

主題交換機，這個交換機其實跟直連交換機流程差不多，但是它的特點就是在它的路由鍵和綁定鍵之間是有規則的。

簡單地介紹下規則：

* (星号) 用來表示一個單詞 (必須出現的)

# (井号) 用來表示任意數量（零個或多個）單詞

通配的綁定鍵是跟隊列進行綁定的，舉個小例子

隊列Q1 綁定鍵為 *.TT.* 隊列Q2綁定鍵為 TT.#

如果一條消息攜帶的路由鍵為 A.TT.B，那麼隊列Q1将會收到；

如果一條消息攜帶的路由鍵為TT.AA.BB，那麼隊列Q2将會收到；

主題交換機是非常強大的，為啥這麼膨脹？

當一個隊列的綁定鍵為 "#"（井号）的時候，這個隊列将會無視消息的路由鍵，接收所有的消息。

當 * (星号) 和 # (井号) 這兩個特殊字元都未在綁定鍵中出現的時候，此時主題交換機就擁有的直連交換機的行為。

是以主題交換機也就實作了扇形交換機的功能，和直連交換機的功能。

另外還有 Header Exchange 頭交換機，Default Exchange 預設交換機，Dead Letter Exchange 死信交換機，這幾個該篇暫不做講述。

好了，一些簡單的介紹到這裡為止，接下來我們來一起編碼。

本次執行個體教程需要建立2個springboot項目，一個 rabbitmq-provider （生産者），一個rabbitmq-consumer（消費者）。

首先建立 rabbitmq-provider，

pom.xml裡用到的jar依賴：

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

然後application.yml：

server:
  port: 8021
spring:
  #給項目來個名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 伺服器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虛拟host 可以不設定,使用server預設host
    virtual-host: JCcccHost

ps：裡面的虛拟host配置項不是必須的，我自己在rabbitmq服務上建立了自己的虛拟host，是以我配置了；你們不建立，就不用加這個配置項。

那麼怎麼建一個單獨的host呢？假如我就是想給某個項目接入，使用一個單獨host，順便使用一個單獨的賬号，就好像我文中配置的 root 這樣。

其實也很簡便：

virtual-host的建立：

賬号user的建立：

然後記得給賬号配置設定權限，指定使用某個virtual host：

其實還可以特定指定交換機使用權等等：

回歸正題，繼續繼續。

接着我們先使用下direct exchange(直連型交換機),建立DirectRabbitConfig.java（對于隊列和交換機持久化以及連接配接使用設定，在注釋裡有說明，後面的不同交換機的配置就不做同樣說明了）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Configuration
public class DirectRabbitConfig {

    //隊列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        // durable:是否持久化,預設是false,持久化隊列：會被存儲在磁盤上，當消息代理重新開機時仍然存在，暫存隊列：目前連接配接有效
        // exclusive:預設也是false，隻能被目前建立的連接配接使用，而且當連接配接關閉後隊列即被删除。此參考優先級高于durable
        // autoDelete:是否自動删除，當沒有生産者或者消費者使用此隊列，該隊列會自動删除。
        //   return new Queue("TestDirectQueue",true,true,false);

        //一般設定一下隊列的持久化就好,其餘兩個就是預設false
        return new Queue("TestDirectQueue",true);
    }

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

    //綁定  将隊列和交換機綁定, 并設定用于比對鍵：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }



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



}

然後寫個簡單的接口進行消息推送（根據需求也可以改為定時任務等等，具體看需求），SendMessageController.java：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@RestController
public class SendMessageController {

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

    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        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伺服器上面了。

接下來，建立rabbitmq-consumer項目：

pom.xml裡的jar依賴：

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

然後是 application.yml：

server:
  port: 8022
spring:
  #給項目來個名字
  application:
    name: rabbitmq-consumer
  #配置rabbitMq 伺服器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虛拟host 可以不設定,使用server預設host
    virtual-host: JCcccHost

然後一樣，建立DirectRabbitConfig.java（消費者單純的使用，其實可以不用添加這個配置，直接建後面的監聽就好，使用注解來讓監聽器監聽對應的隊列即可。配置上了的話，其實消費者也是生成者的身份，也能推送該消息。）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Configuration
public class DirectRabbitConfig {

    //隊列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue",true);
    }

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

    //綁定  将隊列和交換機綁定, 并設定用于比對鍵：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
}

然後是建立消息接收監聽類，DirectReceiver.java：

@Component
@RabbitListener(queues = "TestDirectQueue")//監聽的隊列名稱 TestDirectQueue
public class DirectReceiver {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("DirectReceiver消費者收到消息  : " + testMessage.toString());
    }

}

然後将rabbitmq-consumer項目運作起來，可以看到把之前推送的那條消息消費下來了：

然後可以再繼續調用rabbitmq-provider項目的推送消息接口，可以看到消費者即時消費消息：

那麼直連交換機既然是一對一，那如果咱們配置多台監聽綁定到同一個直連互動的同一個隊列，會怎麼樣？

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

接着，我們使用Topic Exchange 主題交換機。

在rabbitmq-provider項目裡面建立TopicRabbitConfig.java：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/

@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.#");
    }

}

然後添加多2個接口，用于推送消息到主題交換機：

@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);
        rabbitTemplate.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);
        rabbitTemplate.convertAndSend("topicExchange", "topic.woman", womanMap);
        return "ok";
    }
}

生産者這邊已經完事，先不急着運作，在rabbitmq-consumer項目上，建立TopicManReceiver.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicManReceiver消費者收到消息  : " + testMessage.toString());
    }
}

再建立一個TopicTotalReceiver.java：

package com.elegant.rabbitmqconsumer.receiver;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/

@Component
@RabbitListener(queues = "topic.woman")
public class TopicTotalReceiver {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicTotalReceiver消費者收到消息  : " + testMessage.toString());
    }
}

同樣，加主題交換機的相關配置，TopicRabbitConfig.java（消費者一定要加這個配置嗎？不需要的其實，理由在前面已經說過了。）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/

@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.#");
    }

}

然後把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

是以可以看到兩個監聽消費者隻有TopicTotalReceiver成功消費到了消息。

接下來是使用Fanout Exchang 扇型交換機。

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

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/

@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());
    }
}

然後是寫一個接口用于推送消息，

@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);
        rabbitTemplate.convertAndSend("fanoutExchange", null, map);
        return "ok";
    }

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

FanoutReceiverA.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverA消費者收到消息  : " +testMessage.toString());
    }

}

FanoutReceiverB.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverB消費者收到消息  : " +testMessage.toString());
    }

}

FanoutReceiverC.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費者收到消息  : " +testMessage.toString());
    }

}

然後加上扇型交換機的配置類，FanoutRabbitConfig.java（消費者真的要加這個配置嗎？不需要的其實，理由在前面已經說過了）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@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());
    }
}

最後将rabbitmq-provider和rabbitmq-consumer項目都跑起來，調用下接口/sendFanoutMessage ：

然後看看rabbitmq-consumer項目的控制台情況：

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

到了這裡其實三個常用的交換機的使用我們已經完畢了，那麼接下來我們繼續講講消息的回調，其實就是消息确認（生産者推送消息成功，消費者接收消息成功）。

在rabbitmq-provider項目的application.yml檔案上，加上消息确認的配置項後：

ps：本篇文章使用springboot版本為 2.1.7.RELEASE ;

如果你們在配置确認回調，測試發現無法觸發回調函數，那麼存在原因也許是因為版本導緻的配置項不起效，

可以把publisher-confirms: true 替換為

server:
  port: 8021
spring:
  #給項目來個名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 伺服器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虛拟host 可以不設定,使用server預設host
    virtual-host: JCcccHost
      #确認消息已發送到交換機(Exchange)
    #publisher-confirms: true
    publisher-confirm-type: correlated
      #确認消息已發送到隊列(Queue)
    publisher-returns: true

然後是配置相關的消息确認回調函數，RabbitConfig.java：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@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 ack, String cause) {
                System.out.println("ConfirmCallback:     "+"相關資料："+correlationData);
                System.out.println("ConfirmCallback:     "+"确認情況："+ack);
                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 ，一個叫 RetrunCallback；

那麼以上這兩種回調函數都是在什麼情況會觸發呢？

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

①消息推送到server，但是在server裡找不到交換機

②消息推送到server，找到交換機了，但是沒找到隊列

③消息推送到sever，交換機和隊列啥都沒找到

④消息推送成功

那麼我先寫幾個接口來分别測試和認證下以上4種情況，消息确認觸發回調函數的情況：

①消息推送到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);
        rabbitTemplate.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
        return "ok";
    }

調用接口，檢視rabbitmq-provuder項目的控制台輸出情況（原因裡面有說，沒有找到交換機'non-existent-exchange'）：

2019-09-04 09:37:45.197 ERROR 8172 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'JCcccHost', class-id=60, method-id=40)
ConfirmCallback:     相關資料：null
ConfirmCallback:     确認情況：false
ConfirmCallback:     原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'JCcccHost', class-id=60, method-id=40)

結論： ①這種情況觸發的是 ConfirmCallback 回調函數。

②消息推送到server，找到交換機了，但是沒找到隊列

這種情況就是需要新增一個交換機，但是不給這個交換機綁定隊列，我來簡單地在DirectRabitConfig裡面新增一個直連交換機，名叫‘lonelyDirectExchange’，但沒給它做任何綁定配置操作：

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

然後寫個測試接口，把消息推送到名為‘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";
    }

調用接口，檢視rabbitmq-provuder項目的控制台輸出情況：

ReturnCallback:     消息：(Body:'{createTime=2019-09-04 09:48:01, messageId=563077d9-0a77-4c27-8794-ecfb183eac80, messageData=message: lonelyDirectExchange test message }' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:     回應碼：312
ReturnCallback:     回應資訊：NO_ROUTE
ReturnCallback:     交換機：lonelyDirectExchange
ReturnCallback:     路由鍵：TestDirectRouting

ConfirmCallback:     相關資料：null
ConfirmCallback:     确認情況：true
ConfirmCallback:     原因：null

可以看到這種情況，兩個函數都被調用了；

這種情況下，消息是推送成功到伺服器了的，是以ConfirmCallback對消息确認情況是true；

而在RetrunCallback回調函數的列印參數裡面可以看到，消息是推送到了交換機成功了，但是在路由分發給隊列的時候，找不到隊列，是以報了錯誤 NO_ROUTE 。

結論：②這種情況觸發的是 ConfirmCallback和RetrunCallback兩個回調函數。

③消息推送到sever，交換機和隊列啥都沒找到

這種情況其實一看就覺得跟①很像，沒錯，③和①情況回調是一緻的，是以不做結果說明了。

結論： ③這種情況觸發的是 ConfirmCallback 回調函數。

④消息推送成功

那麼測試下，按照正常調用之前消息推送的接口就行，就調用下 /sendFanoutMessage接口，可以看到控制台輸出：

ConfirmCallback:     相關資料：null
ConfirmCallback:     确認情況：true
ConfirmCallback:     原因：null

結論： ④這種情況觸發的是 ConfirmCallback 回調函數。

以上是生産者推送消息的消息确認回調函數的使用介紹（可以在回調函數根據需求做對應的擴充或者業務資料處理）。

接下來我們繼續，消費者接收到消息的消息确認機制。

和生産者的消息确認機制不同，因為消息接收本來就是在監聽消息，符合條件的消息就會消費下來。

是以，消息接收的确認機制主要存在三種模式：

①自動确認，這也是預設的消息确認情況。 AcknowledgeMode.NONE

RabbitMQ成功将消息發出（即将消息成功寫入TCP Socket）中立即認為本次投遞已經被正确處理，不管消費者端是否成功處理本次投遞。

是以這種情況如果消費端消費邏輯抛出異常，也就是消費端沒有處理成功這條消息，那麼就相當于丢失了消息。

一般這種情況我們都是使用try catch捕捉異常後，列印日志用于追蹤資料，這樣找出對應資料再做後續處理。

② 根據情況确認，這個不做介紹

③ 手動确認，這個比較關鍵，也是我們配置接收消息确認機制時，多數選擇的模式。

消費者收到消息後，手動調用basic.ack/basic.nack/basic.reject後，RabbitMQ收到這些消息後，才認為本次投遞成功。

basic.ack用于肯定确認

basic.nack用于否定确認（注意：這是AMQP 0-9-1的RabbitMQ擴充）

basic.reject用于否定确認，但與basic.nack相比有一個限制:一次隻能拒絕單條消息

消費者端以上的3個方法都表示消息已經被正确投遞，但是basic.ack表示消息已經被正确處理。

而basic.nack,basic.reject表示沒有被正确處理：

着重講下reject，因為有時候一些場景是需要重新入列的。

channel.basicReject(deliveryTag, true); 拒絕消費目前消息，如果第二參數傳入true，就是将資料重新丢回隊列裡，那麼下次還會消費這消息。設定false，就是告訴伺服器，我已經知道這條消息資料了，因為一些原因拒絕它，而且伺服器也把這個消息丢掉就行。下次不想再消費這條消息了。

使用拒絕後重新入列這個确認模式要謹慎，因為一般都是出現異常的時候，catch異常再拒絕入列，選擇是否重入列。

但是如果使用不當會導緻一些每次都被你重入列的消息一直消費-入列-消費-入列這樣循環，會導緻消息積壓。

順便也簡單講講 nack，這個也是相當于設定不消費某條消息。

channel.basicNack(deliveryTag, false, true);

第一個參數依然是目前消息到的資料的唯一id;

第二個參數是指是否針對多條消息；如果是true，也就是說一次性針對目前通道的消息的tagID小于目前這條消息的，都拒絕确認。

第三個參數是指是否重新入列，也就是指不确認的消息是否重新丢回到隊列裡面去。

同樣使用不确認後重新入列這個确認模式要謹慎，因為這裡也可能因為考慮不周出現消息一直被重新丢回去的情況，導緻積壓。

看了上面這麼多介紹，接下來我們一起配置下，看看一般的消息接收手動确認是怎麼樣的。

在消費者項目裡，

建立MessageListenerConfig.java上添加代碼相關的配置代碼：

import com.elegant.rabbitmqconsumer.receiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
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 : JCccc
 * @CreateTime : 2019/9/4
 * @Description :
 **/
@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）：

//之前的相關監聽器可以先注釋掉，以免造成多個同類型監聽器都監聽同一個隊列。

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.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.Map;

@Component

public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            byte[] body = message.getBody();
            ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(body));
            Map<String,String> msgMap = (Map<String,String>) ois.readObject();
            String messageId = msgMap.get("messageId");
            String messageData = msgMap.get("messageData");
            String createTime = msgMap.get("createTime");
            ois.close();
            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();
        }
    }
    
}

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

到這裡，我們其實已經掌握了怎麼去使用消息消費的手動确認了。

但是這個場景往往不夠！因為很多夥伴之前給我評論反應，他們需要這個消費者項目裡面，監聽的好幾個隊列都想變成手動确認模式，而且處理的消息業務邏輯不一樣。

沒有問題，接下來看代碼

場景：除了直連交換機的隊列TestDirectQueue需要變成手動确認以外，我們還需要将一個其他的隊列

或者多個隊列也變成手動确認，而且不同隊列實作不同的業務處理。

那麼我們需要做的第一步，往SimpleMessageListenerContainer裡添加多個隊列：

然後我們的手動确認消息監聽類，MyAckReceiver.java 就可以同時将上面設定到的隊列的消息都消費下來。

但是我們需要做不用的業務邏輯處理，那麼隻需要根據消息來自的隊列名進行區分處理即可，如：

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.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.Map;


@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            byte[] body = message.getBody();
            ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(body));
            Map<String,String> msgMap = (Map<String,String>) ois.readObject();
            String messageId = msgMap.get("messageId");
            String messageData = msgMap.get("messageData");
            String createTime = msgMap.get("createTime");
            ois.close();

            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();
        }
    }


}

ok，這時候我們來分别往不同隊列推送消息，看看效果：

調用接口/sendDirectMessage 和 /sendFanoutMessage ，

Springboot 整合RabbitMq ，用心看完這一篇就夠了

好了，一些簡單的介紹到這裡為止，接下來我們來一起編碼。

接下來我們繼續，消費者接收到消息的消息确認機制。

看了上面這麼多介紹，接下來我們一起配置下，看看一般的消息接收手動确認是怎麼樣的。

到這裡，我們其實已經掌握了怎麼去使用消息消費的手動确認了。

繼續閱讀

關于Springboot中JSCH使用說明1. JSCH簡介2. JSCH依賴3. 使用方法4. 使用連接配接池

路徑錯誤Maven：Non-resolvable parent POM: Failure to find錯誤

jsch連接配接伺服器執行指令傳回的資料中文亂碼問題問題分析解決

LoadRunner 的多種Vuser類型

springboot項目時間不一緻解決方案一、springboot項目時間不一緻解決方案

spring mvc 頁面傳遞數組背景接收方式spring mvc 頁面傳遞數組背景接收方式

Swagger-UI的使用一、添加依賴二、建立配置類三、注解

将swagger-ui中的接口導入postman

聊一聊Swagger-UI的那些事情軟體簡介Swagger 原生Bootstrap-ui Bootstrap的UIKnife4j Bootstrap的更新版的UI

springboot使用 swagger-ui 2.10.5 有關版本更新帶來的問題

rabbitmq 指令行操作

Rabbitmq指令行手動建立隊列

RabbitMQ指令行的基本操作

微服務-性能壓測\緩存redis和分布式鎖redisson和SpringCache

java 8 Lamda 求 List 中 Bigdecimal類型的各種值求最大值求最小值求和求平均值

(SpringBoot)日志種類：log、monitor、access、out、gc、backup

Springboot 整合RabbitMq ，用心看完這一篇就夠了

好了，一些簡單的介紹到這裡為止， 接下來我們來一起編碼。

接下來我們繼續， 消費者接收到消息的消息确認機制。

看了上面這麼多介紹，接下來我們一起配置下，看看一般的消息接收 手動确認是怎麼樣的。 ​​​​​​

到這裡，我們其實已經掌握了怎麼去使用消息消費的手動确認了。

繼續閱讀

好了，一些簡單的介紹到這裡為止，接下來我們來一起編碼。

接下來我們繼續，消費者接收到消息的消息确認機制。

看了上面這麼多介紹，接下來我們一起配置下，看看一般的消息接收手動确認是怎麼樣的。