AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

SpringBoot RabbitMq 六大模式

2023-06-19 144
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 依赖、配置依赖：<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId></dependency>

依赖、配置

依赖：

<dependency>

  <groupId>org.springframework.boot</groupId>

  <artifactId>spring-boot-starter-amqp</artifactId>

</dependency>


配置：


spring:

 rabbitmq:

   host: 192.168.31.10

   port: 5672 #通过控制台可以查看

   username: guest

   password: guest

   virtual-host: /vhost_sys_logs #可以不配置，会使用的是默认virtual-host

简单队列

模型

8a6426db2b60452fbf78504d69dccf5e.png

简单队列，consumer和producer通过队列直连。

代码示例

配置类：

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queue() {
        //name，名字；durable，是否开启持久化
        return new Queue("logs",false);
    }
}

生产者：

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest(classes=PrivilegeSystemMain.class)
public class RabbitMQTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void simpleTest() {
        rabbitTemplate.convertAndSend("logs","helo world!");
    }
}

消费者：

@Component
@Slf4j
public class ConsumeBean {
    @RabbitListener(queues={"logs"})
    public void getMsg(String message){
        log.info(message);
    }
}

工作队列

模型a7caab6b0ec84d77a6fc89868204584c.png

工作队列（work queue），让多个消费者去消费同一个消息队列中的消息，支持轮询分发（默认）、公平分发两种分发模式。

代码示例

配置类：

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queue() {
        //name，名字；durable，是否开启持久化
        return new Queue("logs",false);
    }
}


生产者：

@SpringBootTest(classes=PrivilegeSystemMain.class)
public class RabbitMQTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void simpleTest() {
        rabbitTemplate.convertAndSend("logs","helo world 01!");
        rabbitTemplate.convertAndSend("logs","helo world 02!");
    }
}

消费者：

@Component
@Slf4j
public class ConsumeBean {
    @RabbitListener(queues={"logs"})
    public void consumer_01(String message){
        log.info("consumer_01 get message "+message);
    }
    @RabbitListener(queues={"logs"})
    public void consumer_02(String message){
        log.info("consumer_02 get message "+message);
    }
}

订阅模式

模型

9786507cef9c4beeaf8ed021da2558c9.png

订阅模式（fanout），也叫广播模式，见名知意，其特点是将消息广播出去。通过交换机将生产者生产的消息分发到多个队列中去，从而支持生产者生产的一个消息被多个消费者消费。

代码示例

配置类：

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queue_01() {
        //name，名字；durable，是否开启持久化
        return new Queue("queue_01",false);
    }
    @Bean
    public Queue queue_02(){
        return new Queue("queue_02",false);
    }
    //订阅模式的交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout_exchange",false,false);
    }
    //将队列绑定到交换机上
    @Bean
    public Binding bindingSmsQueue_01(@Qualifier("queue_01") Queue logsAccess, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(logsAccess).to(fanoutExchange);
    }
    @Bean
    public Binding bindingSmsQueue_02(@Qualifier("queue_02") Queue logsError, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(logsError).to(fanoutExchange);
    }
}

生产者：

@SpringBootTest(classes=PrivilegeSystemMain.class)
public class RabbitMQTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void simpleTest() {
        rabbitTemplate.convertAndSend("fanout_exchange","","helo world 01!");
    }
}

消费者：

@Component
@Slf4j
public class ConsumeBean {
    @RabbitListener(queues={"queue_01"})
    public void consumer_01(String message){
        log.info("consumer_01 get message "+message);
    }
    @RabbitListener(queues={"queue_02"})
    public void consumer_02(String message){
        log.info("consumer_02 get message "+message);
    }
}

路由模式

模型

9b9825acb8b04fe98b03096752b91e93.png

路由模式（direct），在订阅模式支持一条消息被多个消费者消费的特性上增加了分类投递的特性，通过交换机，支持消息以类别（routing key）的方式投送到不同的消息队列中去。

代码示例

配置类：

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queue_01() {
        //durable，是否开启持久化
        return new Queue("queue_01",false);
    }
    @Bean
    public Queue queue_02(){
        return new Queue("queue_02",false);
    }
    //路由模式的交换机
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("direct_exchange",false,false);
    }
    //将队列绑定到交换机上
    @Bean
    public Binding bindingSmsQueue_01(@Qualifier("queue_01") Queue logsAccess, DirectExchange directExchange) {
        return BindingBuilder.bind(logsAccess).to(directExchange).with("routing_key_01");
    }
    @Bean
    public Binding bindingSmsQueue_02(@Qualifier("queue_02") Queue logsError, DirectExchange directExchange) {
        return BindingBuilder.bind(logsError).to(directExchange).with("routing_key_02");
    }
}

