Spring Boot 消息队列与异步通信

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

【Java Web学习 | 第四篇】CSS(3) -背景

【Java Web学习 | 第四篇】CSS(3) -背景

🌈个人主页: Hygge_Code🔥热门专栏:从0开始学习Java | Linux学习| 计算机网络💫个人格言: “既然选择了远方,便不顾风雨兼程” 文章目录 * CSS背景样式全解析🥝 * 4.1 背景颜色 (`background-color`) * 4.2 背景图片 (`background-image`) * 4.3 背景平铺 (`background-repeat`) * 4.4 背景图片位置 (`background-position`) * 4.5 背景图像固定 (`background-attachment`) * 4.6 背景属性复合写法 * 4.7 背景色半透明 (`rgba`) * 综合代码演示 * 学习资源推荐🐦‍🔥 CSS背景样式全解析🥝 在网页设计中,背景样式是塑造页面视觉效果的关键元素之一。通过CSS的背景属性,我们可以为页面添加丰富的视觉效果,包括背景颜色、背景图片、平铺方式、定位以及固定等。

飞算JavaAI插件深度体验:三天任务半小时搞定,AI编程的革命性突破

飞算JavaAI插件深度体验:三天任务半小时搞定,AI编程的革命性突破

文章目录 * 飞算JavaAI插件深度体验:三天任务半小时搞定,AI编程的革命性突破 * 一、引言:从绝望到惊喜的转变 * 二、飞算JavaAI初体验:对话即开发 * 2.1 插件安装与配置 * 2.2 需求分析:AI理解业务场景 * 2.3 接口设计:从需求到API * 2.4 数据库设计:智能表结构生成 * 2.5 业务逻辑生成:核心功能实现 * 2.6 一键生成源码:项目快速构建 * 三、项目构建完成:效率提升的震撼体验 * 3.1 代码质量评估 * 四、实际运行测试:验证AI生成代码的可用性 * 4.1 接口测试结果 * 五、性能对比:传统开发 VS AI辅助开发

【JDK】-JDK 17 新特性整理(比较全)

【JDK】-JDK 17 新特性整理(比较全)

JDK 17 新特性 1、JDK 17中的Pattern类增强了哪些功能? 1. 新增asMatchPredicate方法: JDK 17的Pattern类新增了asMatchPredicate方法,可以将正则表达式编译为Predicate。 2. 增强了Unicode属性支持: JDK 17中的Pattern类增强了对Unicode属性的支持,使得正则表达式可以更好地处理Unicode字符。 3. 引入了新的转义语法: JDK 17引入了一种新的转义语法,可以更方便地转义正则表达式中的特殊字符,提高了正则表达式的可读性和可维护性。 4. 优化了性能: JDK 17对Pattern类的底层实现进行了优化,提升了正则表达式匹配的性能和效率。 5. 增加了对断言的支持: JDK 17中的Pattern类增加了对断言的支持,可以更灵活地进行正则表达式的匹配和处理。 2、JDK 17中的HTTP/2 Client有哪些新特性? 1. 提供了WebSocket支持: JDK 17中的HTTP/2 Client新增了对WebSocket的支持,使得开发者可以更方便地进行WebS