Spring?Boot?消息隊(duì)列與異步處理的應(yīng)用小結(jié)
Spring Boot 消息隊(duì)列與異步處理
36.1 學(xué)習(xí)目標(biāo)與重點(diǎn)提示
學(xué)習(xí)目標(biāo):掌握Spring Boot消息隊(duì)列與異步處理的核心概念與使用方法,包括消息隊(duì)列的定義與特點(diǎn)、異步處理的定義與特點(diǎn)、Spring Boot與消息隊(duì)列的集成、Spring Boot與異步處理的集成、Spring Boot的實(shí)際應(yīng)用場(chǎng)景,學(xué)會(huì)在實(shí)際開(kāi)發(fā)中處理消息隊(duì)列與異步處理問(wèn)題。
重點(diǎn):消息隊(duì)列的定義與特點(diǎn)、異步處理的定義與特點(diǎn)、Spring Boot與消息隊(duì)列的集成、Spring Boot與異步處理的集成、Spring Boot的實(shí)際應(yīng)用場(chǎng)景。
36.2 消息隊(duì)列與異步處理概述
消息隊(duì)列與異步處理是Java開(kāi)發(fā)中的重要組件,用于實(shí)現(xiàn)系統(tǒng)的異步處理和消息傳遞。
36.2.1 消息隊(duì)列的定義
定義:消息隊(duì)列是一種用于存儲(chǔ)和傳遞消息的中間件,支持異步通信和消息處理。
作用:
- 實(shí)現(xiàn)系統(tǒng)的異步處理。
- 提高系統(tǒng)的響應(yīng)速度。
- 實(shí)現(xiàn)系統(tǒng)的解耦。
常見(jiàn)的消息隊(duì)列:
- RabbitMQ:開(kāi)源的消息隊(duì)列。
- ActiveMQ:開(kāi)源的消息隊(duì)列。
- Kafka:分布式流處理平臺(tái)。
? 結(jié)論:消息隊(duì)列是一種用于存儲(chǔ)和傳遞消息的中間件,作用是實(shí)現(xiàn)系統(tǒng)的異步處理、提高系統(tǒng)的響應(yīng)速度、實(shí)現(xiàn)系統(tǒng)的解耦。
36.2.2 異步處理的定義
定義:異步處理是指系統(tǒng)在處理請(qǐng)求時(shí),不需要等待請(qǐng)求完成就可以繼續(xù)處理其他請(qǐng)求。
作用:
- 提高系統(tǒng)的響應(yīng)速度。
- 提高系統(tǒng)的吞吐量。
- 實(shí)現(xiàn)系統(tǒng)的解耦。
常見(jiàn)的異步處理:
- 異步調(diào)用。
- 異步任務(wù)。
- 異步消息。
? 結(jié)論:異步處理是指系統(tǒng)在處理請(qǐng)求時(shí),不需要等待請(qǐng)求完成就可以繼續(xù)處理其他請(qǐng)求,作用是提高系統(tǒng)的響應(yīng)速度、提高系統(tǒng)的吞吐量、實(shí)現(xiàn)系統(tǒng)的解耦。
36.3 Spring Boot與消息隊(duì)列的集成
Spring Boot與消息隊(duì)列的集成是Java開(kāi)發(fā)中的重要內(nèi)容。
36.3.1 集成RabbitMQ的步驟
定義:集成RabbitMQ的步驟是指使用Spring Boot與RabbitMQ集成的方法。
步驟:
- 創(chuàng)建Spring Boot項(xiàng)目。
- 添加所需的依賴。
- 配置RabbitMQ。
- 創(chuàng)建消息發(fā)送類(lèi)。
- 創(chuàng)建消息接收類(lèi)。
- 創(chuàng)建控制器類(lèi)。
- 測(cè)試應(yīng)用。
示例:
pom.xml文件中的依賴:
<dependencies>
<!-- Web依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- 消息隊(duì)列依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 數(shù)據(jù)驗(yàn)證依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<!-- 測(cè)試依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>application.properties文件中的配置:
# 服務(wù)器端口 server.port=8080 # RabbitMQ配置 spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.rabbitmq.virtual-host=/
消息發(fā)送類(lèi):
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class ProductMessageSender {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendProductMessage(String message) {
rabbitTemplate.convertAndSend("product-exchange", "product-routing-key", message);
System.out.println("發(fā)送產(chǎn)品消息:" + message);
}
}消息接收類(lèi):
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class ProductMessageReceiver {
@RabbitListener(queues = "product-queue")
public void receiveProductMessage(String message) {
System.out.println("接收產(chǎn)品消息:" + message);
// 處理消息
processProductMessage(message);
}
private void processProductMessage(String message) {
// 模擬處理消息的耗時(shí)操作
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("處理產(chǎn)品消息完成:" + message);
}
}配置類(lèi):
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
@Bean
public Queue productQueue() {
return new Queue("product-queue", true);
}
@Bean
public DirectExchange productExchange() {
return new DirectExchange("product-exchange");
}
@Bean
public Binding productBinding(Queue productQueue, DirectExchange productExchange) {
return BindingBuilder.bind(productQueue).to(productExchange).with("product-routing-key");
}
}控制器類(lèi):
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.BindingResult;
import org.springframework.validation.FieldError;
import org.springframework.web.bind.annotation.*;
import javax.validation.Valid;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@RestController
@RequestMapping("/api/products")
public class ProductController {
@Autowired
private ProductMessageSender productMessageSender;
@PostMapping("/send")
public Map<String, Object> sendProductMessage(@RequestParam String message) {
Map<String, Object> result = new HashMap<>();
productMessageSender.sendProductMessage(message);
result.put("success", true);
result.put("message", "產(chǎn)品消息發(fā)送成功");
return result;
}
}應(yīng)用啟動(dòng)類(lèi):
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class RabbitMQApplication {
public static void main(String[] args) {
SpringApplication.run(RabbitMQApplication.class, args);
}
}測(cè)試類(lèi):
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.boot.web.server.LocalServerPort;
import org.springframework.http.*;
import java.util.HashMap;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class RabbitMQApplicationTests {
@LocalServerPort
private int port;
@Autowired
private TestRestTemplate restTemplate;
@Test
void contextLoads() {
}
@Test
void testSendProductMessage() {
ResponseEntity<Map> response = restTemplate.getForEntity("http://localhost:" + port + "/api/products/send?message=測(cè)試產(chǎn)品消息", Map.class);
assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK);
assertThat(response.getBody().get("success")).isEqualTo(true);
}
}? 結(jié)論:集成RabbitMQ的步驟包括創(chuàng)建Spring Boot項(xiàng)目、添加所需的依賴、配置RabbitMQ、創(chuàng)建消息發(fā)送類(lèi)、創(chuàng)建消息接收類(lèi)、創(chuàng)建控制器類(lèi)、測(cè)試應(yīng)用。
36.4 Spring Boot與異步處理的集成
Spring Boot與異步處理的集成是Java開(kāi)發(fā)中的重要內(nèi)容。
36.4.1 集成Spring Boot異步處理的步驟
定義:集成Spring Boot異步處理的步驟是指使用Spring Boot與異步處理集成的方法。
步驟:
- 創(chuàng)建Spring Boot項(xiàng)目。
- 添加所需的依賴。
- 配置異步處理。
- 創(chuàng)建異步任務(wù)類(lèi)。
- 創(chuàng)建控制器類(lèi)。
- 測(cè)試應(yīng)用。
示例:
pom.xml文件中的依賴:
<dependencies>
<!-- Web依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- 異步處理依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-task</artifactId>
</dependency>
<!-- 數(shù)據(jù)驗(yàn)證依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<!-- 測(cè)試依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>application.properties文件中的配置:
# 服務(wù)器端口 server.port=8080 # 異步處理配置 spring.task.execution.pool.core-size=5 spring.task.execution.pool.max-size=10 spring.task.execution.pool.queue-capacity=100
異步任務(wù)類(lèi):
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
@Component
public class ProductAsyncTask {
@Async
public void processProduct(String productId) {
System.out.println("開(kāi)始處理產(chǎn)品:" + productId);
// 模擬處理產(chǎn)品的耗時(shí)操作
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("產(chǎn)品處理完成:" + productId);
}
}控制器類(lèi):
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.BindingResult;
import org.springframework.validation.FieldError;
import org.springframework.web.bind.annotation.*;
import javax.validation.Valid;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@RestController
@RequestMapping("/api/products")
public class ProductController {
@Autowired
private ProductAsyncTask productAsyncTask;
@PostMapping("/process")
public Map<String, Object> processProduct(@RequestParam String productId) {
Map<String, Object> result = new HashMap<>();
productAsyncTask.processProduct(productId);
result.put("success", true);
result.put("message", "產(chǎn)品處理任務(wù)已提交");
return result;
}
}應(yīng)用啟動(dòng)類(lèi):
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
@SpringBootApplication
@EnableAsync
public class AsyncApplication {
public static void main(String[] args) {
SpringApplication.run(AsyncApplication.class, args);
}
}測(cè)試類(lèi):
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.boot.web.server.LocalServerPort;
import org.springframework.http.*;
import java.util.HashMap;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class AsyncApplicationTests {
@LocalServerPort
private int port;
@Autowired
private TestRestTemplate restTemplate;
@Test
void contextLoads() {
}
@Test
void testProcessProduct() {
ResponseEntity<Map> response = restTemplate.getForEntity("http://localhost:" + port + "/api/products/process?productId=P001", Map.class);
assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK);
assertThat(response.getBody().get("success")).isEqualTo(true);
}
}? 結(jié)論:集成Spring Boot異步處理的步驟包括創(chuàng)建Spring Boot項(xiàng)目、添加所需的依賴、配置異步處理、創(chuàng)建異步任務(wù)類(lèi)、創(chuàng)建控制器類(lèi)、測(cè)試應(yīng)用。
36.5 Spring Boot的實(shí)際應(yīng)用場(chǎng)景
在實(shí)際開(kāi)發(fā)中,Spring Boot消息隊(duì)列與異步處理的應(yīng)用場(chǎng)景非常廣泛,如:
- 實(shí)現(xiàn)訂單創(chuàng)建的異步處理。
- 實(shí)現(xiàn)郵件發(fā)送的異步處理。
- 實(shí)現(xiàn)數(shù)據(jù)同步的異步處理。
- 實(shí)現(xiàn)消息通知的異步處理。
示例:
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
class ProductMessageSender {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendProductMessage(String message) {
rabbitTemplate.convertAndSend("product-exchange", "product-routing-key", message);
System.out.println("發(fā)送產(chǎn)品消息:" + message);
}
}
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
class ProductMessageReceiver {
@RabbitListener(queues = "product-queue")
public void receiveProductMessage(String message) {
System.out.println("接收產(chǎn)品消息:" + message);
processProductMessage(message);
}
private void processProductMessage(String message) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("處理產(chǎn)品消息完成:" + message);
}
}
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
class RabbitMQConfig {
@Bean
public Queue productQueue() {
return new Queue("product-queue", true);
}
@Bean
public DirectExchange productExchange() {
return new DirectExchange("product-exchange");
}
@Bean
public Binding productBinding(Queue productQueue, DirectExchange productExchange) {
return BindingBuilder.bind(productQueue).to(productExchange).with("product-routing-key");
}
}
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
@Component
class ProductAsyncTask {
@Async
public void processProduct(String productId) {
System.out.println("開(kāi)始處理產(chǎn)品:" + productId);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("產(chǎn)品處理完成:" + productId);
}
}
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.BindingResult;
import org.springframework.validation.FieldError;
import org.springframework.web.bind.annotation.*;
import javax.validation.Valid;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@RestController
@RequestMapping("/api/products")
class ProductController {
@Autowired
private ProductMessageSender productMessageSender;
@Autowired
private ProductAsyncTask productAsyncTask;
@PostMapping("/send")
public Map<String, Object> sendProductMessage(@RequestParam String message) {
Map<String, Object> result = new HashMap<>();
productMessageSender.sendProductMessage(message);
result.put("success", true);
result.put("message", "產(chǎn)品消息發(fā)送成功");
return result;
}
@PostMapping("/process")
public Map<String, Object> processProduct(@RequestParam String productId) {
Map<String, Object> result = new HashMap<>();
productAsyncTask.processProduct(productId);
result.put("success", true);
result.put("message", "產(chǎn)品處理任務(wù)已提交");
return result;
}
}
@SpringBootApplication
@EnableAsync
public class AsyncAndRabbitMQApplication {
public static void main(String[] args) {
SpringApplication.run(AsyncAndRabbitMQApplication.class, args);
}
}
// 測(cè)試類(lèi)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class AsyncAndRabbitMQApplicationTests {
@LocalServerPort
private int port;
@Autowired
private TestRestTemplate restTemplate;
@Test
void contextLoads() {
}
@Test
void testSendProductMessage() {
ResponseEntity<Map> response = restTemplate.getForEntity("http://localhost:" + port + "/api/products/send?message=測(cè)試產(chǎn)品消息", Map.class);
assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK);
assertThat(response.getBody().get("success")).isEqualTo(true);
}
@Test
void testProcessProduct() {
ResponseEntity<Map> response = restTemplate.getForEntity("http://localhost:" + port + "/api/products/process?productId=P001", Map.class);
assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK);
assertThat(response.getBody().get("success")).isEqualTo(true);
}
}輸出結(jié)果:
- 發(fā)送產(chǎn)品消息:測(cè)試產(chǎn)品消息。
- 接收產(chǎn)品消息:測(cè)試產(chǎn)品消息。
- 處理產(chǎn)品消息完成:測(cè)試產(chǎn)品消息。
- 開(kāi)始處理產(chǎn)品:P001。
- 產(chǎn)品處理完成:P001。
? 結(jié)論:在實(shí)際開(kāi)發(fā)中,Spring Boot消息隊(duì)列與異步處理的應(yīng)用場(chǎng)景非常廣泛,需要根據(jù)實(shí)際問(wèn)題選擇合適的消息隊(duì)列和異步處理方法。
總結(jié)
本章我們學(xué)習(xí)了Spring Boot消息隊(duì)列與異步處理,包括消息隊(duì)列的定義與特點(diǎn)、異步處理的定義與特點(diǎn)、Spring Boot與消息隊(duì)列的集成、Spring Boot與異步處理的集成、Spring Boot的實(shí)際應(yīng)用場(chǎng)景,學(xué)會(huì)了在實(shí)際開(kāi)發(fā)中處理消息隊(duì)列與異步處理問(wèn)題。其中,消息隊(duì)列的定義與特點(diǎn)、異步處理的定義與特點(diǎn)、Spring Boot與消息隊(duì)列的集成、Spring Boot與異步處理的集成、Spring Boot的實(shí)際應(yīng)用場(chǎng)景是本章的重點(diǎn)內(nèi)容。從下一章開(kāi)始,我們將學(xué)習(xí)Spring Boot的其他組件、微服務(wù)等內(nèi)容。
到此這篇關(guān)于Spring Boot 消息隊(duì)列與異步處理的文章就介紹到這了,更多相關(guān)Spring Boot 消息隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- SpringBoot實(shí)現(xiàn)消息隊(duì)列與異步通信
- SpringBoot 集成消息隊(duì)列實(shí)戰(zhàn)指南(RabbitMQ/Kafka):異步通信與解耦,落地高可靠消息傳遞
- SpringBoot使用Redis Stream實(shí)現(xiàn)輕量消息隊(duì)列的示例代碼
- Springboot RabbitMQ 消息隊(duì)列使用示例詳解
- SpringAMQP消息隊(duì)列(SpringBoot集成RabbitMQ方式)
- SpringBoot使用Redis實(shí)現(xiàn)消息隊(duì)列的方法小結(jié)
- springboot整合消息隊(duì)列RabbitMQ
- springboot整合redis之消息隊(duì)列
- SpringBoot集成Redis實(shí)現(xiàn)消息隊(duì)列的方法
相關(guān)文章
MybatisPlus使用queryWrapper如何實(shí)現(xiàn)復(fù)雜查詢
這篇文章主要介紹了MybatisPlus使用queryWrapper如何實(shí)現(xiàn)復(fù)雜查詢,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教。2022-01-01
Java Graphics實(shí)現(xiàn)界面顯示文字并換行
Java中Graphics類(lèi)提供了一些基本的幾何圖形繪制方法,本文將利用Graphics實(shí)現(xiàn)界面顯示文字并換行效果,感興趣的小伙伴可以動(dòng)手嘗試一下2022-08-08
Springboot+Redis實(shí)現(xiàn)API接口防刷限流的項(xiàng)目實(shí)踐
本文主要介紹了Springboot+Redis實(shí)現(xiàn)API接口防刷限流的項(xiàng)目實(shí)踐,通過(guò)限流可以讓系統(tǒng)維持在一個(gè)相對(duì)穩(wěn)定的狀態(tài),為更多的客戶提供服務(wù),具有一定的參考價(jià)值,感興趣的可以了解一下2024-07-07
自定義注解實(shí)現(xiàn)Spring容器注入Bean方式(類(lèi)似于mybatis的@MapperScans)
本文介紹了如何通過(guò)自定義注解@MyService和@MyServiceScans在SpringBoot項(xiàng)目中自動(dòng)將指定包下的類(lèi)注入Spring容器,詳細(xì)解釋了創(chuàng)建自定義注解、定義包掃描器ClassPathBeanDefinitionScanner的作用與實(shí)現(xiàn)2024-09-09

