Spring Boot 消息队列集成与异步通信实战

接着，在 application.properties 中配置 Broker 地址及认证信息：

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

消息发送通常使用 JmsTemplate，而接收端则通过 @JmsListener 注解监听特定队列：

import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

@Component
public class MessageProducer {
    private final JmsTemplate jmsTemplate;

    public MessageProducer(JmsTemplate jmsTemplate) {
        this.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);
    }
}

配合 Controller 暴露接口进行测试，即可验证消息流转是否正常。

Spring Boot 与 RabbitMQ 集成

RabbitMQ 支持更复杂的交换器（Exchange）模式，适合需要灵活路由的场景。除了添加 spring-boot-starter-amqp 依赖外，还需要定义 Queue、Exchange 和 Binding 的配置类。

配置文件示例：

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

配置交换机和队列绑定关系：

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

生产者使用 RabbitTemplate 指定 Exchange 和 RoutingKey，消费者使用 @RabbitListener 监听队列：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

@Component
public class MessageProducer {
    private final RabbitTemplate rabbitTemplate;

    public MessageProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void sendMessage(String exchange, String routingKey, String message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
        System.out.println("发送消息：" + message);
    }
}

Spring Boot 与 Kafka 集成

Kafka 以高吞吐量著称，非常适合日志收集和事件流处理。集成时需注意序列化器的配置，确保生产和消费双方格式一致。

依赖与配置：

<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.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

核心代码较为简洁，使用 KafkaTemplate 发送，@KafkaListener 消费：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class MessageProducer {
    private final KafkaTemplate<String, String> kafkaTemplate;

    public MessageProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = 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);
    }
}

异步通信基本方法

除了引入外部消息中间件，Spring Boot 也提供了原生的异步处理能力，适用于轻量级任务。

使用 @Async 注解

这是最便捷的异步方式。只需在启动类上添加 @EnableAsync，并在需要异步执行的方法上标注 @Async。Spring 容器会自动将调用放入线程池执行，不阻塞主线程。

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

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

使用 CompletableFuture

对于需要组合多个异步结果或进行链式调用的场景，Java 8 提供的 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 调用成功";
    }
}

实际应用场景

在实际业务中，这些技术常用于用户注册后的欢迎邮件发送、订单状态通知等。例如，用户注册成功后立即返回响应，后台通过异步任务触发邮件服务，既提升了用户体验，又避免了数据库写入或网络请求阻塞主线程。

通过合理选择消息队列类型和异步策略，可以显著提升系统的响应速度和稳定性。开发者应根据具体业务需求，权衡延迟、吞吐量及一致性要求，选择合适的技术方案。