Spring Boot 中接入 MQ 与异步处理

技术架构图

在分布式系统里，消息队列和异步调用经常一起出现。前者负责把服务之间的耦合拆开，后者负责把本地耗时任务从主流程里挪出去。Spring Boot 对这两类能力都支持得比较顺手，接入 ActiveMQ、RabbitMQ、Kafka，或者直接用 @Async、CompletableFuture 做本地异步，成本都不高。

消息队列到底解决什么问题

MQ 的核心不是'更高级的传输方式'，而是把发送和消费拆开。消息先落到中间件里，谁来消费、什么时候消费，不再和发送方强绑定。

它最常见的价值就这几个：

异步处理：接口先返回，慢任务后面再做。
服务解耦：系统之间不用直接互相调用。
削峰：流量上来时先缓一缓，别把后端打穿。
可靠投递：消息可以持久化，减少直接丢失的风险。

ActiveMQ、RabbitMQ、Kafka 都属于这一类，但定位不一样。老项目里我见到 ActiveMQ 还挺多；路由规则复杂一些，RabbitMQ 更顺手；要是吞吐和日志流处理放在第一位，Kafka 更合适。

接入 ActiveMQ

ActiveMQ 是比较典型的 JMS 方案，适合那些已经在 Java EE 体系里跑了很久的系统。新项目不一定非它不可，但如果团队已经熟，迁移和维护都省事。

依赖与配置

先在 pom.xml 里加 Web 和 ActiveMQ 的 starter：

<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 里配置 Broker 地址和账号密码：

# 服务器端口
server.port=8080
# ActiveMQ 配置
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin

生产者和消费者

发送消息用 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);
    }
}

控制器接口：

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 用的是 AMQP 协议，强项在路由。队列不复杂，但分发规则多的时候，它比'只有一个 topic 的思路'更灵活。

依赖与配置

先加 spring-boot-starter-amqp：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <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>

配置如下：

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

交换机和队列绑定

RabbitMQ 的关键是 Exchange、Queue 和 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");
    }
}

生产和消费逻辑：

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

// Consumer
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 更偏向高吞吐和流式处理。它不是'发一条、收一条'这么简单，分区、消费组这些概念会直接影响系统设计。要是业务里消息量上得快，Kafka 往往比前两者更能扛。

依赖与配置

Kafka 用 spring-kafka：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <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>

配置示例：

server.port=8080
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

生产和消费

Kafka 发送用 KafkaTemplate，监听用 @KafkaListener。

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

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

本地异步处理

不一定每个异步场景都要上 MQ。只是把一个慢方法拆出去的话，Spring Boot 自己的异步支持已经够用了。

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

调用入口：

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 "异步方法调用成功";
    }
}

`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.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 偏传统 JMS，RabbitMQ 强在路由，Kafka 更适合高吞吐和流式场景。要是只是把本地耗时逻辑拆出去，@Async 和 CompletableFuture 已经能解决不少问题。真正要注意的不是'能不能发消息'，而是消息失败、重复消费、幂等这些后续问题，很多坑都藏在这里。