SpringBoot 集成 RabbitMQ 常见消息模式示例 | 极客日志

Javajava

SpringBoot 集成 RabbitMQ 常见消息模式示例

本文介绍在 SpringBoot 项目中集成 RabbitMQ 的三种典型消息模式：工作队列（Work Queue）、发布订阅（Fanout）和路由模式（Direct）。内容包括依赖引入、配置说明、队列与交换机声明、消息发送及监听实现。通过 Bean 定义管理队列生命周期，结合 RabbitTemplate 和@RabbitListener 注解完成消息收发，适用于分布式系统中的异步通信场景。

DebugKing发布于 2026/3/270 浏览

SpringBoot 集成 RabbitMQ 常见消息模式示例

一、公共流程

1. 引入依赖

对于 SpringBoot 项目，首先需要引入 starter 依赖。

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

2. 配置 application.yml 参数

以 Docker 启动的 RabbitMQ 服务为例，映射到本地。

spring:
  application:
    name: spring-rabbitmq-demo
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: admin
    virtual-host: ran

二、Work Queue（工作队列模式）

1. 声明队列

将队列的声明交给 Spring 管理，使用 Bean 修饰。

package com.ran.jh.rabbitmq.config;

 com.ran.jh.rabbitmq.constant.Constants;
 org.springframework.amqp.core.Queue;
 org.springframework.amqp.core.QueueBuilder;
 org.springframework.context.annotation.Bean;
 org.springframework.context.annotation.Configuration;


   {
    
    
     Queue  {
         QueueBuilder.durable(Constants.WORK_QUEUE).build();
    }
}

极客日志微信公众号二维码

更多推荐文章

相关免费在线工具

Keycode 信息
查找任何按下的键的javascript键代码、代码、位置和修饰符。在线工具，Keycode 信息在线工具，online
Escape 与 Native 编解码
JavaScript 字符串转义/反转义；Java 风格 \uXXXX（Native2Ascii）编码与解码。在线工具，Escape 与 Native 编解码在线工具，online
JavaScript / HTML 格式化
使用 Prettier 在浏览器内格式化 JavaScript 或 HTML 片段。在线工具，JavaScript / HTML 格式化在线工具，online
JavaScript 压缩与混淆
Terser 压缩、变量名混淆，或 javascript-obfuscator 高强度混淆（体积会增大）。在线工具，JavaScript 压缩与混淆在线工具，online
Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。在线工具，Base64 字符串编码/解码在线工具，online
Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。在线工具，Base64 文件转换器在线工具，online

package com.ran.jh.rabbitmq.controller;

import com.ran.jh.rabbitmq.constant.Constants;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RequestMapping("/producer")
@RestController
public class ProducerController {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 获取模版
    @RequestMapping("/work")
    public String work() {
        for (int i = 0; i < 5; i++) {
            rabbitTemplate.convertAndSend("", Constants.WORK_QUEUE, "hello, spring amqp 工作队列模式...");
        }
        return "发送成功";
    }
}

package com.ran.jh.rabbitmq.listener;

import com.rabbitmq.client.Channel;
import com.ran.jh.rabbitmq.constant.Constants;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class WorkListeners {
    @RabbitListener(queues = Constants.WORK_QUEUE)
    public void RabbitListener1(Message message, Channel channel) {
        System.out.println("Listener1, 消息所有参数 message: " + message + " , 通道 Channel: " + channel);
    }

    @RabbitListener(queues = Constants.WORK_QUEUE)
    public void RabbitListener2(String msg) {
        System.out.println("Listener2, 消息主体内容 msg: " + msg);
    }
}

// 声明订阅模式队列
@Bean("fanoutQueue1")
public Queue fanoutQueue1() {
    return QueueBuilder.durable(Constants.FANOUT_QUEUE1).build();
}

@Bean("fanoutQueue2")
public Queue fanoutQueue2() {
    return QueueBuilder.durable(Constants.FANOUT_QUEUE2).build();
}

