Spring Boot 消息队列与异步通信

Ne0inhk

— 12 min read
Spring Boot 消息队列与异步通信

Spring Boot 消息队列与异步通信

在这里插入图片描述
21.1 学习目标与重点提示

学习目标：掌握Spring Boot消息队列与异步通信的核心概念与使用方法，包括消息队列的定义与特点、Spring Boot与ActiveMQ的集成、Spring Boot与RabbitMQ的集成、Spring Boot与Kafka的集成、Spring Boot异步通信的基本方法、Spring Boot的实际应用场景，学会在实际开发中处理消息队列与异步通信问题。
重点：消息队列的定义与特点、Spring Boot与ActiveMQ的集成、Spring Boot与RabbitMQ的集成、Spring Boot与Kafka的集成、Spring Boot异步通信的基本方法、Spring Boot的实际应用场景。

21.2 消息队列概述

消息队列是Java开发中的重要组件。

21.2.1 消息队列的定义

定义：消息队列是一种异步通信机制，用于在应用程序之间传递消息。
作用：

  • 实现应用程序之间的异步通信。
  • 实现应用程序之间的解耦。
  • 提高应用程序的性能。

常见的消息队列：

  • ActiveMQ：Apache ActiveMQ是一款开源的消息队列。
  • RabbitMQ：RabbitMQ是一款开源的消息队列。
  • Kafka：Apache Kafka是一款开源的消息队列。

✅ 结论：消息队列是一种异步通信机制，作用是实现应用程序之间的异步通信、解耦、提高应用程序的性能。

21.2.2 消息队列的特点

定义：消息队列的特点是指消息队列的特性。
特点：

  • 异步通信：消息发送者不需要等待消息接收者的响应。
  • 解耦：消息发送者与消息接收者之间不需要直接通信。
  • 可靠性：消息队列提供消息的可靠传输。
  • 可扩展性：消息队列可以扩展到多个应用程序之间的通信。

✅ 结论：消息队列的特点包括异步通信、解耦、可靠性、可扩展性。

21.3 Spring Boot与ActiveMQ的集成

Spring Boot与ActiveMQ的集成是Java开发中的重要内容。

21.3.1 集成ActiveMQ的步骤

定义：集成ActiveMQ的步骤是指使用Spring Boot与ActiveMQ集成的方法。
步骤：

  1. 创建Spring Boot项目。
  2. 添加所需的依赖。
  3. 配置ActiveMQ。
  4. 创建消息生产者。
  5. 创建消息消费者。
  6. 测试应用。

示例：
pom.xml文件中的依赖：

<dependencies><!-- Web依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- ActiveMQ依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-activemq</artifactId></dependency><!-- 测试依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

application.properties文件中的ActiveMQ配置：

# 服务器端口 server.port=8080 # ActiveMQ配置 spring.activemq.broker-url=tcp://localhost:61616 spring.activemq.user=admin spring.activemq.password=admin

消息生产者：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.jms.core.JmsTemplate;importorg.springframework.stereotype.Component;@ComponentpublicclassMessageProducer{@AutowiredprivateJmsTemplate jmsTemplate;publicvoidsendMessage(String destination,String message){ jmsTemplate.convertAndSend(destination, message);System.out.println("发送消息："+ message);}}

消息消费者：

importorg.springframework.jms.annotation.JmsListener;importorg.springframework.stereotype.Component;@ComponentpublicclassMessageConsumer{@JmsListener(destination ="test-queue")publicvoidreceiveMessage(String message){System.out.println("接收消息："+ message);}}

控制器类：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RequestParam;importorg.springframework.web.bind.annotation.RestController;@RestControllerpublicclassMessageController{@AutowiredprivateMessageProducer messageProducer;@GetMapping("/send")publicStringsendMessage(@RequestParamString message){ messageProducer.sendMessage("test-queue", message);return"消息发送成功";}}

测试类：

