Spring Boot 集成 RabbitMQ 实现异步消息处理

接着，在 application.properties 中配置连接信息。默认情况下，RabbitMQ 运行在 localhost 的 5672 端口，用户名为 guest。

# 服务器端口
server.port=8080
# RabbitMQ 连接信息
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

2. 定义交换机与队列

为了让消息正确路由，我们需要在代码中声明 Exchange（交换机）、Queue（队列）以及 Binding（绑定关系）。这里采用 DirectExchange 模式。

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    private static final String EXCHANGE_NAME = "product-exchange";
    private static final String ROUTING_KEY = "product-routing-key";
    private static final String QUEUE_NAME = "product-queue";

    @Bean
    public DirectExchange productExchange() {
        return new DirectExchange(EXCHANGE_NAME);
    }

    @Bean
    public Queue productQueue() {
        return new Queue(QUEUE_NAME);
    }

    @Bean
    public Binding productBinding(Queue productQueue, DirectExchange productExchange) {
        return BindingBuilder.bind(productQueue).to(productExchange).with(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 ProductMessageProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private static final String EXCHANGE_NAME = "product-exchange";
    private static final String ROUTING_KEY = "product-routing-key";

    public void sendProductMessage(String message) {
        // convertAndSend 方法会自动序列化消息
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, message);
        System.out.println("消息已发送：" + message);
    }
}

消费者组件：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class ProductMessageConsumer {

    @RabbitListener(queues = "product-queue")
    public void receiveProductMessage(String message) {
        System.out.println("消息已接收：" + message);
        processProductMessage(message);
    }

    private void processProductMessage(String message) {
        try {
            // 模拟耗时业务逻辑，例如发送邮件、生成报表等
            Thread.sleep(2000);
            System.out.println("消息处理完成：" + message);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
        }
    }
}

4. 暴露接口与测试

最后，通过 Controller 暴露 HTTP 接口触发消息发送，并启动应用进行测试。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/products")
public class ProductController {

    @Autowired
    private ProductMessageProducer productMessageProducer;

    @PostMapping("/send")
    public String sendProductMessage(@RequestBody String message) {
        productMessageProducer.sendProductMessage(message);
        return "消息已发送";
    }
}

启动应用后，访问 http://localhost:8080/api/products/send 并传入 JSON 字符串 {"message": "Hello, RabbitMQ!"}。控制台会立即输出'消息已发送'，随后消费者会在约 2 秒后打印'消息处理完成'。这种非阻塞的处理方式正是异步架构的优势所在。

总结

通过上述步骤，我们完成了 Spring Boot 与 RabbitMQ 的基础集成。在实际开发中，消息队列不仅适用于订单处理、支付通知等场景，还能有效应对突发流量。关键在于根据业务需求选择合适的队列模型，并做好异常处理和消息持久化配置，确保数据不丢失。