本项目基于Spring boot的AMQP模块,整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ

本文主要是介绍本项目基于Spring boot的AMQP模块,整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

  • 在业务逻辑的异步处理,系统解耦,分布式通信以及控制高并发的场景下,消息队列有着广泛的应用。本项目基于Spring的AMQP模块,整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ添加和读取消息的功能。并比较了两种模式:生产者-消费者模式和发布-订阅模式的区别。AMQP作为比JMS更加高级的消息协议,支持更多的消息路由和消息模式。

  • 包含的特性如下:

输入图片说明

  1. 如上图,生产者消费者模型:添加了一个队列,并创建了两个消费者用于监听队列消息,我们发现,当有消息到达时,两个消费者会交替收到消息。这一过程虽然不用创建交换机,但会使用默认的交换机,并用默认的直连(default-direct)策略连接队列;

下面来看具体真使用:

1.安装,关于安装之前已经介绍过了,有兴趣的话,可以参考之前的文章:

PHP安装rabbitmq扩展_php 按照rabbitmq 扩展-CSDN博客

不过使用的是php版本的,原理上是一样的。

安装好之后我们打开后台:

 1.yml文件配置rabbitmq:

server.port=8081
logging.path=D:/
logging.level.boot.spring=DEBUG
spring.application.name=spring-boot-rabbitmq
spring.rabbitmq.host=192.168.33.10
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=test123

pom文件添加依赖:

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

生产者:

@Transactional
@Service("producer")
public class ProducerImpl implements Producer{@AutowiredRabbitTemplate rabbitTemplate;public void sendMail(String queue,Mail mail) {rabbitTemplate.setQueue(queue);rabbitTemplate.convertAndSend(queue,mail);}}

 消费者:

@Configuration
public class ProducerConsumerConfig {@Beanpublic Queue myQueue() {Queue queue=new Queue("myqueue");return queue;}}

将队列绑定到交换机:

@Configuration
public class DirectExchangeConfig {@Beanpublic DirectExchange directExchange(){DirectExchange directExchange=new DirectExchange("direct");return directExchange;}@Beanpublic Queue directQueue1() {Queue queue=new Queue("directqueue1");return queue;}//将交换机和相应队列连起来@Beanpublic Binding bindingorange(){Binding binding=BindingBuilder.bind(directQueue1()).to(directExchange()).with("orange");return binding;}
}

写个控制器:

@Controller
public class RabbitMQController {@AutowiredProducerImpl producer;@AutowiredPublisherImpl publisher;@RequestMapping(value="/produce",produces = {"application/json;charset=UTF-8"})@ResponseBodypublic void produce(@ModelAttribute("mail")Mail mail) throws Exception{producer.sendMail("myqueue",mail);}}

 通过监听器来监听消费:

@Component
public class DirectListener1 {@RabbitListener(queues = "directqueue1")public void displayMail(Mail mail) throws Exception {System.out.println("directqueue1队列监听器1号收到消息"+mail.toString());}
}

启动项目:我们来调用一下:

 看到队列的消费情况:

上篇文章介绍 rocketmq,这2种消息队列在实际应用中都等起道会很好的作用,但在实际应中,rocketmq使用java语言开发的,跨平台比较好。rabbitmq 架构设计简单,在一般的轻量级的服务用用中,可能会比较好些。官方的对比也给了说明:

适用于易用性和灵活性要求较高的场景

1、应用场景

  • 异步任务处理:RabbitMQ提供可靠的消息传递机制,适用于处理异步任务,例如将耗时的任务放入消息队列中,然后由消费者异步处理,提高系统的响应速度和可伸缩性。
  • 解耦系统组件:通过使用RabbitMQ作为消息中间件,不同的系统组件可以通过消息进行解耦,实现松耦合的架构,提高系统的可维护性和灵活性。
  • 事件驱动架构:RabbitMQ的发布-订阅模式可以用于构建事件驱动架构,将系统中的事件作为消息发布到相应的主题,不同的消费者可以订阅感兴趣的主题进行相应的处理。

2.RocketMQ:

适用于大规模数据处理和高吞吐量的场景

  • 分布式事务:RocketMQ支持分布式事务消息,适用于涉及多个业务系统的分布式事务场景,确保消息的一致性和可靠性,同时提供高吞吐量的消息传递能力。
  • 实时日志处理:由于RocketMQ具备高吞吐量和低延迟的特点,可以用于实时日志处理,例如日志收集和分析、日志聚合等场景。

