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

NewBie-image-Exp0.1从零开始：Python调用大模型生成图片教程

NewBie-image-Exp0.1从零开始：Python调用大模型生成图片教程 你是否也曾经被那些精美的动漫角色图吸引，却苦于不会画画？或者想快速生成一批风格统一的角色素材，但手动设计成本太高？今天我们要聊的这个工具，或许能彻底改变你的创作方式。 NewBie-image-Exp0.1 是一个专注于高质量动漫图像生成的大模型项目。它不仅具备强大的视觉表现力，还引入了独特的 XML 提示词机制，让你可以像写代码一样精确控制每一个角色的属性。更棒的是，现在有一个预配置好的镜像版本，省去了繁琐的环境搭建和依赖安装过程，真正实现“开箱即用”。 本文将带你一步步上手使用这个镜像，从最基础的运行测试脚本，到理解其核心功能，再到如何自定义提示词来生成你想要的画面。无论你是AI绘画的新手，还是有一定经验的技术爱好者，都能在这篇文章中找到实用的信息。 1. 镜像简介与核心优势 NewBie-image-Exp0.1 并不是一个简单的开源项目打包，而是一个经过深度优化和修复的完整推理环境。它的最大价值在于解决了原项目部署过程中常见的三大难题：环境冲突、源码Bug 和模型下载困难。 1.1

By Ne0inhk
【Python 量化入门】AKshare 保姆级使用教程：零成本获取股票 / 基金 / 期货全市场金融数据

【Python 量化入门】AKshare 保姆级使用教程：零成本获取股票 / 基金 / 期货全市场金融数据

做量化交易、财经数据分析、投资复盘的开发者和投资者，经常会遇到核心痛点：付费金融数据接口成本高、免费 API 注册流程繁琐、多市场数据分散难以整合。告别 QMT 回测烦恼！手把手教你搭建 MiniQMT+Backtrader 量化回测框架 本文就给大家详细讲解 Python 量化圈的开源神器AKshare，从安装到核心功能实战全覆盖，代码可直接复制运行，零基础也能一键获取全市场金融行情数据。 一、AKshare 是什么？ AKshare 是一款基于 Python 开发的开源金融数据接口库，专为个人投资者、量化爱好者、财经数据分析人员打造，是目前国内生态最完善、维护最活跃的免费金融数据工具之一。 它支持股票、期货、基金、外汇、债券、指数、加密货币等多种主流金融市场的数据获取，核心优势如下： * 免费开源：完全开源免费，无隐藏收费，个人非商用零成本使用，无需开通付费会员 * 数据覆盖全面：A 股、

By Ne0inhk

Python 2026 年发展局势：AI 时代的 “通用基础设施语言”

2026 年的 Python 已从 “热门编程语言” 进化为全球数字生态的核心基础设施语言，其地位不仅稳固且进一步强化，同时也面临新的机遇与挑战，整体呈现 “一核多翼、优势固化、局部竞争” 的格局。 一、核心优势：AI + 全生态双轮驱动，地位无可替代 1. AI / 大模型领域的绝对霸主这是 Python 最核心的护城河。2026 年大模型落地、AI Agent 开发、多模态应用、低代码 AI 工具等场景中，Python 依然是95% 以上开发者的首选语言： * 生态垄断：PyTorch 3.0、TensorFlow 2.18、LangChain 2.0、Transformers 等核心框架均以 Python 为第一开发语言； * 效率优势：

By Ne0inhk
Python 实战：Boss 直聘职位信息爬虫开发全解析​

Python 实战：Boss 直聘职位信息爬虫开发全解析​

在求职和职场数据分析场景中，获取结构化的职位信息能为我们提供极大的便利 —— 无论是对比薪资水平、分析行业需求，还是研究企业招聘偏好，都需要可靠的数据源支持。本文将手把手教你用 Python 开发一个 Boss 直聘爬虫，通过监听网络请求的方式高效获取职位数据，并将结果保存为 Excel 文件。 一、开发前准备：环境与工具 在开始编码前，我们需要搭建好开发环境并明确核心依赖库的作用，确保后续开发过程顺畅。 1. 环境要求 * Python 3.8 及以上版本（推荐 3.10，兼容性更好） * 浏览器：Chrome 或 Edge（需与 Chromium 内核驱动版本匹配） 2. 核心依赖库 本文爬虫主要依赖 4 个关键库，可通过pip install 库名命令安装： * DrissionPage：一款强大的浏览器自动化工具，支持控制浏览器、监听网络请求，

By Ne0inhk