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

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

# 服务器端口
server.port=8080
# ActiveMQ 配置
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);
    }
}

控制器接口：

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 协议，以灵活的路由机制著称，适合复杂的消息分发场景。

1. 依赖与配置

RabbitMQ 需要添加 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

2. 交换机与队列绑定

RabbitMQ 的核心在于 Exchange（交换机）、Queue（队列）和 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");
    }
}

生产与消费逻辑：

// 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 擅长高吞吐量的日志处理和流式计算，采用 Topic 分区模型。

1. 依赖与配置

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

2. 生产与消费

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

// 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 也提供了轻量级的本地异步解决方案。

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

2. 使用 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();
        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 也是高效的工具。掌握这些技术，能帮助我们在高并发场景下构建更稳健的系统架构。