生产者：

@SpringBootTest(classes=PrivilegeSystemMain.class)
public class RabbitMQTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void simpleTest() {
        rabbitTemplate.convertAndSend("direct_exchange","routing_key_01","helo world!");
    }
}

消费者：

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class ConsumeBean {
    @RabbitListener(queues={"queue_01"})
    public void consumer_01(String message){
        log.info("consumer_01 get message "+message);
    }
    @RabbitListener(queues={"queue_02"})
    public void consumer_02(String message){
        log.info("consumer_02 get message "+message);
    }
}

主题模式

模型

7064eba2b47e40b09b7a29cec6f6e09d.png

主题模式，也叫通配符模式，在路由模式以类别进行消息投送的基础上增加了对通配符的支持，这样就可以使用通配符将多个类别聚合成一个主题。

代码示例

配置类：

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queue_01() {
        //durable，是否开启持久化
        return new Queue("queue_01",false);
    }
    @Bean
    public Queue queue_02(){
        return new Queue("queue_02",false);
    }
    //主题模式的交换机
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("topic_exchange",false,false);
    }
    //将队列绑定到交换机上
    @Bean
    public Binding bindingSmsQueue_01(@Qualifier("queue_01") Queue logsAccess, TopicExchange topicExchange) {
        return BindingBuilder.bind(logsAccess).to(topicExchange).with("#.error.#");
    }
    @Bean
    public Binding bindingSmsQueue_02(@Qualifier("queue_02") Queue logsError, TopicExchange topicExchange) {
        return BindingBuilder.bind(logsError).to(topicExchange).with("#.info.#");
    }
}

生产者：

@SpringBootTest(classes=PrivilegeSystemMain.class)
public class RabbitMQTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void simpleTest() {
        rabbitTemplate.convertAndSend("topic_exchange","test.error.test","helo world!");
    }
}

消费者：

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class ConsumeBean {
    @RabbitListener(queues={"queue_01"})
    public void consumer_01(String message){
        log.info("consumer_01 get message "+message);
    }
    @RabbitListener(queues={"queue_02"})
    public void consumer_02(String message){
        log.info("consumer_02 get message "+message);
    }
}

RPC

几乎不会使用，略......

