SpringBoot整合RabbitMQ消息隊列-學習筆記

時間 2019-11-05

標籤 springboot 整合 rabbitmq 消息隊列學習筆記欄目 Spring 简体版

原文原文鏈接

SpringBoot整合RabbitMQ消息隊列-學習筆記

2018年08月30日 14:50:50 Calon Mo 閱讀數 3672html

本文連接：http://www.javashuo.com/article/p-pqjrwton-r.htmlgit

前言

本篇文章主要用於記錄我的學習RabbitMQ的過程，記錄下來方便往後查看，若有錯誤的地方，還望指正。github

本篇文章比較詳細地記錄本人在學習過程當中的每個步驟，比較適合對RabbitMQ不熟的同窗學習，跟着本文操做一遍，就能夠大概知道RabbitMQ的基礎知識了。web

準備階段

首先把RabbitMQ環境安裝好，下面再詳細介紹RabbitMQ各個知識點和如何使用。spring

因爲是基於Centos7的操做系統安裝RabbitMQ-3.7.7。json

爲了方便操做，先把防火牆幹掉，生產環境固然不能這麼幹，我的學習隨意，如下是相關命令：centos

centos7關閉並禁止防火牆啓動命令：瀏覽器

systemctl stop firewalldspringboot
systemctl disable firewalld

RabbitMQ安裝

這裏介紹一種比較簡單的安裝方法-依賴安裝，不用單獨安裝erlang等依賴。

首先到RabbitMQ官網下載：http://www.rabbitmq.com/download.html，

選擇合適你的操做系統版本，本人的操做系統是Centos7.5，因此選擇RHEL/CentOS 7.x這個。

把下載好的rabbitmq-server-3.7.7-1.el7.noarch.rpm放到/home目錄，因爲RabbitMQ-3.7.7須要安裝比較新的erlang-v19.3以上，而yum上並無這麼高的版本，因此須要在/etc/yum.repos.d/目錄下建立文件rabbitmq-erlang.repo，命令以下：

cd /etc/yum.repos.d/
touch rabbitmq-erlang.repo

編輯rabbitmq-erlang.repo命令以下：

vi rabbitmq-erlang.repo

添加如下內容到rabbitmq-erlang.repo：

[rabbitmq-erlang]
name=rabbitmq-erlang
baseurl=https://dl.bintray.com/rabbitmq/rpm/erlang/21/el/7/
gpgcheck=1
gpgkey=https://dl.bintray.com/rabbitmq/Keys/rabbitmq-release-signing-key.asc
repo_gpgcheck=0
enabled=1

上面baseurl是指向erlang-v21版本的連接。

cd到/home目錄，執行安裝RabbitMQ的命令：

yum install -y rabbitmq-server-3.7.7-1.el7.noarch.rpm

這個過程會下載安裝依賴的erlang等依賴，等待安裝完成，會出現下面的界面，則說明RabbitMQ就已經安裝完成了。

Running transaction check
Running transaction test
Transaction test succeeded
Running transaction
正在安裝 : erlang-21.0.5-1.el7.centos.x86_64 1/3
正在安裝 : socat-1.7.3.2-2.el7.x86_64 2/3
正在安裝 : rabbitmq-server-3.7.7-1.el7.noarch 3/3
驗證中 : socat-1.7.3.2-2.el7.x86_64 1/3
驗證中 : rabbitmq-server-3.7.7-1.el7.noarch 2/3
驗證中 : erlang-21.0.5-1.el7.centos.x86_64 3/3
已安裝:
rabbitmq-server.noarch 0:3.7.7-1.el7
做爲依賴被安裝:
erlang.x86_64 0:21.0.5-1.el7.centos socat.x86_64 0:1.7.3.2-2.el7
完畢！

RabbitMQ設置

啓動RabbitMQ服務：

service rabbitmq-server start

剛安裝好的RabbitMQ是尚未用戶的，也不能訪問RabbitMQ的web管理後臺，接下來先添加一個叫root的用戶：

rabbitmqctl add_user root root
rabbitmqctl set_user_tags root administrator
rabbitmqctl set_permissions -p / root "." "." ".*"
#更多命令查看：rabbitmqctl --help

啓用web訪問權限：