// 声明交换机
@Bean("fanoutExchange")
public FanoutExchange fanoutExchange() {
    return ExchangeBuilder.fanoutExchange(Constants.FANOUT_EXCHANGE).durable(true).build();
}

// 声明绑定关系
@Bean("fanoutQueueBindExchange1")
public Binding fanoutQueueBindExchange1(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,
                                        @Qualifier("fanoutQueue1") Queue queue) {
    return BindingBuilder.bind(queue).to(fanoutExchange);
}

@Bean("fanoutQueueBindExchange2")
public Binding fanoutQueueBindExchange2(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,
                                        @Qualifier("fanoutQueue2") Queue queue) {
    return BindingBuilder.bind(queue).to(fanoutExchange);
}

@RequestMapping("/fanout")
public String fanout() {
    for (int i = 0; i < 5; i++) {
        rabbitTemplate.convertAndSend(Constants.FANOUT_EXCHANGE, "", "hello, spring amqp 发布订阅模式..." + i);
    }
    return "发送成功";
}

package com.ran.jh.rabbitmq.listener;

import com.ran.jh.rabbitmq.constant.Constants;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class FanoutListeners {
    @RabbitListener(queues = Constants.FANOUT_QUEUE1)
    public void RabbitListener1(String msg) {
        System.out.println("Listener1," + "队列 [" + Constants.FANOUT_QUEUE1 + "] 消息主体内容 msg: " + msg);
    }

    @RabbitListener(queues = Constants.FANOUT_QUEUE2)
    public void RabbitListener2(String msg) {
        System.out.println("Listener2," + "队列 [" + Constants.FANOUT_QUEUE2 + "] 消息主体内容 msg: " + msg);
    }
}

@Bean("directQueue1")
public Queue directQueue1() {
    return QueueBuilder.durable(Constants.DIRECT_QUEUE1).build();
}

@Bean("directQueue2")
public Queue directQueue2() {
    return QueueBuilder.durable(Constants.DIRECT_QUEUE2).build();
}

// 声明交换机
@Bean("directExchange")
public DirectExchange directExchange() {
    return ExchangeBuilder.directExchange(Constants.DIRECT_EXCHANGE).durable(true).build();
}

// 声明绑定关系
@Bean("directQueueBindExchange1")
public Binding directQueueBindExchange1(@Qualifier("directExchange") DirectExchange directExchange,
                                        @Qualifier("directQueue1") Queue queue) {
    return BindingBuilder.bind(queue).to(directExchange).with("orange");
}

@Bean("directQueueBindExchange2")
public Binding directQueueBindExchange2(@Qualifier("directExchange") DirectExchange directExchange,
                                        @Qualifier("directQueue2") Queue queue) {
    return BindingBuilder.bind(queue).to(directExchange).with("orange");
}

@Bean("directQueueBindExchange3")
public Binding directQueueBindExchange3(@Qualifier("directExchange") DirectExchange directExchange,
                                        @Qualifier("directQueue2") Queue queue) {
    return BindingBuilder.bind(queue).to(directExchange).with("black");
}

// 路由模式
@RequestMapping("/direct/{routingKey}")
public String direct(@PathVariable("routingKey") String routingKey) {
    rabbitTemplate.convertAndSend(Constants.DIRECT_EXCHANGE, routingKey, "hello, spring amqp 路由模式..." + " ,routingKey: " + routingKey);
    return "发送成功，routingKey: " + routingKey;
}

package com.ran.jh.rabbitmq.listener;

import com.ran.jh.rabbitmq.constant.Constants;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class DirectListeners {
    @RabbitListener(queues = Constants.DIRECT_QUEUE1)
    public void RabbitListener1(String msg) {
        System.out.println("Listener1," + "队列 [" + Constants.DIRECT_QUEUE1 + "] 消息主体内容 msg: " + msg);
    }

    @RabbitListener(queues = Constants.DIRECT_QUEUE2)
    public void RabbitListener2(String msg) {
        System.out.println("Listener2," + "队列 [" + Constants.DIRECT_QUEUE2 + "] 消息主体内容 msg: " + msg);
    }
}