Spring Boot 消息队列与异步通信

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、创建消息生产者、创建消息消费者、测试应用。

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

消息生产者：

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、创建消息生产者、创建消息消费者、测试应用。

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

消息生产者：

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、创建消息生产者、创建消息消费者、测试应用。

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>

异步配置类：

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 异步通信的基本方法之一，作用是实现异步通信、提高应用程序的性能。

21.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 异步通信的基本方法之一，作用是实现异步通信、提高应用程序的性能。

21.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 = "[email protected]";
        String response = restTemplate.getForObject("http://localhost:" + port + "/register?email=" + email, String.class);
        assertThat(response).contains("用户注册成功");
    }
}

输出结果：

• 访问 http://localhost:8080/[email protected]：返回用户注册成功。

控制台输出：

用户注册：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 的其他组件、微服务等内容。