rabbitmq-plugins enable rabbitmq_management

重啓RabbitMQ服務：

service rabbitmq-server restart

而後在瀏覽器輸入:http://ip:15672/ ，這時能夠看到RabbitMQ管理頁面了，輸入剛剛添加的帳號root,密碼root便可進入。

登陸進去後界面以下：

RabbitMQ是基於Virtual Host來進行權限控制的，如今爲咱們剛剛添加的root用戶添加一個Virtual Host，在RabbitMQ的web管理後臺，根據下圖進行添加一個virtual host，添加成功後默認分配給root用戶了。

RabbitMQ簡介

RabbitMQ是一個開源的AMQP實現，服務器端用Erlang語言編寫，支持多種語言平臺的客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等。用於在分佈式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。

一般咱們談到消息隊列, 會有三個概念：消息生產者(Provider)、隊列(Queue)、消息消費者(Consumer)，RabbitMQ 在這個基本概念上, 多作了一層抽象, 在消息生產者和隊列之間, 加入了交換器 (Exchange)。這樣消息生產者和隊列就沒有直接聯繫, 變成消息生產者把消息發送給交換器, 交換器根據調度策略再把消息發送給隊列。

左側P表明消息生產者，也就是往RabbitMQ發消息的程序。
中間便是RabbitMQ，其中包括交換機(Exchange)和隊列(Queue)。
右側C表明消費者，也就是往RabbitMQ拿消息的程序。

其中比較重要的概念有：虛擬主機(Virtual Host)、交換機(Exchange)、隊列(Queue)、綁定(Binding)。

虛擬主機(Virtual Hosts)

在上面已經說明如何爲一個用戶建立一個Virtual Host，一個虛擬主機持有一組交換機、隊列和綁定。在RabbitMQ當中，用戶只能在虛擬主機這個粒度上進行權限的控制。若是須要禁止A組訪問B組的交換機/隊列/綁定，必須爲A和B分別建立一個虛擬主機。每個RabbitMQ服務器都有一個默認的虛擬主機「/」。

交換機(Exchange)

交換機的功能主要是接收消息而且根據轉發策略轉發到對應的隊列，交換機不存儲消息，在啓用ack模式後，交換機找不到隊列會返回錯誤，這個ack模式後面再詳細討論。交換機有四種類型：Direct, topic, Headers and Fanout

隊列(Queue)

隊列用於存放消息的載體，通常是和交換機進行綁定，交換機根據轉發策略把消息轉發到隊列裏。

綁定(Binding)

也就是交換機須要和隊列相綁定，這其中如上圖所示，是多對多的關係。

交換機類型介紹

Direct Exchange：

direct 類型的行爲是」先匹配, 再投送」. 即在綁定時設定一個binding_key, 消息的routing_key與binding_key匹配時, 纔會被交換器投送到綁定的隊列中去.

Topic：

轉發消息主要是根據通配符。在這種交換機下，隊列和交換機的綁定會定義一種路由模式，通配符就要在這種路由模式和路由鍵之間匹配後交換機才能轉發消息。

