SpringBoot实现消息队列与异步通信

更新时间：2026年03月08日 10:50:29 作者：星辰徐哥

本文介绍了Spring Boot中消息队列与异步通信的核心概念与应用方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

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的实际应用场景。

2 消息队列概述

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

2.1 消息队列的定义

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

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

常见的消息队列：

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

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

2.2 消息队列的特点

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

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

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

3 Spring Boot与ActiveMQ的集成

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

3.1 集成ActiveMQ的步骤

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

创建Spring Boot项目。
添加所需的依赖。
配置ActiveMQ。
创建消息生产者。
创建消息消费者。
测试应用。

示例：

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

消息生产者：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

@Component
public class MessageProducer {
    @Autowired
    private JmsTemplate jmsTemplate;
    
    public void sendMessage(String destination, String message) {
        jmsTemplate.convertAndSend(destination, message);
        System.out.println("发送消息：" + message);
    }
}

消息消费者：

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class MessageConsumer {
    @JmsListener(destination = "test-queue")
    public void receiveMessage(String message) {
        System.out.println("接收消息：" + message);
    }
}

控制器类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MessageController {
    @Autowired
    private MessageProducer messageProducer;
    
    @GetMapping("/send")
    public String sendMessage(@RequestParam String message) {
        messageProducer.sendMessage("test-queue", message);
        return "消息发送成功";
    }
}

测试类：

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 static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class ActiveMQApplicationTests {
    @LocalServerPort
    private int port;
    
    @Autowired
    private TestRestTemplate restTemplate;
    
    @Test
    void contextLoads() {
    }
    
    @Test
    void testSendMessage() {
        String message = "Hello, ActiveMQ!";
        String response = restTemplate.getForObject("http://localhost:" + port + "/send?message=" + message, String.class);
        assertThat(response).contains("消息发送成功");
    }
}

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

4 Spring Boot与RabbitMQ的集成

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

4.1 集成RabbitMQ的步骤

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

创建Spring Boot项目。
添加所需的依赖。
配置RabbitMQ。
创建消息生产者。
创建消息消费者。
测试应用。

示例：

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

消息生产者：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MessageProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendMessage(String exchange, String routingKey, String message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
        System.out.println("发送消息：" + message);
    }
}

消息消费者：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MessageConsumer {
    @RabbitListener(queues = "test-queue")
    public void receiveMessage(String message) {
        System.out.println("接收消息：" + message);
    }
}

RabbitMQ配置类：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue testQueue() {
        return new Queue("test-queue", true);
    }
    
    @Bean
    public DirectExchange testExchange() {
        return new DirectExchange("test-exchange");
    }
    
    @Bean
    public Binding testBinding() {
        return BindingBuilder.bind(testQueue()).to(testExchange()).with("test-routing-key");
    }
}

控制器类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MessageController {
    @Autowired
    private MessageProducer messageProducer;
    
    @GetMapping("/send")
    public String sendMessage(@RequestParam String message) {
        messageProducer.sendMessage("test-exchange", "test-routing-key", message);
        return "消息发送成功";
    }
}

测试类：

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 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 testSendMessage() {
        String message = "Hello, RabbitMQ!";
        String response = restTemplate.getForObject("http://localhost:" + port + "/send?message=" + message, String.class);
        assertThat(response).contains("消息发送成功");
    }
}

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

5 Spring Boot与Kafka的集成

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

5.1 集成Kafka的步骤

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

创建Spring Boot项目。
添加所需的依赖。
配置Kafka。
创建消息生产者。
创建消息消费者。
测试应用。

示例：
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

消息生产者：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class MessageProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
        System.out.println("发送消息：" + message);
    }
}

消息消费者：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MessageConsumer {
    @KafkaListener(topics = "test-topic", groupId = "test-group")
    public void receiveMessage(String message) {
        System.out.println("接收消息：" + message);
    }
}

控制器类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MessageController {
    @Autowired
    private MessageProducer messageProducer;
    
    @GetMapping("/send")
    public String sendMessage(@RequestParam String message) {
        messageProducer.sendMessage("test-topic", message);
        return "消息发送成功";
    }
}

测试类：

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 static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class KafkaApplicationTests {
    @LocalServerPort
    private int port;
    
    @Autowired
    private TestRestTemplate restTemplate;
    
    @Test
    void contextLoads() {
    }
    
    @Test
    void testSendMessage() {
        String message = "Hello, Kafka!";
        String response = restTemplate.getForObject("http://localhost:" + port + "/send?message=" + message, String.class);
        assertThat(response).contains("消息发送成功");
    }
}

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

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

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

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>

异步配置类：

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;

@Configuration
@EnableAsync
public class AsyncConfig {
}

异步服务类：

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

@Service
public class AsyncService {
    @Async
    public void asyncMethod() {
        System.out.println("异步方法执行：" + Thread.currentThread().getName());
    }
}

控制器类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class AsyncController {
    @Autowired
    private AsyncService asyncService;
    
    @GetMapping("/async")
    public String asyncMethod() {
        System.out.println("主线程执行：" + Thread.currentThread().getName());
        asyncService.asyncMethod();
        return "异步方法调用成功";
    }
}

