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

SpringBoot RabbitMq 六大模式

2023-06-19 114
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 依赖、配置依赖：<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
关键词：
Spring Boot模式
云消息队列 MQ模式
Spring boot rabbitmq
springboot云消息队列 MQ
云消息队列 MQ springboot
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
_BugMan
目录
相关文章
别惹CC
|
2月前
|
负载均衡 IDE Java
SpringBoot整合XXL-JOB【04】- 以GLUE模式运行与执行器负载均衡策略
在本节中，我们将介绍XXL-JOB的GLUE模式和集群模式下的路由策略。GLUE模式允许直接在线上改造方法为定时任务，无需重新部署。通过一个测试方法，展示了如何在调度中心配置并使用GLUE模式执行定时任务。接着，我们探讨了多实例环境下的负载均衡策略，确保任务不会重复执行，并可通过修改路由策略（如轮训）实现任务在多个实例间的均衡分配。最后，总结了GLUE模式和负载均衡策略的应用，帮助读者更深入理解XXL-JOB的使用。
别惹CC
70 9
SpringBoot整合XXL-JOB【04】- 以GLUE模式运行与执行器负载均衡策略
小王老师呀
|
6月前
|
消息中间件
RabbitMQ的 RPC 消息模式你会了吗？
【9月更文挑战第11天】RabbitMQ 的 RPC（远程过程调用）消息模式允许客户端向服务器发送请求并接收响应。其基本原理包括：1) 客户端发送请求，创建回调队列并设置关联标识符；2) 服务器接收请求并发送响应至回调队列；3) 客户端根据关联标识符接收并匹配响应。实现步骤涵盖客户端和服务器的连接、信道创建及请求处理。注意事项包括关联标识符唯一性、回调队列管理、错误处理及性能考虑。RPC 模式适用于构建可靠的分布式应用程序，但需根据需求调整优化。
小王老师呀
99 2
如夜了我衣衫太薄便归家靠路灯
|
7月前
|
消息中间件 Java 网络架构
springboot 整合rabbitmq，实现发送提醒消息
springboot 整合rabbitmq，实现发送提醒消息
如夜了我衣衫太薄便归家靠路灯
371 4
遇见明天
|
3月前
|
消息中间件 网络协议 RocketMQ
RocketMQ Controller 模式 始终更新成本机ip
ontrollerAddr=192.168.24.241:8878 但是日志输出Update controller leader address to 127.0.0.1:8878。导致访问失败
遇见明天
75 3
一位隐者
|
3月前
|
消息中间件 监控 Java
如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
一位隐者
53 6
热爱技术的小郑
|
7月前
|
消息中间件 Java 测试技术
消息中间件RabbitMQ---SpringBoot整合RabbitMQ【三】
这篇文章是关于如何在SpringBoot应用中整合RabbitMQ的消息中间件。内容包括了在SpringBoot项目中添加RabbitMQ的依赖、配置文件设置、启动类注解，以及如何通过单元测试来创建交换器、队列、绑定，并发送和接收消息。文章还介绍了如何配置消息转换器以支持对象的序列化和反序列化，以及如何使用注解`@RabbitListener`来接收消息。
热爱技术的小郑
161 2
消息中间件RabbitMQ---SpringBoot整合RabbitMQ【三】
土木林森
|
7月前
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配：掌握消息路由的艺术！
【8月更文挑战第24天】在消息队列（MQ）体系中，交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器，它通过模式匹配实现消息的精准分发，适用于发布-订阅模式。不同于直接交换器和扇形交换器，`topic`交换器支持更复杂的路由策略，通过带有通配符（如 * 和 #）的模式字符串来定义队列与交换器间的绑定关系。
土木林森
111 2
VipSoft
|
7月前
|
网络协议 Java 物联网
MQTT(EMQX) - SpringBoot 整合MQTT 连接池 Demo - 附源代码 + 在线客服聊天架构图
MQTT(EMQX) - SpringBoot 整合MQTT 连接池 Demo - 附源代码 + 在线客服聊天架构图
VipSoft
1229 3
如夜了我衣衫太薄便归家靠路灯
|
7月前
|
消息中间件 Java Maven
springboot 整合rocketMQ，实现发送提醒消息
springboot 整合rocketMQ，实现发送提醒消息
如夜了我衣衫太薄便归家靠路灯
185 4
秋夜雨巷
|
7月前
|
消息中间件
RabbitMQ广播模式
RabbitMQ广播模式
秋夜雨巷
105 1

热门文章

最新文章

  • 1
    Springboot使用Redis实现分布式锁
  • 2
    SpringBoot-打包&部署
  • 3
    Springboot静态资源映射及文件映射
  • 4
    SpringBoot集成Tomcat、DispatcherServlet
  • 5
    SpringBoot 2.0 多图片上传加回显
  • 6
    基于SpringBoot的智慧停车场微信小程序源码分享
  • 7
    springBoot集成websocket实时消息推送
  • 8
    Spring 集成 DeepSeek 的 3大方法（史上最全）
  • 9
    SpringBoot 通过集成 Flink CDC 来实时追踪 MySql 数据变动
  • 10
    支持 40+ 插件，Spring AI Alibaba 简化智能体私有数据集成
  • 1
    RabbitMQ如何保证消息可靠性？
    45
  • 2
    【有奖实践】轻量消息队列（原 MNS）订阅 OSS 事件实时处理文件变动
    23
  • 3
    Docker部署RocketMQ5.2.0集群
    197
  • 4
    轻量消息队列（原 MNS）订阅 OSS 事件实践
    40
  • 5
    嵌入式C++、STM32、MySQL、GPS、InfluxDB和MQTT协议数据可视化
    144
  • 6
    一个绿色的免费MQTT调试工具，实测连接移动OneNET
    155
  • 7
    乐刻运动：基于 RocketMQ + MQTT 实现健身产业数字化升级
    78
  • 8
    【RabbitMQ】-SpringAMQP以及Work模型
    55
  • 9
    招行面试：10Wqps场景，RocketMQ 顺序消费 的性能 如何提升 ？
    82
  • 10
    云端问道16期方案教学-高弹性低成本的云消息队列RabbitMQ版
    32

    • 相关课程

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

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 快速上手使用Dubbo进行远程调用
  • 部署RabbitMQ
  • Spring-cloud-bus-rocketmq入门与实践
  • Spring-cloud-stream-binder-rocketmq入门与实践
  • Rocketmq-spring入门与实践
  • RocketMQ一站式入门使用
    • 下一篇
    通义万相：视觉生成大模型再进化