importorg.junit.jupiter.api.Test;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.boot.test.web.client.TestRestTemplate;importorg.springframework.boot.web.server.LocalServerPort;importstaticorg.assertj.core.api.Assertions.assertThat;@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classActiveMQApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestSendMessage(){String message ="Hello, ActiveMQ!";String response = restTemplate.getForObject("http://localhost:"+ port +"/send?message="+ message,String.class);assertThat(response).contains("消息发送成功");}}

✅ 结论：集成ActiveMQ的步骤包括创建Spring Boot项目、添加所需的依赖、配置ActiveMQ、创建消息生产者、创建消息消费者、测试应用。

21.4 Spring Boot与RabbitMQ的集成

Spring Boot与RabbitMQ的集成是Java开发中的重要内容。

21.4.1 集成RabbitMQ的步骤

定义：集成RabbitMQ的步骤是指使用Spring Boot与RabbitMQ集成的方法。
步骤：

  1. 创建Spring Boot项目。
  2. 添加所需的依赖。
  3. 配置RabbitMQ。
  4. 创建消息生产者。
  5. 创建消息消费者。
  6. 测试应用。

示例：
pom.xml文件中的依赖：

<dependencies><!-- Web依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- RabbitMQ依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><!-- 测试依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

application.properties文件中的RabbitMQ配置：

# 服务器端口 server.port=8080 # RabbitMQ配置 spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest

消息生产者：

importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;@ComponentpublicclassMessageProducer{@AutowiredprivateRabbitTemplate rabbitTemplate;publicvoidsendMessage(String exchange,String routingKey,String message){ rabbitTemplate.convertAndSend(exchange, routingKey, message);System.out.println("发送消息："+ message);}}

消息消费者：

importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassMessageConsumer{@RabbitListener(queues ="test-queue")publicvoidreceiveMessage(String message){System.out.println("接收消息："+ message);}}

RabbitMQ配置类：

importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.DirectExchange;importorg.springframework.amqp.core.Queue;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassRabbitMQConfig{@BeanpublicQueuetestQueue(){returnnewQueue("test-queue",true);}@BeanpublicDirectExchangetestExchange(){returnnewDirectExchange("test-exchange");}@BeanpublicBindingtestBinding(){returnBindingBuilder.bind(testQueue()).to(testExchange()).with("test-routing-key");}}

控制器类：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RequestParam;importorg.springframework.web.bind.annotation.RestController;@RestControllerpublicclassMessageController{@AutowiredprivateMessageProducer messageProducer;@GetMapping("/send")publicStringsendMessage(@RequestParamString message){ messageProducer.sendMessage("test-exchange","test-routing-key", message);return"消息发送成功";}}

测试类：

importorg.junit.jupiter.api.Test;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.boot.test.web.client.TestRestTemplate;importorg.springframework.boot.web.server.LocalServerPort;importstaticorg.assertj.core.api.Assertions.assertThat;@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classRabbitMQApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestSendMessage(){String message ="Hello, RabbitMQ!";String response = restTemplate.getForObject("http://localhost:"+ port +"/send?message="+ message,String.class);assertThat(response).contains("消息发送成功");}}

✅ 结论：集成RabbitMQ的步骤包括创建Spring Boot项目、添加所需的依赖、配置RabbitMQ、创建消息生产者、创建消息消费者、测试应用。

21.5 Spring Boot与Kafka的集成

Spring Boot与Kafka的集成是Java开发中的重要内容。

21.5.1 集成Kafka的步骤

定义：集成Kafka的步骤是指使用Spring Boot与Kafka集成的方法。
步骤：

  1. 创建Spring Boot项目。
  2. 添加所需的依赖。
  3. 配置Kafka。
  4. 创建消息生产者。
  5. 创建消息消费者。
  6. 测试应用。

示例：
pom.xml文件中的依赖：

