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

YOLO26来了，更好、更快、更小的 YOLO 模型，使用YOLO26训练自己的数据集和推理(附YOLO26网络结构图)，租用 GPU 服务器训练教程，YOLO26创新点解析

YOLO26来了，更好、更快、更小的 YOLO 模型，使用YOLO26训练自己的数据集和推理(附YOLO26网络结构图)，租用 GPU 服务器训练教程，YOLO26创新点解析

目录 * 摘要 * YOLO26更新点 * ⚡⚡C3k2 小优化 * ☑️ YOLO26 C3k2代码 * ☑️ YOLO11 C3k2代码 * ⚡⚡移除分布焦点损失（DFL） * ⚡⚡端到端、无需 NMS 推理 * ⚡⚡ProgLoss 与 STAL * ☑️ProgLoss * ☑️STAL （小目标感知标签分配） * ⚡⚡MuSGD优化器 * 从机器人到制造业：YOLO26 的用例 * 🐴一、YOLO26 源码下载与模型下载 * ⚡⚡YOLO26模型结构图 * ⚡⚡1.源码下载 * ⚡⚡2.官网的预训练模型下载 * 🐴二、数据集准备 * ⚡⚡LabelImg & Labelme * ☑️ LabelImg（仅限矩形检测框） * ☑️ Labelme * ⚡⚡ X-AnyLabeling * ⚡⚡ 旋转框 (OBB) 标注工具：roLabelImg * ⚡⚡1.目标检测数据集标注软件 * ⚡⚡2.voc数据集格式转换

By Ne0inhk
高并发系统的网络参数优化与服务超时管理（QPS 10w+）

高并发系统的网络参数优化与服务超时管理（QPS 10w+）

个人名片 🎓作者简介：java领域优质创作者 🌐个人主页：码农阿豪 📞工作室：新空间代码工作室（提供各种软件服务) 💌个人邮箱：[[email protected]] 📱个人微信：15279484656 🌐个人导航网站：www.forff.top 💡座右铭：总有人要赢。为什么不能是我呢？ * 专栏导航： 码农阿豪系列专栏导航 面试专栏：收集了java相关高频面试题，面试实战总结🍻🎉🖥️ Spring5系列专栏：整理了Spring5重要知识点与实战演练，有案例可直接使用🚀🔧💻 Redis专栏：Redis从零到一学习分享，经验总结，案例实战💐📝💡 全栈系列专栏：海纳百川有容乃大，可能你想要的东西里面都有🤸🌱🚀 目录 * 高并发系统的网络参数优化与服务超时管理 * 背景与现状 * 问题分析 * 网络参数优化实践 * 1. 文件句柄限制 * 2. 调整端口范围 * 3. 连接复用 * 4. 最大连接数 * 5.

By Ne0inhk
Flutter 组件 fletch 的适配 鸿蒙Harmony 实战 - 驾驭高性能网络爬虫、实现鸿蒙端多并发与自定义拦截器的资产自动化抓取方案

Flutter 组件 fletch 的适配 鸿蒙Harmony 实战 - 驾驭高性能网络爬虫、实现鸿蒙端多并发与自定义拦截器的资产自动化抓取方案

欢迎加入开源鸿蒙跨平台社区：https://openharmonycrossplatform.ZEEKLOG.net Flutter 组件 fletch 的适配 鸿蒙Harmony 实战 - 驾驭高性能网络爬虫、实现鸿蒙端多并发与自定义拦截器的资产自动化抓取方案 前言 在数据驱动的鸿蒙（OpenHarmony）应用开发中，很多时候我们需要从外部网络环境大规模采集实时资讯、获取海量资源路径或者是进行自动化的接口探测。传统的 http 库虽然简单，但在面对数十路并发下载、复杂的 Cookie 状态维持以及多级的请求拦截（Interceptor）时，往往显得捉襟见肘。 fletch 正是一款专为高性能、工业级抓取任务设计的 Dart 网络增强库。它不仅支持极致的并发限流，更提供了一套类似拦截器管线的强大插件化能力。 适配到鸿蒙系统后，配合鸿蒙底层的网络切片和能效策略，fletch 能让你的数据采集应用在保持低功耗的同时，展现出前所未有的吞吐力。本文将为你深入剖析 fletch 在鸿蒙实战环境下的深度集成与优化。 一、原理解析 / 概念介绍 1.1

By Ne0inhk