当前位置：首页 > article >正文

Spring Boot 整合 RabbitMQ：注解声明队列与交换机详解

article 2025/3/26 1:30:35

RabbitMQ 作为一款高性能的消息中间件，在分布式系统中广泛应用。Spring Boot 通过 spring-boot-starter-amqp 提供了对 RabbitMQ 的无缝集成，开发者可以借助注解快速声明队列、交换机及绑定规则，极大简化了配置流程。本文将通过代码示例和原理分析，详细介绍如何用注解实现 RabbitMQ 的集成，并深入解析交换机的作用与类型。

一、环境准备

1. 添加依赖

在 pom.xml 中引入 Spring Boot 的 AMQP 依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. 配置 RabbitMQ 连接

在 application.yml 中配置 RabbitMQ 连接信息：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

二、注解声明队列与交换机

Spring Boot 提供 @Configuration 和 @Bean 注解来声明队列、交换机及绑定关系，同时也支持 @RabbitListener 简化消费者监听逻辑。

1. 声明队列与交换机

创建配置类 RabbitMQConfig.java：

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    // 声明 Direct 交换机
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("direct.exchange");
    }

    // 声明队列
    @Bean
    public Queue demoQueue() {
        return new Queue("demo.queue", true); // 持久化队列
    }

    // 绑定队列到交换机，并指定路由键
    @Bean
    public Binding binding(Queue demoQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(demoQueue)
                .to(directExchange)
                .with("demo.routing.key");
    }
}

2. 生产者发送消息

通过 RabbitTemplate 发送消息到交换机：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MessageProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String message) {
        // 发送到交换机，指定路由键
        rabbitTemplate.convertAndSend("direct.exchange", "demo.routing.key", message);
    }
}

3. 消费者监听消息

使用 @RabbitListener 注解监听队列：

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

@Component
public class MessageConsumer {

    @RabbitListener(queues = "demo.queue")
    public void handleMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

三、交换机（Exchange）详解

1. 交换机的作用

交换机是消息路由的核心组件，负责根据 路由规则 将生产者发送的消息分发到队列。其行为由 交换机类型 和 绑定规则（Binding） 共同决定。

2. 常见交换机类型

交换机类型	路由规则	适用场景
Direct	精确匹配 `Routing Key` 和 `Binding Key`。	一对一精准路由（如订单状态更新）。
Topic	支持通配符 `*`（匹配一个词）和 `#`（匹配零或多个词）。	多维度分类消息（如日志分级）。
Fanout	忽略 `Routing Key`，广播到所有绑定队列。	群发通知（如系统公告）。
Headers	通过消息头（Headers）的键值对匹配，而非 `Routing Key`。	复杂条件路由（较少使用）。

3. 示例：Topic 交换机配置

@Configuration
public class TopicExchangeConfig {

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topic.exchange");
    }

    @Bean
    public Queue topicQueue1() {
        return new Queue("topic.queue1");
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue("topic.queue2");
    }

    @Bean
    public Binding binding1(TopicExchange topicExchange, Queue topicQueue1) {
        // 路由键格式：order.* 匹配 order.create、order.pay 等
        return BindingBuilder.bind(topicQueue1)
                .to(topicExchange)
                .with("order.*");
    }

    @Bean
    public Binding binding2(TopicExchange topicExchange, Queue topicQueue2) {
        // 路由键格式：order.# 匹配 order.create.success、order.pay.failed 等
        return BindingBuilder.bind(topicQueue2)
                .to(topicExchange)
                .with("order.#");
    }
}

四、高级特性：消息确认与持久化

1. 生产者确认（Publisher Confirm）

在 application.yml 中启用确认机制：

spring:
  rabbitmq:
    publisher-confirm-type: correlated # 异步确认
    publisher-returns: true

2. 消费者手动 ACK

修改消费者监听逻辑，手动确认消息：

@RabbitListener(queues = "demo.queue")
public void handleMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        System.out.println("Received message: " + message);
        channel.basicAck(tag, false); // 手动确认
    } catch (IOException e) {
        channel.basicNack(tag, false, true); // 拒绝并重新入队
    }
}

3. 消息持久化

在声明队列和交换机时启用持久化：

@Bean
public Queue durableQueue() {
    return new Queue("durable.queue", true, true, false); // 参数：名称、持久化、独占、自动删除
}

@Bean
public DirectExchange durableExchange() {
    return new DirectExchange("durable.exchange", true, false); // 参数：名称、持久化、自动删除
}