<dependencies><!-- Web依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Kafka依赖 --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><!-- 测试依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

application.properties文件中的Kafka配置：

# 服务器端口 server.port=8080 # Kafka配置 spring.kafka.bootstrap-servers=localhost:9092 spring.kafka.consumer.group-id=test-group spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

消息生产者：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.stereotype.Component;@ComponentpublicclassMessageProducer{@AutowiredprivateKafkaTemplate<String,String> kafkaTemplate;publicvoidsendMessage(String topic,String message){ kafkaTemplate.send(topic, message);System.out.println("发送消息："+ message);}}

消息消费者：

importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;@ComponentpublicclassMessageConsumer{@KafkaListener(topics ="test-topic", groupId ="test-group")publicvoidreceiveMessage(String message){System.out.println("接收消息："+ message);}}

控制器类：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RequestParam;importorg.springframework.web.bind.annotation.RestController;@RestControllerpublicclassMessageController{@AutowiredprivateMessageProducer messageProducer;@GetMapping("/send")publicStringsendMessage(@RequestParamString message){ messageProducer.sendMessage("test-topic", message);return"消息发送成功";}}

测试类：

importorg.junit.jupiter.api.Test;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.boot.test.web.client.TestRestTemplate;importorg.springframework.boot.web.server.LocalServerPort;importstaticorg.assertj.core.api.Assertions.assertThat;@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classKafkaApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestSendMessage(){String message ="Hello, Kafka!";String response = restTemplate.getForObject("http://localhost:"+ port +"/send?message="+ message,String.class);assertThat(response).contains("消息发送成功");}}

✅ 结论：集成Kafka的步骤包括创建Spring Boot项目、添加所需的依赖、配置Kafka、创建消息生产者、创建消息消费者、测试应用。

21.6 Spring Boot异步通信的基本方法

Spring Boot异步通信的基本方法包括使用@Async注解、使用CompletableFuture、使用消息队列。

21.6.1 使用@Async注解

定义：使用@Async注解是指使用Spring Boot异步通信的基本方法之一。
作用：

  • 实现异步通信。
  • 提高应用程序的性能。

示例：
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-test</artifactId><scope>test</scope></dependency></dependencies>

异步配置类：

importorg.springframework.context.annotation.Configuration;importorg.springframework.scheduling.annotation.EnableAsync;@Configuration@EnableAsyncpublicclassAsyncConfig{}

异步服务类：

importorg.springframework.scheduling.annotation.Async;importorg.springframework.stereotype.Service;@ServicepublicclassAsyncService{@AsyncpublicvoidasyncMethod(){System.out.println("异步方法执行："+Thread.currentThread().getName());}}

控制器类：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;@RestControllerpublicclassAsyncController{@AutowiredprivateAsyncService asyncService;@GetMapping("/async")publicStringasyncMethod(){System.out.println("主线程执行："+Thread.currentThread().getName()); asyncService.asyncMethod();return"异步方法调用成功";}}

测试类：

importorg.junit.jupiter.api.Test;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.boot.test.web.client.TestRestTemplate;importorg.springframework.boot.web.server.LocalServerPort;importstaticorg.assertj.core.api.Assertions.assertThat;@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classAsyncApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestAsyncMethod(){String response = restTemplate.getForObject("http://localhost:"+ port +"/async",String.class);assertThat(response).contains("异步方法调用成功");}}

✅ 结论：使用@Async注解是指使用Spring Boot异步通信的基本方法之一，作用是实现异步通信、提高应用程序的性能。

21.6.2 使用CompletableFuture

定义：使用CompletableFuture是指使用Spring Boot异步通信的基本方法之一。
作用：

  • 实现异步通信。
  • 提高应用程序的性能。

示例：
控制器类：

importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;importjava.util.concurrent.CompletableFuture;importjava.util.concurrent.ExecutionException;@RestControllerpublicclassCompletableFutureController{@GetMapping("/completableFuture")publicStringcompletableFuture()throwsExecutionException,InterruptedException{System.out.println("主线程执行："+Thread.currentThread().getName());CompletableFuture<Void> future =CompletableFuture.runAsync(()->{System.out.println("异步方法执行："+Thread.currentThread().getName());}); future.get();return"CompletableFuture调用成功";}}

测试类：

importorg.junit.jupiter.api.Test;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.boot.test.web.client.TestRestTemplate;importorg.springframework.boot.web.server.LocalServerPort;importstaticorg.assertj.core.api.Assertions.assertThat;@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classCompletableFutureApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestCompletableFuture(){String response = restTemplate.getForObject("http://localhost:"+ port +"/completableFuture",String.class);assertThat(response).contains("CompletableFuture调用成功");}}

✅ 结论：使用CompletableFuture是指使用Spring Boot异步通信的基本方法之一，作用是实现异步通信、提高应用程序的性能。

21.7 Spring Boot的实际应用场景

在实际开发中，Spring Boot消息队列与异步通信的应用场景非常广泛，如：

  • 实现用户注册的异步处理。
  • 实现订单的异步处理。
  • 实现邮件发送的异步处理。
  • 实现日志的异步处理。

示例：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;importorg.springframework.scheduling.annotation.EnableAsync;importorg.springframework.scheduling.annotation.Async;importorg.springframework.stereotype.Service;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;@SpringBootApplication@EnableAsyncpublicclassUserRegistrationApplication{publicstaticvoidmain(String[] args){SpringApplication.run(UserRegistrationApplication.class, args);}}@ServiceclassUserRegistrationService{@AsyncpublicvoidsendWelcomeEmail(String email){System.out.println("发送欢迎邮件："+ email);try{Thread.sleep(2000);}catch(InterruptedException e){ e.printStackTrace();}System.out.println("邮件发送成功："+ email);}}@RestControllerclassUserRegistrationController{@AutowiredprivateUserRegistrationService userRegistrationService;@GetMapping("/register")publicStringregisterUser(String email){System.out.println("用户注册："+ email); userRegistrationService.sendWelcomeEmail(email);return"用户注册成功";}}// 测试类@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classUserRegistrationApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestRegisterUser(){String email ="[email protected]";String response = restTemplate.getForObject("http://localhost:"+ port +"/register?email="+ email,String.class);assertThat(response).contains("用户注册成功");}}

输出结果：

控制台输出：

用户注册：[email protected] 发送欢迎邮件：[email protected] 邮件发送成功：[email protected]

✅ 结论：在实际开发中，Spring Boot消息队列与异步通信的应用场景非常广泛，需要根据实际问题选择合适的异步通信方法。

总结

本章我们学习了Spring Boot消息队列与异步通信，包括消息队列的定义与特点、Spring Boot与ActiveMQ的集成、Spring Boot与RabbitMQ的集成、Spring Boot与Kafka的集成、Spring Boot异步通信的基本方法、Spring Boot的实际应用场景，学会了在实际开发中处理消息队列与异步通信问题。其中，消息队列的定义与特点、Spring Boot与ActiveMQ的集成、Spring Boot与RabbitMQ的集成、Spring Boot与Kafka的集成、Spring Boot异步通信的基本方法、Spring Boot的实际应用场景是本章的重点内容。从下一章开始，我们将学习Spring Boot的其他组件、微服务等内容。

Read more

什么是web3

什么是web3

“Web3” 是当前科技与金融领域最热门的概念之一，但它常被误解为“加密货币”或“炒币”的代名词。实际上，Web3 是一场关于互联网权力结构、数据归属和价值分配的范式革命。要真正理解它，必须将其与 Web1、Web2 放在一起对比。 下面从 时间演进、核心特征、用户角色、技术基础、商业模式 五个维度，系统解析三者的本质区别： 一、时间线与历史背景 阶段时间名称核心驱动力Web11990s – 2004 年左右“只读网络”（Read-only Web）信息数字化、门户网站兴起Web22004 – 至今“可读可写网络”（Read-Write Web）社交媒体、移动互联网、平台经济Web32014 提出，2021+ 爆发“可读可写可拥有网络”（Read-Write-Own Web）区块链、去中心化、用户主权

