Javajava

Spring Boot 整合 RabbitMQ 实战

综述由AI生成Spring Boot 整合 RabbitMQ 支持 Fanout、Direct 和 Topic 三种模式。通过配置类或注解方式定义交换机、队列及绑定关系，演示了生产者和消费者的业务代码实现。重点展示了依赖引入、YAML 配置以及消息发送与接收的具体逻辑，适用于基于 Spring AMQP 的消息队列开发场景。

ApiHolic发布于 2026/3/15更新于 2026/4/2614 浏览

本节讲解 RabbitMQ 与 Spring Boot 的整合方式。需提前准备好 Spring 项目基本框架。

整合时有两种方式创建交换机、队列及它们的绑定关系：

添加配置类（加上@Configuration注解和@Bean注解），适用于 topic 模式、direct 模式，也需要用到@RabbitListener，只起监听队列的作用。
用注解创建（@RabbitListener），注解中的内容比之增加，用来创建交换机、队列及它们的绑定关系，以及监听队列。

Fanout 模式

1. 引入依赖

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

然后在 Spring 项目中创建两个模块，一个生产者模块，一个消费者模块。

2. 在 yml/properties 文件进行相应的配置

分别在两个模块中都进行创建。

server:
  port: 8080
spring:
  application:
    name: RabbitMQ-demo
  rabbitmq:
    username: guest
    password: guest
    virtual-host: /
    host: 127.0.0.1
    port: 5672

3. 创建配置类（创建相应的交换机与队列，且绑定关系）

分别在两个模块中都进行创建（防止一方没有配置类启动时报错）。

package com.lx.producer.config;

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


   {
    
    
     FanoutExchange  {
        
          (, , );
    }

    
    
     Queue  {
          (, );
    }

    
     Queue  {
          (, );
    }

    
     Queue  {
          (, );
    }

    
    
     Binding  {
         BindingBuilder.bind(smsQueue()).to(fanoutExchange());
    }

    
     Binding  {
         BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
    }

    
     Binding  {
         BindingBuilder.bind(emailQueue()).to(fanoutExchange());
    }
}

package com.lx.producer.service;

import jakarta.annotation.Resource;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import java.util.UUID;

@Service
public class OrderService {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void makeOrder(String userId, String productId, Integer numbers) {
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功：" + orderId);
        // 通过 MQ 完成消息的发送
        String exchangeName = "fanout_order_exchange";
        String routingKey = "";
        rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId);
    }
}

package com.lx.consumer.service;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(queues = {"duanxin.fanout.queue"})
public class FanoutDuanxinConsumer {
    @RabbitHandler
    public void receiveMessage(String message) {
        System.out.println("duanxin.fanout---接收到的信息是：" + message);
    }
}

package com.lx.consumer.service;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(queues = {"email.fanout.queue"})
public class FanoutEmailConsumer {
    @RabbitHandler
    public void receiveMessage(String message) {
        System.out.println("email.fanout---接收到的信息是：" + message);
    }
}

package com.lx.consumer.service;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(queues = {"sms.fanout.queue"})
public class FanoutSmsConsumer {
    @RabbitHandler
    public void receiveMessage(String message) {
        System.out.println("sms.fanout---接收到的信息是：" + message);
    }
}

package com.lx.producer.config;

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

@Configuration
public class RabbitMqConfiguration {
    // 创建一个 direct 类型的交换机
    @Bean
    public DirectExchange directExchangeExchange() {
        // 第一个参数是交换机的名字，第二个是是否持久化，第三个是是否自动删除
        return new DirectExchange("direct_order_exchange", true, false);
    }

    // 创建三个队列
    @Bean
    public Queue smsQueue() {
        return new Queue("sms.direct.queue", true);
    }

    @Bean
    public Queue duanxinQueue() {
        return new Queue("duanxin.direct.queue", true);
    }

    @Bean
    public Queue emailQueue() {
        return new Queue("email.direct.queue", true);
    }

    // 三个队列分别进行绑定关系
    @Bean
    public Binding smsBinding() {
        return BindingBuilder.bind(smsQueue()).to(directExchangeExchange()).with("sms");
    }

    @Bean
    public Binding duanxinBinding() {
        return BindingBuilder.bind(duanxinQueue()).to(directExchangeExchange()).with("duanxin");
    }

    @Bean
    public Binding emailBinding() {
        return BindingBuilder.bind(emailQueue()).to(directExchangeExchange()).with("email");
    }
}

package com.lx.producer.service;

import jakarta.annotation.Resource;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import java.util.UUID;

@Service
public class OrderService {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void makeOrder(String userId, String productId, Integer numbers) {
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功：" + orderId);
        // 通过 MQ 完成消息的发送
        String exchangeName = "direct_order_exchange";
        String routingKey1 = "sms";
        String routingKey2 = "duanxin";
        rabbitTemplate.convertAndSend(exchangeName, routingKey1, orderId);
        rabbitTemplate.convertAndSend(exchangeName, routingKey2, orderId);
    }
}

package com.lx.consumer.service;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "duanxin.topic.queue", durable = "true", autoDelete = "false"),
    exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC),
    key = "#.duanxin.#"
))
public class TopicDuanxinConsumer {
    @RabbitHandler
    public void receiveMessage(String message) {
        System.out.println("duanxin.topic---接收到的信息是：" + message);
    }
}

package com.lx.consumer.service;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "topic.email.queue", durable = "true", autoDelete = "false"),
    exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC),
    key = "*.email.#"
))
public class TopicEmailConsumer {
    @RabbitHandler
    public void receiveMessage(String message) {
        System.out.println("email.topic---接收到的信息是：" + message);
    }
}

package com.lx.consumer.service;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;

@Service
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "sms.topic.queue", durable = "true", autoDelete = "false"),
    exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC),
    key = "com.#"
))
public class TopicSmsConsumer {
    @RabbitHandler
    public void receiveMessage(String message) {
        System.out.println("sms.topic---接收到的信息是：" + message);
    }
}

package com.lx.producer.service;

import jakarta.annotation.Resource;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import java.util.UUID;

@Service
public class OrderService {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void makeOrder(String userId, String productId, Integer numbers) {
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功：" + orderId);
        // 通过 MQ 完成消息的发送
        String exchangeName = "topic_order_exchange";
        String routingKey1 = "com.email.duanxin";
        // String routingKey2 = "duanxin";
        rabbitTemplate.convertAndSend(exchangeName, routingKey1, orderId);
        // rabbitTemplate.convertAndSend(exchangeName, routingKey2, orderId);
    }
}

Spring Boot 整合 RabbitMQ 实战

Fanout 模式

1. 引入依赖

2. 在 yml/properties 文件进行相应的配置

3. 创建配置类（创建相应的交换机与队列，且绑定关系）

Spring Boot 整合 RabbitMQ 实战

Fanout 模式

1. 引入依赖

2. 在 yml/properties 文件进行相应的配置

3. 创建配置类（创建相应的交换机与队列，且绑定关系）

更多推荐文章

相关免费在线工具

4. 编写业务代码

Direct 模式

Topic 模式

更多推荐文章

相关免费在线工具

Spring Boot 整合 RabbitMQ 实战

Fanout 模式

1. 引入依赖

2. 在 yml/properties 文件进行相应的配置

3. 创建配置类（创建相应的交换机与队列，且绑定关系）

Spring Boot 整合 RabbitMQ 实战

Fanout 模式

1. 引入依赖

2. 在 yml/properties 文件进行相应的配置

3. 创建配置类（创建相应的交换机与队列，且绑定关系）

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

4. 编写业务代码

Direct 模式

Topic 模式

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具