文章标签：
云消息队列 MQ
Java
消息中间件
Spring
关键词：
springboot云消息队列 MQ
Spring Boot模式
云消息队列 MQ模式
Spring boot rabbitmq
云消息队列 MQ springboot
相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
_BugMan
目录
相关文章
varin
|
2月前
|
消息中间件 Ubuntu Java
SpringBoot整合MQTT实战：基于EMQX实现双向设备通信
本教程指导在Ubuntu上部署EMQX 5.9.0并集成Spring Boot实现MQTT双向通信，涵盖服务器搭建、客户端配置及生产实践，助您快速构建企业级物联网消息系统。
varin
1155 1
HELLO程序员
|
2月前
|
消息中间件 Java Kafka
消息队列比较：Spring 微服务中的 Kafka 与 RabbitMQ
本文深入解析了 Kafka 和 RabbitMQ 两大主流消息队列在 Spring 微服务中的应用与对比。内容涵盖消息队列的基本原理、Kafka 与 RabbitMQ 的核心概念、各自优势及典型用例，并结合 Spring 生态的集成方式，帮助开发者根据实际需求选择合适的消息中间件，提升系统解耦、可扩展性与可靠性。
HELLO程序员
221 1
消息队列比较：Spring 微服务中的 Kafka 与 RabbitMQ
HELLO程序员
|
2月前
|
消息中间件 存储 Java
RabbitMQ 和 Spring Cloud Stream 实现异步通信
本文介绍了在微服务架构中，如何利用 RabbitMQ 作为消息代理，并结合 Spring Cloud Stream 实现高效的异步通信。内容涵盖异步通信的优势、RabbitMQ 的核心概念与特性、Spring Cloud Stream 的功能及其与 RabbitMQ 的集成方式。通过这种组合，开发者可以构建出具备高可用性、可扩展性和弹性的分布式系统，满足现代应用对快速响应和可靠消息传递的需求。
HELLO程序员
194 2
RabbitMQ 和 Spring Cloud Stream 实现异步通信
蓝易云
|
6月前
|
消息中间件 缓存 NoSQL
基于Spring Data Redis与RabbitMQ实现字符串缓存和计数功能（数据同步）
总的来说，借助Spring Data Redis和RabbitMQ，我们可以轻松实现字符串缓存和计数的功能。而关键的部分不过是一些"厨房的套路"，一旦你掌握了这些套路，那么你就像厨师一样可以准备出一道道饕餮美食了。通过这种方式促进数据处理效率无疑将大大提高我们的生产力。
蓝易云
239 32
啦啦啦191
|
5月前
|
监控 安全 Java
Java 开发中基于 Spring Boot 3.2 框架集成 MQTT 5.0 协议实现消息推送与订阅功能的技术方案解析
本文介绍基于Spring Boot 3.2集成MQTT 5.0的消息推送与订阅技术方案，涵盖核心技术栈选型（Spring Boot、Eclipse Paho、HiveMQ）、项目搭建与配置、消息发布与订阅服务实现，以及在智能家居控制系统中的应用实例。同时，详细探讨了安全增强（TLS/SSL）、性能优化（异步处理与背压控制）、测试监控及生产环境部署方案，为构建高可用、高性能的消息通信系统提供全面指导。附资源下载链接：[https://pan.quark.cn/s/14fcf913bae6](https://pan.quark.cn/s/14fcf913bae6)。
啦啦啦191
949 0
小王老师呀
|
消息中间件
RabbitMQ的 RPC 消息模式你会了吗？
【9月更文挑战第11天】RabbitMQ 的 RPC（远程过程调用）消息模式允许客户端向服务器发送请求并接收响应。其基本原理包括：1) 客户端发送请求，创建回调队列并设置关联标识符；2) 服务器接收请求并发送响应至回调队列；3) 客户端根据关联标识符接收并匹配响应。实现步骤涵盖客户端和服务器的连接、信道创建及请求处理。注意事项包括关联标识符唯一性、回调队列管理、错误处理及性能考虑。RPC 模式适用于构建可靠的分布式应用程序，但需根据需求调整优化。
小王老师呀
385 3
别惹CC
|
10月前
|
负载均衡 IDE Java
SpringBoot整合XXL-JOB【04】- 以GLUE模式运行与执行器负载均衡策略
在本节中，我们将介绍XXL-JOB的GLUE模式和集群模式下的路由策略。GLUE模式允许直接在线上改造方法为定时任务，无需重新部署。通过一个测试方法，展示了如何在调度中心配置并使用GLUE模式执行定时任务。接着，我们探讨了多实例环境下的负载均衡策略，确保任务不会重复执行，并可通过修改路由策略（如轮训）实现任务在多个实例间的均衡分配。最后，总结了GLUE模式和负载均衡策略的应用，帮助读者更深入理解XXL-JOB的使用。
别惹CC
546 9
SpringBoot整合XXL-JOB【04】- 以GLUE模式运行与执行器负载均衡策略
遇见明天
|
11月前
|
消息中间件 网络协议 RocketMQ
RocketMQ Controller 模式 始终更新成本机ip
ontrollerAddr=192.168.24.241:8878 但是日志输出Update controller leader address to 127.0.0.1:8878。导致访问失败
遇见明天
232 3

热门文章

最新文章

  • 1
    基于springboot的宠物领养系统
  • 2
    基于springboot的大学生兼职系统
  • 3
    SpringBoot启动时连接Redis报错：ERR This instance has cluster support disabled - 如何解决？
  • 4
    Spring AI Alibaba Admin 开源！以数据为中心的 Agent 开发平台
  • 5
    基于 Spring AI Alibaba + Nacos 的分布式 Multi-Agent 构建指南
  • 6
    2025 OSCAR丨与创新者同频！Apache RocketMQ 邀您共赴开源之约
  • 7
    SpringBoot 整合 AMQP（RabbitMQ）
  • 8
    消息中间件选型分析——从Kafka与RabbitMQ的对比来看全局
  • 9
    RabbitMQ Windows 安装、配置、使用 - 小白教程
  • 10
    SpringBoot里rabbitMq设置不自动检查连接
  • 1
    Docker环境下快速部署RabbitMQ教程。
    344
  • 2
    一文带你从入门到实战全面掌握RocketMQ核心概念、架构部署、实践应用和高级特性
    4315
  • 3
    （手把手）在华为云、阿里云搭建自己的物联网MQTT消息服务器，免费IOT平台
    2445
  • 4
    如何在腾讯云等平台搭建自己的物联网MQTT服务器Broker
    550
  • 5
    RocketMQ主题路由信息缺失问题分析
    252
  • 6
    基于Spring Data Redis与RabbitMQ实现字符串缓存和计数功能（数据同步）
    239
  • 7
    美团面试：对比分析 RocketMQ、Kafka、RabbitMQ 三大MQ常见问题？
    1874
  • 8
    RocketMQ实战—10.营销系统代码优化
    223
  • 9
    RocketMQ实战—9.营销系统代码初版
    247
  • 10
    RocketMQ实战—8.营销系统业务和方案介绍
    274

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
    • 下一篇
    阿里云弹性公网IP线路类型【BGP(多线)_精品】是什么意思？