2、服务架构

  • RabbitMQ

  1. Channel(通道):Channel是RabbitMQ与应用程序之间的虚拟连接。通过在物理连接(connection)上创建多个通道,应用程序可以并发地进行消息传递操作。通道负责发送和接收消息,并执行一些与消息相关的操作,如声明队列、定义交换机和绑定等。通道可以看作是轻量级的会话,通过一个物理连接与RabbitMQ进行交互。
  2. Exchange(交换机):交换机是消息的接收和转发中心。当消息发送到RabbitMQ时,会通过交换机进行路由。交换机根据其类型和绑定规则,将消息路由到一个或多个队列中。常见的交换机类型包括直连交换机(direct)、主题交换机(topic)、扇形交换机(fanout)和头部交换机(headers)。
  3. Queue(队列):队列是RabbitMQ用于存储消息的缓冲区。当消息无法立即路由到消费者时,会被存储在队列中,等待消费者来获取和处理。每个队列都有一个唯一的名称,并且按照FIFO(先进先出)的顺序进行消息的投递和消费。
  4. Virtual Host(虚拟主机):虚拟主机是逻辑上的隔离环境,用于将RabbitMQ服务器划分为多个独立的部分。每个虚拟主机都有自己的交换机、队列、绑定和权限设置。虚拟主机可以帮助不同应用程序或服务之间进行隔离,并提供安全性和资源管理的控制。
  5. Broker(代理):Broker是RabbitMQ消息队列服务器的实例,负责接收、存储和路由消息。它充当中间人的角色,将生产者发送的消息传递给消费者。一个RabbitMQ实例可以包含多个Broker,每个Broker可以承载多个虚拟主机和队列。

  • RocketMQ

  1. NameServer(命名服务器):NameServer是RocketMQ的命名服务组件,用于管理和提供Broker的路由信息。它充当元数据的中心,负责维护Broker的注册信息、Topic的路由信息等。Producer和Consumer在发送和接收消息之前,需要与NameServer进行交互以获取正确的Broker信息。
  2. Controller(控制器):Controller是RocketMQ的控制器组件,负责协调和管理整个RocketMQ集群的工作。它监控Broker的状态变化,处理集群的扩容和缩容,进行负载均衡等操作。Controller是RocketMQ集群的核心组件之一,确保集群的可靠运行和自动化管理。
  3. Broker(代理):Broker是RocketMQ的消息存储和处理节点。它负责接收来自Producer的消息,并将其存储在磁盘上。Broker还负责处理Consumer的消息拉取请求,并将消息推送给Consumer进行消费。一个RocketMQ集群可以包含多个Broker,每个Broker负责存储一部分Topic的消息数据。

这篇关于本项目基于Spring boot的AMQP模块,整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/451121

相关文章

Java实现检查多个时间段是否有重合

《Java实现检查多个时间段是否有重合》这篇文章主要为大家详细介绍了如何使用Java实现检查多个时间段是否有重合,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录流程概述步骤详解China编程步骤1:定义时间段类步骤2:添加时间段步骤3:检查时间段是否有重合步骤4:输出结果示例代码结语作

Java中String字符串使用避坑指南

《Java中String字符串使用避坑指南》Java中的String字符串是我们日常编程中用得最多的类之一,看似简单的String使用,却隐藏着不少“坑”,如果不注意,可能会导致性能问题、意外的错误容... 目录8个避坑点如下:1. 字符串的不可变性:每次修改都创建新对象2. 使用 == 比较字符串,陷阱满

Java判断多个时间段是否重合的方法小结

《Java判断多个时间段是否重合的方法小结》这篇文章主要为大家详细介绍了Java中判断多个时间段是否重合的方法,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录判断多个时间段是否有间隔判断时间段集合是否与某时间段重合判断多个时间段是否有间隔实体类内容public class D

使用C++实现链表元素的反转

《使用C++实现链表元素的反转》反转链表是链表操作中一个经典的问题,也是面试中常见的考题,本文将从思路到实现一步步地讲解如何实现链表的反转,帮助初学者理解这一操作,我们将使用C++代码演示具体实现,同... 目录问题定义思路分析代码实现带头节点的链表代码讲解其他实现方式时间和空间复杂度分析总结问题定义给定

IDEA编译报错“java: 常量字符串过长”的原因及解决方法

《IDEA编译报错“java:常量字符串过长”的原因及解决方法》今天在开发过程中,由于尝试将一个文件的Base64字符串设置为常量,结果导致IDEA编译的时候出现了如下报错java:常量字符串过长,... 目录一、问题描述二、问题原因2.1 理论角度2.2 源码角度三、解决方案解决方案①:StringBui

Java覆盖第三方jar包中的某一个类的实现方法

《Java覆盖第三方jar包中的某一个类的实现方法》在我们日常的开发中,经常需要使用第三方的jar包,有时候我们会发现第三方的jar包中的某一个类有问题,或者我们需要定制化修改其中的逻辑,那么应该如何... 目录一、需求描述二、示例描述三、操作步骤四、验证结果五、实现原理一、需求描述需求描述如下:需要在

Java中ArrayList和LinkedList有什么区别举例详解

《Java中ArrayList和LinkedList有什么区别举例详解》:本文主要介绍Java中ArrayList和LinkedList区别的相关资料,包括数据结构特性、核心操作性能、内存与GC影... 目录一、底层数据结构二、核心操作性能对比三、内存与 GC 影响四、扩容机制五、线程安全与并发方案六、工程

部署Vue项目到服务器后404错误的原因及解决方案

《部署Vue项目到服务器后404错误的原因及解决方案》文章介绍了Vue项目部署步骤以及404错误的解决方案,部署步骤包括构建项目、上传文件、配置Web服务器、重启Nginx和访问域名,404错误通常是... 目录一、vue项目部署步骤二、404错误原因及解决方案错误场景原因分析解决方案一、Vue项目部署步骤

JavaScript中的reduce方法执行过程、使用场景及进阶用法

《JavaScript中的reduce方法执行过程、使用场景及进阶用法》:本文主要介绍JavaScript中的reduce方法执行过程、使用场景及进阶用法的相关资料,reduce是JavaScri... 目录1. 什么是reduce2. reduce语法2.1 语法2.2 参数说明3. reduce执行过程

如何使用Java实现请求deepseek

《如何使用Java实现请求deepseek》这篇文章主要为大家详细介绍了如何使用Java实现请求deepseek功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1.deepseek的api创建2.Java实现请求deepseek2.1 pom文件2.2 json转化文件2.2