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

時間  2019-11-16
標籤 消息 中間件 rabbitmq 整合 springboot 實戰 欄目 RabbitMQ 简体版
原文   原文鏈接

求關注
RabbitMQ整合SpringBoot實戰！（全）

前言

1. SpringBoot整合配置詳解

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

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

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

2. 代碼演示

2.1 生產端

2.1.1 新建項目springboot-producer

pom.xmlgithub

<?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.propertiesspring

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 數據庫

Exchange-1
添加成功

添加Queue apache

Queue-1

Exchange綁定Queue 編程

綁定
綁定成功

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

returnCallback打印日誌-錯誤

這時候咱們發現報錯了

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

複製代碼

returnCallback打印日誌-正確

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直接經過這個組合註解一次性搞定消費端交換機、隊列、綁定、路由、而且配置監聽功能等。

@RabbitListener註解

好比在方法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 消費端優化

一、先定義一個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;
	}
}


複製代碼

注意：咱們在傳輸對象的時候，必須序列化。不然會傳輸失敗。

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

三、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 生產端優化

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

二、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);
}


複製代碼

三、添加測試方法

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

複製代碼

3.3測試

運行testSender2()方法。

生產端打印消息

生產端打印消息

消費端打印消息

消費端打印消息

至此，RabbitMQ整合SpringBoot完畢，在實際工做中，使用場景也是差很少的。

文末

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

文章收錄至 Github: github.com/CoderMerlin… Gitee: gitee.com/573059382/c… 歡迎關注並star~

微信公衆號

參考文章：

《RabbitMQ消息中間件精講》

推薦文章：

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

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

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

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程