测试类：

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 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 testAsyncMethod() {
        String response = restTemplate.getForObject("http://localhost:" + port + "/async", String.class);
        assertThat(response).contains("异步方法调用成功");
    }
}

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

6.2 使用CompletableFuture

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

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

示例：
控制器类：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@RestController
public class CompletableFutureController {
    @GetMapping("/completableFuture")
    public String completableFuture() throws ExecutionException, InterruptedException {
        System.out.println("主线程执行：" + Thread.currentThread().getName());
        
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println("异步方法执行：" + Thread.currentThread().getName());
        });
        
        future.get();
        return "CompletableFuture调用成功";
    }
}

测试类：

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 static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class CompletableFutureApplicationTests {
    @LocalServerPort
    private int port;
    
    @Autowired
    private TestRestTemplate restTemplate;
    
    @Test
    void contextLoads() {
    }
    
    @Test
    void testCompletableFuture() {
        String response = restTemplate.getForObject("http://localhost:" + port + "/completableFuture", String.class);
        assertThat(response).contains("CompletableFuture调用成功");
    }
}

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

7 Spring Boot的实际应用场景

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

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

示例：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication
@EnableAsync
public class UserRegistrationApplication {
    public static void main(String[] args) {
        SpringApplication.run(UserRegistrationApplication.class, args);
    }
}

@Service
class UserRegistrationService {
    @Async
    public void sendWelcomeEmail(String email) {
        System.out.println("发送欢迎邮件：" + email);
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("邮件发送成功：" + email);
    }
}

@RestController
class UserRegistrationController {
    @Autowired
    private UserRegistrationService userRegistrationService;
    
    @GetMapping("/register")
    public String registerUser(String email) {
        System.out.println("用户注册：" + email);
        userRegistrationService.sendWelcomeEmail(email);
        return "用户注册成功";
    }
}

// 测试类
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class UserRegistrationApplicationTests {
    @LocalServerPort
    private int port;
    
    @Autowired
    private TestRestTemplate restTemplate;
    
    @Test
    void contextLoads() {
    }
    
    @Test
    void testRegisterUser() {
        String email = "test@example.com";
        String response = restTemplate.getForObject("http://localhost:" + port + "/register?email=" + email, String.class);
        assertThat(response).contains("用户注册成功");
    }
}

输出结果：

访问http://localhost:8080/register?email=test@example.com：返回用户注册成功。

控制台输出：

用户注册：test@example.com
发送欢迎邮件：test@example.com
邮件发送成功：test@example.com

✅ 结论：在实际开发中，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的其他组件、微服务等内容。

到此这篇关于SpringBoot实现消息队列与异步通信的文章就介绍到这了,更多相关SpringBoot 消息队列与异步通信内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

SpringBoot 集成消息队列实战指南(RabbitMQ/Kafka):异步通信与解耦,落地高可靠消息传递

解决异常:Invalid keystore format,springboot配置ssl证书格式不合法问题
这篇文章主要介绍了解决异常:Invalid keystore format,springboot配置ssl证书格式不合法问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2024-03-03
Java过滤器模式原理及用法实例
这篇文章主要介绍了Java过滤器模式原理及用法实例,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-05-05
深入理解Java @Entity注解及其与数据库表的关联
这篇文章主要介绍了深入理解Java @Entity注解及其与数据库表的关联,@Entity注解在JPA中占据核心地位,它建立起Java实体类和数据库表之间的映射关系,需要的朋友可以参考下
2025-05-05
JAVA简单实现MD5注册登录加密实例代码
本篇文章主要介绍了JAVA简单实现MD5注册登录加密实例代码，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-03-03
Java Swing BoxLayout箱式布局的实现代码
这篇文章主要介绍了Java Swing BoxLayout箱式布局的实现代码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-12-12
IntelliJ IDEA配置java环境及解决IDEA不能直接运行单个JAVA文件的问题
这篇文章主要介绍了IntelliJ IDEA配置java环境及解决IDEA不能直接运行单个JAVA文件的问题,本文通过图文并茂的形式给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2020-07-07
Java与Python两种编程语言的比较与应用举例详解
这篇文章主要介绍了Java与Python两种编程语言比较与应用的相关资料,Java和Python各有特点,Java适用于企业级应用开发,Python则在数据科学和机器学习领域占优势,两者在语法、应用领域、性能、开发效率等方面存在差异,需要的朋友可以参考下
2025-02-02
解决MyBatis返回结果类型为Boolean的问题
这篇文章主要介绍了解决MyBatis返回结果类型为Boolean的问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-11-11
logback自定义json日志输出示例详解
这篇文章主要为大家介绍了logback自定义json日志输出，就是通过logback日志体系以及logstash提供的json log依赖将数据以json格式记录到日志文件的例子
2022-03-03
spring-ai-alibaba文本转语音的实例
在使用Spring-ai-alibaba进行文本转语音项目时,遇到DashScopeException错误,原因是响应中的字段未被识别,通过在项目中创建目录类,复制DashScopeAudioTranscriptionApi类并在指定位置增加字段,解决了该问题
2025-11-11