路由鍵必須是一串字符，用句號(.)隔開，好比說 topic.message，或者 topic.message.detail 等。
路由模式必須包含一個星號(*)，主要用於匹配路由鍵指定位置的一個單詞，好比說，一個路由模式是這樣：topic.*，那麼就只能匹配路由鍵是：topic.message、topic.other等，第一個單詞是 topic，第二個單詞能夠是任意一個單詞。井號(#)就表示一個或者多個單詞，例如一個匹配模式是topic.#，那麼能夠匹配到例如：topic.message、topic.message.detail等，以topic.開頭的路由鍵均可以匹配到。

Fanout：

Fanout類型相似於消息廣播，無論路由鍵或者是路由模式，會把消息發給綁定給它的所有隊列，若是配置了routing_key會被忽略。

Headers：

設置header attribute參數類型的交換機

項目簡介

本文是基於Springboot-1.5.15整合RabbitMQ來進行講解，在真實工做中，生產者和消費者通常是在不一樣的項目裏，各自負責不一樣的職責，這裏爲了模擬真實環境，建立兩個不一樣的項目進行演示。建立兩個maven項目，消息生產者mq-rabbit-provider和消息消費者mq-rabbit-consumer，兩個項目的pom.xml文件添加相同依賴：

<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>

mq-rabbit-provider項目的application.properties內容以下：

server.port=8080
spring.application.name=springboot-rabbitmq-provider
spring.rabbitmq.host=10.211.55.3
spring.rabbitmq.port=5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
#RabbitMQ的虛擬host
spring.rabbitmq.virtual-host=CalonHost

mq-rabbit-consumer項目的application.properties內容以下：

server.port=9090
spring.application.name=springboot-rabbitmq-consumer
spring.rabbitmq.host=10.211.55.3
spring.rabbitmq.port=5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
#RabbitMQ的虛擬host
spring.rabbitmq.virtual-host=CalonHost

這裏只是端口和應用名不一樣，其餘都同樣。

接下來分別介紹Direct、Topic、Fanout等3種不一樣交換機的使用例子。

Direct Exchange

在mq-rabbit-provider項目建一個配置類DirectRabbitConfig.java，配置交換機、隊列、BindingKey=CalonDirectRouting的綁定關係，代碼以下：

@Configuration
public class DirectRabbitConfig {
//隊列
@Bean
public Queue CalonDirectQueue() {
return new Queue("CalonDirectQueue",true);
}
//Direct交換機
@Bean
DirectExchange CalonDirectExchange() {
return new DirectExchange("CalonDirectExchange");
}
//綁定
@Bean
Binding bindingDirect() {
return BindingBuilder.bind(CalonDirectQueue()).to(CalonDirectExchange()).with("CalonDirectRouting");
}
}

建立一個實體類User.java，這裏說明一下，該實體類是消息的主體，因此必須實現Serializable接口，不然在消息消費者項目讀取消息時會報錯，代碼以下：

package mq.rabbit.entity;
import java.io.Serializable;
public class User implements Serializable{
private static final long serialVersionUID = 1L;
private String id;
private String username;
private String password;
private String type;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public String getType() {
return type;
}
public void setType(String type) {
this.type = type;
}
public User() {
super();
}
public User(String id, String username, String password, String type) {
super();
this.id = id;
this.username = username;
this.password = password;
this.type = type;
}
}

下面建立一個Controller，利用http請求進行調試，CalonDirectExchange是上面配置的交換機標識，CalonDirectRouting就是上面綁定好的queue名字，因爲上面已經配置好交換機和隊列的綁定關係，這兩個組合就能夠知道消息最終是發送到隊列CalonDirectQueue裏面去了,Controller類的代碼以下：

@Controller
public class SendController {
@Autowired
private RabbitTemplate template;
@GetMapping("/sendDirect")
private @ResponseBody String sendDirect(String message) throws Exception {
User user = new User(UUID.randomUUID().toString(), message, "123456", "sendDirect");
template.convertAndSend("CalonDirectExchange", "CalonDirectRouting", user);
return "OK,sendDirect:" + message;
}
}

啓動mq-rabbit-provider項目，在瀏覽器輸入：

http://localhost:8080/sendDirect?message=123

再去RabbitMQ的web管理後臺查看，你會發如今Queue裏找到剛剛添加的那個隊列，後面的數字就是消息數量有變化，說明消息已經存儲進去了：

把mq-rabbit-provider項目裏的User類和DirectRabbitConfig類複製到mq-rabbit-consumer項目，User類用於讀取消息時接收消息對象，DirectRabbitConfig能夠不復制，可是若是RabbitMQ裏尚未被監聽的隊列時會報錯，複製過來是爲了讓RabbitMQ裏尚未被監聽的隊列時自動建立該隊列，防止報錯。

建立隊列監聽類DirectReceiver.java，代碼以下：

package mq.rabbit.receiver;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import mq.rabbit.entity.User;
@Component
@RabbitListener(queues = "CalonDirectQueue")//CalonDirectQueue爲隊列名稱
public class DirectReceiver {
@RabbitHandler
public void process(User user) {
System.out.println("DirectReceiver消費者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
}
}

啓動mq-rabbit-consumer項目，就會收到以前發送到CalonDirectQueue隊列的消息了，繼續調用上面的請求/sendDirect，消息消費者會繼續收到消息。

Topic Exchange

在mq-rabbit-provider項目建一個配置類TopicRabbitConfig.java，配置交換機、隊列、BindingKey的綁定關係，代碼以下：

package mq.rabbit.config;
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;
/**
* Topic Exchange類型交換機
* @author calon
*
*/
@Configuration
public class TopicRabbitConfig {
public final static String first = "topic.first";
public final static String second = "topic.second";
@Bean
public Queue firstQueue() {
return new Queue(TopicRabbitConfig.first);
}
@Bean
public Queue secondQueue() {
return new Queue(TopicRabbitConfig.second);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("topicExchange");
}
//綁定topic.first隊列到routingKey爲topic.first，只有topic.first的routingKey消息才發送到此隊列
@Bean
Binding bindingExchangeMessage() {
return BindingBuilder.bind(firstQueue()).to(exchange()).with(first);
}
//綁定topic.second隊列到topic.#，凡是topic.開頭的routingKey消息都發送到此隊列
@Bean
Binding bindingExchangeMessage2() {
return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
}
}

Topic Exchange類型的交換機是基於模糊匹配規則，因此這裏建立兩個Queue，分別綁定到兩個BindingKey：topic.first和topic.#，用來測試消息進到哪一個隊列裏。

在SendController類裏添加兩個request，代碼以下：

@Controller
public class SendController {
@Autowired
private RabbitTemplate template;
@GetMapping("/sendTopicFirst")
private @ResponseBody String sendTopicFirst(String message) {
User user = new User(UUID.randomUUID().toString(), message, "123456", "sendTopicFirst");
template.convertAndSend("topicExchange", "topic.first", user);
return "OK,sendTopicFirst:" + message;
}
@GetMapping("/sendTopicSecond")
private @ResponseBody String sendTopicSecond(String message) {
User user = new User(UUID.randomUUID().toString(), message, "123456", "sendTopicSecond");
template.convertAndSend("topicExchange", "topic.second", user);
return "OK,sendTopicSecond:" + message;
}
}

當咱們調用/sendTopicFirst請求時，交換機爲topicExchange，routingKey爲topic.first，按照上面bindingKey的配置，能夠匹配到topic.first和topic.#規則，對應的隊列是topic.first和topic.second,因此一條消息進到兩個隊列裏。

當調用/sendTopicSecond請求時，交換機爲topicExchange，routingKey爲topic.second，匹配到topic.#規則，對應的隊列是topic.second，因此消息進到topic.second隊列裏，除了#匹配規則，你們能夠自行試試星號(*)這個匹配規則，*符號是匹配一個單詞的。

把mq-rabbit-provider項目裏的TopicRabbitConfig類複製到mq-rabbit-consumer項目，分別建立TopicFirstReceiver和TopicSecondReceiver消息監聽類，代碼以下：

package mq.rabbit.receiver;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import mq.rabbit.entity.User;
@Component
@RabbitListener(queues = "topic.first")
public class TopicFirstReceiver {
@RabbitHandler
public void process(User user) {
System.out.println("TopicFirstReceiver消費者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
}
}

package mq.rabbit.receiver;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import mq.rabbit.entity.User;
@Component
@RabbitListener(queues = "topic.second")
public class TopicSecondReceiver {
@RabbitHandler
public void process(User user) {
System.out.println("TopicSecondReceiver消費者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
}
}

啓動mq-rabbit-consumer項目，會發現分別接收到各自監聽的隊列的消息。

Fanout Exchang

在mq-rabbit-provider項目建一個配置類FanoutRabbitConfig.java，配置交換機、隊列的綁定關係，代碼以下：

package mq.rabbit.config;
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;
@Configuration
public class FanoutRabbitConfig {
@Bean
public Queue AMessage() {
return new Queue("fanout.A");
}
@Bean
public Queue BMessage() {
return new Queue("fanout.B");
}
@Bean
public Queue CMessage() {
return new Queue("fanout.C");
}
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
@Bean
Binding bindingExchangeA() {
return BindingBuilder.bind(AMessage()).to(fanoutExchange());
}
@Bean
Binding bindingExchangeB() {
return BindingBuilder.bind(BMessage()).to(fanoutExchange());
}
@Bean
Binding bindingExchangeC() {
return BindingBuilder.bind(CMessage()).to(fanoutExchange());
}
}

這裏建立三個隊列fanout.A、fanout.B、fanout.C，都綁定到FanoutExchange交換機fanoutExchange上。

在SendController類添加一個請求/sendFanout,代碼以下：

package mq.rabbit.controller;
import java.util.UUID;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.ResponseBody;
import com.fasterxml.jackson.databind.ObjectMapper;
import mq.rabbit.entity.User;
@Controller
public class SendController {
@Autowired
private RabbitTemplate template;
@GetMapping("/sendFanout")
private @ResponseBody String sendFanout(String message) {
User user = new User(UUID.randomUUID().toString(), message, "123456", "sendFanout");
template.convertAndSend("fanoutExchange", null, user);
return "OK,sendFanout:" + message;
}
}

當調用/sendFanout請求時，在RabbitMQ的web管理界面看到三個隊列fanout.A、fanout.B、fanout.C都有一條消息，在Fanout交換機裏，若是有設置BindingKey，Fanout交換機會忽略已設置的BindingKey，把消息發送到綁定該交換機的全部隊列裏。

把mq-rabbit-provider項目裏的FanoutRabbitConfig類複製到mq-rabbit-consumer項目，分別建立FanoutReceiverA、FanoutReceiverB和FanoutReceiverC類，代碼以下：

package mq.rabbit.receiver;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import mq.rabbit.entity.User;
@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
@RabbitHandler
public void process(User user) {
System.out.println("FanoutReceiverA消費者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
}
}

package mq.rabbit.receiver;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import mq.rabbit.entity.User;
@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {
@RabbitHandler
public void process(User user) {
System.out.println("FanoutReceiverB消費者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
}
}

package mq.rabbit.receiver;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import mq.rabbit.entity.User;
@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
@RabbitHandler
public void process(User user) {
System.out.println("FanoutReceiverC消費者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
}
}

上面也能夠在一個類裏寫3個方法來進行對隊列的監聽，不一樣的地方在於把@RabbitListener移到方法上便可。

啓動mq-rabbit-consumer，便可收到隊列的消息。

RabbitMQ消息的確認機制

在使用RabbitMQ的時候，咱們能夠經過消息持久化操做來解決由於服務器的異常奔潰致使的消息丟失，除此以外咱們還會遇到一個問題，當消息的生產者在將消息發送出去以後，消息到底有沒有正確到達服務器？若是不進行特殊配置的話，默認狀況下發布消息是不會返回任何信息給生產者的，也就是生產者是不知道消息有沒有正確到達消息服務器，同理，消息消費者在接收消息後，若是在執行業務邏輯過程出現異常崩潰等狀況，會致使消息丟失，因此咱們須要對消息的發送和消費進行確認，確保消息可以被正確的存儲和消費。RabbitMQ爲咱們提供了兩種方式：一、事務機制；二、確認機制。下面介紹消息確認機制。

生產者消息確認機制：

先把例子跑起來，下面再作詳細介紹。在mq-rabbit-provider項目的application.properties文件添加如下屬性：

#確認消息已發送到交換機(Exchange)
spring.rabbitmq.publisher-confirms=true
#確認消息已發送到隊列(Queue)
spring.rabbitmq.publisher-returns=true

在mq-rabbit-provider項目建立配置類RabbitConfig.java，代碼以下：

package mq.rabbit.config;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
@Bean
public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
rabbitTemplate.setMandatory(true);//必須設置爲true，才能讓下面的ReturnCallback函數生效
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("=======ConfirmCallback=========");
System.out.println("correlationData = " + correlationData);
System.out.println("ack = " + ack);
System.out.println("cause = " + cause);
System.out.println("=======ConfirmCallback=========");
}
});
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
System.out.println("--------------ReturnCallback----------------");
System.out.println("message = " + message);
System.out.println("replyCode = " + replyCode);
System.out.println("replyText = " + replyText);
System.out.println("exchange = " + exchange);
System.out.println("routingKey = " + routingKey);
System.out.println("--------------ReturnCallback----------------");
}
});
return rabbitTemplate;
}
}

