消息中間件——RabbitMQ（十）RabbitMQ整合SpringBoot實戰！（全）

前言

1. SpringBoot整合配置詳解

• publisher-confirms,實現一個監聽器用於監聽Broker端給我們返回的確認請求：RabbitTemplate.ConfirmCallback

• publisher-returns,保證消息對Broker端是可達的，如果出現路由鍵不可達的情況，則使用監聽器對不可達的消息進行后續的處理，保證消息的路由成功：RabbitTemplate.ReturnCallback

注意一點，在發送消息的時候對template進行配置mandatory=true保證監聽有效
生產端還可以配置其他屬性，比如發送重試，超時時間，次數，間隔等

2. 代碼演示

2.1 生產端

2.1.1 新建項目springboot-producer

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.cp</groupId>
	<artifactId>springboot-producer</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>springboot-producer</name>
	<description>springboot-producer</description>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.0.4.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>1.8</java.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>
</project>

RabbitSender.java 消息生產者

@Component
public class RabbitSender {

	//自動注入RabbitTemplate模板類
	@Autowired
	private RabbitTemplate rabbitTemplate;  
	
	//回調函數: confirm確認
	final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
		@Override
		public void confirm(CorrelationData correlationData, boolean ack, String cause) {
			System.err.println("correlationData: " + correlationData);
			System.err.println("ack: " + ack);
			if(!ack){
				//可以進行日志記錄、異常處理、補償處理等
				System.err.println("異常處理....");
			}else {
				//更新數據庫，可靠性投遞機制
			}
		}
	};
	
	//回調函數: return返回
	final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
		@Override
		public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
				String exchange, String routingKey) {
			System.err.println("return exchange: " + exchange + ", routingKey: " 
				+ routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
		}
	};
	
	//發送消息方法調用: 構建Message消息
	public void send(Object message, Map<String, Object> properties) throws Exception {
		MessageHeaders mhs = new MessageHeaders(properties);
		Message msg = MessageBuilder.createMessage(message, mhs);
		rabbitTemplate.setConfirmCallback(confirmCallback);
		rabbitTemplate.setReturnCallback(returnCallback);
		//id + 時間戳 全局唯一  用於ack保證唯一一條消息,這邊做測試寫死一個。但是在做補償策略的時候，必須保證這是全局唯一的消息
		CorrelationData correlationData = new CorrelationData("1234567890");
		rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);
	}
	
}

application.properties

spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/vhost_cp
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true

2.1.2 操作管控台

添加Exchange

添加Queue

Exchange綁定Queue

修改routingKey，springboot改為spring，則進入的是returnCallback方法

這時候我們發現報錯了

correlationData: CorrelationData [id=1234567890]
ack: false
異常處理....

2.1.3 解決ack為false問題

這是由於我們在測試方法中進行測試，當測試方法結束，rabbitmq相關的資源也就關閉了，雖然我們的消息發送出去，但異步的ConfirmCallback卻由於資源關閉而出現了上面的問題。
加入Thread.sleep()即可解決。

@Test
public void testSender1() throws Exception {
	 Map<String, Object> properties = new HashMap<>();
	 properties.put("number", "12345");
	 properties.put("send_time", simpleDateFormat.format(new Date()));
	 rabbitSender.send("Hello RabbitMQ For Spring Boot!", properties);
	 Thread.sleep(2000);
}

成功解決~

2.2 消費端

消費端核心配置：

簽收模式-手工簽收

spring.rabbitmq.listener.simple.acknowledge-mode=manual

設置監聽限制：最大10，默認5

spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10

• 首先配置手工確認模式，用於ACK的手工處理，這樣我們可以保證消息的可靠性送達，或者再消費端消費失敗的時候可以做到重回隊列（不建議）、根據業務記錄日志等處理。

• 可以設置消費端的監聽個數和最大個數，用於監控消費端的並發情況

@RabbitListener注解使用

• 消費端監聽@RabbitListener注解,這個對於在實際工作中非常的好用
• @RabbitListener是一個組合注解，里面可以注解配置
• @QueueBinding、@Queue、@Exchange直接通過這個組合注解一次性搞定消費端交換機、隊列、綁定、路由、並且配置監聽功能等。

比如在方法onMessage上加@RabbitListener注解，同時需要加另外一個注解@RabbitHandler,代碼被消費者監聽。

建立綁定，在Value上寫上隊列，設置Exchange，是否持久化，設置Exchange的類型、表達式設置為true以及路由key。通過這種簡單的方式，就可以完成之前很復雜的代碼邏輯。同時建議將配置放入到配置文件中，動態獲取。如果mq中沒有相應的隊列、Exchange等，注解聲明也可以創建它們，大家可以自行測試！

2.2.1 新建項目springboot-consumer

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.cp</groupId>
	<artifactId>springboot-consumer</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>springboot-consumer</name>
	<description>springboot-consumer</description>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.0.2.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>1.8</java.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>


</project>

RabbitReceiver.java 消息生產者