By Ne0inhk
AI浪潮下，FPGA如何实现自我重塑与行业变革

AI浪潮下，FPGA如何实现自我重塑与行业变革

引言：AI 与 FPGA，新时代的碰撞 2025 年，人工智能技术迎来爆发式增长，大模型、生成式 AI 和多模态技术持续突破，人形机器人量产元年正式开启，自动驾驶商业化进程加速，工业数字化转型全面铺开(1)。在这一背景下，FPGA（现场可编程门阵列）作为一种灵活可编程的硬件平台，正经历前所未有的变革与重塑。FPGA 凭借其独特的并行处理能力、低延迟特性和可重构性，成为 AI 加速领域的重要力量，为实时处理、边缘计算和高性能计算提供了全新可能。 随着 AI 应用场景不断扩展，传统的 CPU 和 GPU 在某些场景下已无法满足需求，特别是在低延迟、高吞吐量和能效比方面。FPGA 作为一种高度灵活的硬件平台，能够根据不同的 AI 算法和应用需求进行定制化配置，实现接近 ASIC 的性能和能效，同时保持可编程的灵活性。这种特性使得 FPGA

By Ne0inhk
Windows安装Neo4j保姆级教程（图文详解）

Windows安装Neo4j保姆级教程（图文详解）

文章目录 * 前言 * 系统要求 * 安装Java环境 * 步骤1：检查Java版本 * 步骤2：下载Java JDK * 步骤3：安装Java JDK * 下载Neo4j * 步骤1：访问官方网站下载Neo4j * 步骤2：解压Neo4j * 启动Neo4j服务 * 步骤1：以管理员身份打开命令提示符 * 步骤2：导航到Neo4j的bin目录 * 步骤3：安装Neo4j服务 * 步骤4：启动Neo4j服务 * 步骤5：验证服务状态 * 访问Neo4j * 基本操作和配置 * 常用管理命令 * 配置文件修改 * 常见问题解决 * 问题1：端口被占用 * 问题2：Java版本不匹配 * 问题3：服务启动失败 * 总结 前言 Neo4j是一款强大的图数据库，特别适合处理复杂的关系数据。本教程将手把手教你在Windows系统上安装Neo4j，并配置可视化工具，让你快速上手图数据库的世界。 系统要求 在开始安装之前，请确保你的系统满足以下要求： 操作系统:

By Ne0inhk

全面的System Verilog教程：从基础到高级验证

本文还有配套的精品资源，点击获取 简介：System Verilog是用于系统级验证、芯片设计与验证以及FPGA实现的强大硬件描述语言。它扩展了Verilog的基础特性，支持高级语言结构，如类、接口、任务和函数，优化了验证流程。教程内容涵盖System Verilog的基础概念、结构化编程元素、并发与同步机制、现代验证方法学、UVM验证方法论以及标准库的应用。旨在教授学生掌握System Verilog语法和高级特性，实现高效、可维护的验证代码。 1. System Verilog概述及应用领域 1.1 System Verilog的起源与发展 System Verilog是作为硬件设计和验证领域的重要语言，由Verilog发展而来，随后被进一步扩展以满足现代电子设计自动化的需要。其发展始于20世纪90年代，目的是在原有Verilog HDL的基础上，提供更为强大的设计验证功能。 1.1.1 Verilog与VHDL的区别 虽然Verilog和VHDL都是硬件描述语言（HDL），但它们在语法和使用方法上存在差异。Verilog更接近于C语言，而VHDL的语法结构则更接近

By Ne0inhk