RabbitMQ生產者是依賴兩個回調函數來實現確認的，分別是ConfirmCallback和ConfirmCallback，如上面的代碼。按如下4種狀況進行回調：

一、消息找不到交換機(Exchange)時回調ConfirmCallback，返回ack=false，代碼以下：

=======ConfirmCallback=========
correlationData = null
ack = false
cause = channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'CalonDirectExchange1' in vhost 'calonHost', class-id=60, method-id=40)
=======ConfirmCallback=========
2018-08-30 09:59:37.892 ERROR 55704 --- [0.211.55.3: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 'CalonDirectExchange1' in vhost 'calonHost', class-id=60, method-id=40)

二、消息找到交換機(Exchange)但找不到隊列(Queue)時回調ConfirmCallback和ReturnCallback，返回ack=true，replyCode = 312，replyText = NO_ROUTE，代碼以下：

--------------ReturnCallback----------------
message = (Body:'[B@bf8af5b(byte[179])' MessageProperties [headers={}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/x-java-serialized-object, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=null, receivedExchange=null, receivedRoutingKey=null, receivedDelay=null, deliveryTag=0, messageCount=null, consumerTag=null, consumerQueue=null])
replyCode = 312
replyText = NO_ROUTE
exchange = CalonDirectExchange
routingKey = CalonDirectRouting1
--------------ReturnCallback----------------
=======ConfirmCallback=========
correlationData = null
ack = true
cause = null
=======ConfirmCallback=========

