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

配合 Controller 暴露接口测试：

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 "消息发送成功";
    }
}

集成 RabbitMQ

RabbitMQ 因其灵活的路由策略被广泛采用。Spring Boot 通过 spring-boot-starter-amqp 提供封装。

1. 依赖与配置

<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>
</dependencies>

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

2. 交换机与队列绑定

RabbitMQ 的核心在于 Exchange 和 Binding。我们需要定义 Bean 来管理这些资源。

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. 收发逻辑

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

集成 Kafka

Kafka 更适合高吞吐量的日志或事件流场景。Spring Kafka 模块简化了模板的使用。

1. 依赖与配置

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

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

Spring Boot 异步通信基础

除了引入外部 MQ，Spring Boot 自身也提供了轻量级的异步解决方案。

使用 @Async 注解

这是最便捷的异步方法调用方式。只需在启动类添加 @EnableAsync，并在服务方法上标注 @Async。

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

注意：如果调用发生在同一个类内部，由于 AOP 代理机制，@Async 可能不会生效，建议跨类调用。

使用 CompletableFuture

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

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(); // 阻塞等待结果，实际业务中通常结合 thenApply 等链式调用
        return "CompletableFuture 调用成功";
    }
}

实际应用场景

在实际业务中，异步处理常用于非核心链路，避免阻塞主流程。

以用户注册为例，注册成功后发送邮件通知属于辅助操作，不应影响注册接口的响应速度。

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 "用户注册成功";
    }
}

访问 /[email protected]，控制台会显示注册信息立即返回，而邮件发送在后台线程完成。这种设计显著提升了用户体验。

总结

本章我们梳理了 Spring Boot 在消息队列与异步通信方面的核心能力。无论是选择 ActiveMQ、RabbitMQ 还是 Kafka，关键在于根据业务场景（如吞吐量要求、可靠性需求）进行选型。同时，利用 @Async 和 CompletableFuture 也能在不引入额外中间件的情况下解决部分异步问题。掌握这些技术，能帮助我们在构建高可用、高性能的微服务架构时游刃有余。