Spring Boot 消息队列与异步通信详解

接着在 application.properties 中指定 Broker 地址：

server.port=8080
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin

2. 生产者与消费者

使用 JmsTemplate 发送消息，配合 @JmsListener 注解监听队列。

生产者代码：

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);
    }
}

控制器只需注入生产者并调用即可，测试时可通过 HTTP 接口触发。

集成 RabbitMQ

RabbitMQ 基于 AMQP 协议，支持复杂的路由规则。Spring Boot 通过 spring-boot-starter-amqp 进行集成。

1. 基础配置

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

2. 交换机与队列绑定

RabbitMQ 的核心在于 Exchange 和 Binding。我们需要定义队列、交换机以及它们之间的路由键。

import org.springframework.amqp.core.*;
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");
    }
}

3. 收发消息

使用 RabbitTemplate 发送，@RabbitListener 监听特定队列。

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
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);
    }
}

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

集成 Kafka

Kafka 适用于海量数据场景，采用 Topic 模型，配置相对简单但需注意序列化。

1. 依赖与配置

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

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

2. 生产与消费

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.annotation.KafkaListener;
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);
    }
}

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

异步通信基本方法

除了消息队列，Spring Boot 还提供了更轻量的异步处理方式。

1. 使用 @Async 注解

这是最简单的异步方法调用方式。需要在启动类或配置类上添加 @EnableAsync 注解。

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

@Configuration
@EnableAsync
public class AsyncConfig {}

服务层方法加上 @Async 即可独立线程执行：

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());
    }
}

2. 使用 CompletableFuture

对于需要组合多个异步任务或获取返回值的场景，CompletableFuture 更加灵活。

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

@RestController
public class CompletableFutureController {
    @GetMapping("/completableFuture")
    public String completableFuture() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println("异步任务执行：" + Thread.currentThread().getName());
        });
        future.get(); // 阻塞等待完成，实际生产中通常不直接阻塞主线程
        return "CompletableFuture 调用成功";
    }
}

实际应用场景

在实际开发中，我们常利用上述技术处理耗时操作，避免阻塞主线程。

例如用户注册流程：注册成功后，发送欢迎邮件不需要同步等待，而是异步处理。

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

@Service
public 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);
    }
}

这样，用户注册接口可以立即返回，邮件发送在后台线程完成，用户体验更佳。

总结

本章涵盖了 Spring Boot 对接主流消息队列（ActiveMQ、RabbitMQ、Kafka）的完整流程，以及原生异步注解和 CompletableFuture 的使用。掌握这些工具，能帮助你在微服务架构中更好地处理解耦、削峰填谷及非核心业务的异步化，提升系统整体稳定性。