三、消息既找不到交換機(Exchange)又找不到隊列(Queue)時回調ConfirmCallback，返回ack=false，代碼以下：

=======ConfirmCallback=========
correlationData = null
ack = false
cause = channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'CalonDirectExchange1' in vhost 'calonHost', class-id=60, method-id=40)
=======ConfirmCallback=========
2018-08-30 10:03:22.204 ERROR 55704 --- [0.211.55.3: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 'CalonDirectExchange1' in vhost 'calonHost', class-id=60, method-id=40)

四、消息成功發送回調ConfirmCallback，返回ack=true，代碼以下：

=======ConfirmCallback=========
correlationData = null
ack = true
cause = null
=======ConfirmCallback=========

根據上面4種狀態，咱們能夠在這兩個回調函數里根據返回的狀態進行業務方面的處理，好比業務回滾或者從新發送消息等，能夠基於上面SendController類對其中一個請求進行測試，更改exchange和routingKey來測試一下這4種狀態，這個就是生產消息的確認機制。

消費者消息確認機制：

在mq-rabbit-consumer項目的DirectRabbitConfig配置類進行消息消費確認機制的配置，代碼以下：

package mq.rabbit.config;
import org.springframework.amqp.core.AcknowledgeMode;
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.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;
import mq.rabbit.receiver.DirectAckReceiver;
@Configuration
public class DirectRabbitConfig {
@Bean
public Queue CalonDirectQueue() {
return new Queue("CalonDirectQueue",true);
}
@Bean
DirectExchange CalonDirectExchange() {
return new DirectExchange("CalonDirectExchange");
}
@Bean
Binding bindingDirect() {
return BindingBuilder.bind(CalonDirectQueue()).to(CalonDirectExchange()).with("CalonDirectRouting");
}
@Autowired
private CachingConnectionFactory connectionFactory;
@Autowired
private DirectAckReceiver directAckReceiver;//消息接收處理類
@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setConcurrentConsumers(1);
container.setMaxConcurrentConsumers(1);
container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默認是自動確認，這裏改成手動確認消息
container.setQueues(CalonDirectQueue());
container.setMessageListener(directAckReceiver);
return container;
}
}

在mq-rabbit-consumer項目新建消息監聽類DirectAckReceiver.java，用於處理消息的確認操做，代碼以下：

package mq.rabbit.receiver;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import mq.rabbit.entity.User;
@Component
public class DirectAckReceiver implements ChannelAwareMessageListener {
@Autowired
private ObjectMapper objectMapper;
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
byte[] body = message.getBody();
User user = objectMapper.readValue(body, User.class);
System.out.println("DirectAckReceiver消費者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
channel.basicAck(deliveryTag, true);
// channel.basicReject(deliveryTag, true);//爲true會從新放回隊列
} catch (Exception e) {
channel.basicReject(deliveryTag, false);
e.printStackTrace();
}
}
}