@Component
public class RabbitReceiver {

	
	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "queue-1", 
			durable="true"),
			exchange = @Exchange(value = "exchange-1", 
			durable="true", 
			type= "topic", 
			ignoreDeclarationExceptions = "true"),
			key = "springboot.*"
			)
	)
	@RabbitHandler
	public void onMessage(Message message, Channel channel) throws Exception {
		System.err.println("--------------------------------------");
		System.err.println("消費端Payload: " + message.getPayload());
		Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
		//手工ACK,獲取deliveryTag
		channel.basicAck(deliveryTag, false);
	}
}

application.properties

spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=user_cp
spring.rabbitmq.password=123456
spring.rabbitmq.virtual-host=/vhost_cp
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10

運行Application，查看之前在生產端發送的消息，是否能被消費。

打印結果

這里之前由於我測試的時候多發了消息，所以消費的時候會有這么多。

3. 優化代碼

• 自定義Java對象消息
• @RabbitListener注解中的配置改為動態配置

@Payload：指定具體的消息體Body。
@Headers: 獲取Headers。

3.1 消費端優化

1、先定義一個Order對象

public class Order implements Serializable {

	private String id;
	private String name;
	
	public Order() {
	}
	public Order(String id, String name) {
		super();
		this.id = id;
		this.name = name;
	}
	public String getId() {
		return id;
	}
	public void setId(String id) {
		this.id = id;
	}
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
}

注意：我們在傳輸對象的時候，必須序列化。否則會傳輸失敗。

2、RabbitReceiver添加監聽

/**
	 * 
	 * 	spring.rabbitmq.listener.order.queue.name=queue-2
		spring.rabbitmq.listener.order.queue.durable=true
		spring.rabbitmq.listener.order.exchange.name=exchange-2
		spring.rabbitmq.listener.order.exchange.durable=true
		spring.rabbitmq.listener.order.exchange.type=topic
		spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
		spring.rabbitmq.listener.order.key=springboot.*
	 * @param order
	 * @param channel
	 * @param headers
	 * @throws Exception
	 */
	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}", 
			durable="${spring.rabbitmq.listener.order.queue.durable}"),
			exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}", 
			durable="${spring.rabbitmq.listener.order.exchange.durable}", 
			type= "${spring.rabbitmq.listener.order.exchange.type}", 
			ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),
			key = "${spring.rabbitmq.listener.order.key}"
			)
	)
	@RabbitHandler
	public void onOrderMessage(@Payload com.cp.springboot.entity.Order order, 
			Channel channel, 
			@Headers Map<String, Object> headers) throws Exception {
		System.err.println("--------------------------------------");
		System.err.println("消費端order: " + order.getId());
		Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
		//手工ACK
		channel.basicAck(deliveryTag, false);
	}

已經將配置寫入到了application.properties中，進行動態獲取。也可以像我們公司一樣放入到配置中心當中。例如：攜程開源配置中心Apollo

3、application.properties

spring.rabbitmq.listener.order.queue.name=queue-2
spring.rabbitmq.listener.order.queue.durable=true
spring.rabbitmq.listener.order.exchange.name=exchange-2
spring.rabbitmq.listener.order.exchange.durable=true
spring.rabbitmq.listener.order.exchange.type=topic
spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
spring.rabbitmq.listener.order.key=springboot.*

3.2 生產端優化

1、同樣是一個Order對象，必須跟消費端的保持一致。

2、RabbitSender添加發送消息

//發送消息方法調用: 構建自定義對象消息
public void sendOrder(Order order) throws Exception {
	rabbitTemplate.setConfirmCallback(confirmCallback);
	rabbitTemplate.setReturnCallback(returnCallback);
	//id + 時間戳 全局唯一 
	CorrelationData correlationData = new CorrelationData("0987654321");
	rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);
}

3、添加測試方法

@Test
public void testSender2() throws Exception {
	 Order order = new Order("001", "第一個訂單");
	 rabbitSender.sendOrder(order);
	 //防止資源提前關閉，ConfirmCallback異步回調失敗
	 Thread.sleep(2000);
}

4.測試

運行testSender2()方法。

生產端打印消息

消費端打印消息

至此，RabbitMQ整合SpringBoot完畢，在實際工作中，使用場景也是差不多的。

文末

歡迎關注個人微信公眾號：Coder編程
獲取最新原創技術文章和免費學習資料，更有大量精品思維導圖、面試資料、PMP備考資料等你來領，方便你隨時隨地學習技術知識！
新建了一個qq群：315211365，歡迎大家進群交流一起學習。謝謝了！也可以介紹給身邊有需要的朋友。

文章收錄至
Github: https://github.com/CoderMerlin/coder-programming
Gitee: https://gitee.com/573059382/coder-programming
歡迎關注並star~

參考文章：

《RabbitMQ消息中間件精講》

推薦文章：

消息中間件——RabbitMQ（七）高級特性全在這里!（上）

消息中間件——RabbitMQ（八）高級特性全在這里!（下）

消息中間件——RabbitMQ（九）RabbitMQ整合Spring AMQP實戰！（全）