RabbitMq使用与整合

MQ基本概念

MQ概述

MQ全称 Message Queue（[kjuː]）（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。

（队列是一种容器，用于存放数据的都是容器，存放消息的就是消息队列）

分布式系统的调用：

方式一：直接调用

方式二：间接调用

A将数据存放到中间一个系统，通过中间的系统发送到B

中间系统可以成为中间件MQ

MQ是用于存放消息的中间件

被调用者叫生产者 调用者是消费者

MQ的优势和劣势

优势

应用解耦：提高系统容错性和可维护性。

异步提速：提升用户体验和系统吞吐量。

削峰填谷：提高系统稳定性。

应用解耦

系统的耦合性越高，容错性就越低，可维护性就越低。

使用 MQ 使得应用间解耦，提升容错性和可维护性

异步提速

提升用户体验和系统吞吐量（单位时间内处理请求的数目）。

削峰填谷（削峰）

使用了 MQ 之后，限制消费消息的速度为1000，这样一来，高峰期产生的数据势必会被积压在 MQ 中，高峰就被“削”掉了，但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000，直到消费完积压的消息，这就叫做“填谷”。

使用MQ后，可以提高系统稳定性。

劣势

系统可用性降低

系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

系统复杂度提高

MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

一致性问题

A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理失败。如何保证消息数据处理的一致性？

既然 MQ 有优势也有劣势，那么使用 MQ 需要满足什么条件呢？

消费者-->生产者

1.生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。

2.容许短暂的不一致性。

3.确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。

RabbitMQ基本介绍

2007年，Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。Erlang语言由Ericson设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。

RabbitMQ 基础架构

Broker 中间者 服务

procedure 和consumer都是客户端

客户端通过链接和服务端进行通信 所以需要建立起来连接 然后进行通信a

使用channel（管道）节省资源

一个rabbitmq里面有很多的虚拟机 相当于mysql里面有很多数据库，数据库里面有很多表，都是独立的。

每个虚拟机里面有很多的exchange和queue 独立分区的作用

 RabbitMQ 中的相关概念

Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker。

Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等。

Connection：publisher／consumer 和 broker 之间的 TCP 连接。

Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销。

Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：

        direct (point-to-point)

        topic (publish-subscribe)

        fanout (multicast)

Queue：消息最终被送到这里等待 consumer 取走

Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

RabbitMQ的6 种工作模式

RabbitMQ 提供了 6 种工作模式：

简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing 路由模式、Topics 主题模式、RPC 远程调用模式（远程调用，不太算 MQ；暂不作介绍）。

官网对应模式介绍：RabbitMQ Tutorials — RabbitMQ

RabbitMQ的安装和配置

安装依赖环境

在线安装依赖环境：

yum -y install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz

安装Erlang

rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm

安装RabbitMQ

#安装依赖的包

rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm

#安装rabbitmq

rpm -ivh rabbitmq-server-3.7.18-1.el7.noarch.rpm

rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm socat-1.7.3.2-2.el7.x86_64.rpm rabbitmq-server-3.7.18-1.el7.noarch.rpm

启动RabbitMQ

systemctl start rabbitmq-server # 启动服务

systemctl stop rabbitmq-server # 停止服务

systemctl restart rabbitmq-server # 重启服务

systemctl status rabbitmq-server # 查看状态

开启管理界面及配置

# 开启管理界面

rabbitmq-plugins enable rabbitmq_management

# 修改默认配置信息

vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.18/ebin/rabbit.app

# 比如修改密码、配置等等，例如：loopback_users 中的 <<"guest">>,只保留guest

修改之后重启一下rabbitmq

入门实例

1.添加虚拟主机

2.添加用户

3.重新设置权限

点击虚拟主机设置权限

4.idea连接

项目结构搭建

mq 导入依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version>
</dependency>

生产者

public static void main( String[] args ) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setUsername("root");connectionFactory.setPassword("root");connectionFactory.setHost("192.168.229.16");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/test");
//        建立连接Connection connection = connectionFactory.newConnection();
//        管道Channel channel = connection.createChannel();
//        创建队列/** String queue 队列的名称* boolean durable 持久化* boolean exclusive 是否独占* boolean autoDelete 是否自动删除* Map<String,Object> arguments 参数* */channel.queueDeclare("test01",false,false,false,null);
//        发布消息channel.basicPublish("","test01",null,"第一次发送".getBytes());}

消费者

{public static void main( String[] args ) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setUsername("root");connectionFactory.setPassword("root");connectionFactory.setHost("192.168.229.16");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/test");
//        建立连接Connection connection = connectionFactory.newConnection();
//        管道Channel channel = connection.createChannel();
//        消费信息Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String s = new String(body);System.out.println(s);}};channel.basicConsume("test01",true,consumer);}

同时消费后消息组内为 0

RabbitMQ工作模式

Work queues工作队列模式

Work Queues与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。

应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

代码

Work Queues与入门程序的简单模式的代码是几乎一样的；可以完全复制，并复制多一个消费者进行多个消费者同时消费消息的测试。

1.复制一个消费者

2.先运行起两个消费者

3.在生产者中多发布几条消息

for (int i = 0; i < 10; i++) {channel.basicPublish("","test01",null,("第"+i+"次发送").getBytes());
}

4.两个消费者会采用轮询的方式拿到消息

在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系

订阅模式类型

而在订阅模型中，多了一个exchange角色，而且过程略有变化：

生产者发消息给交换机，交换机将消息路由分发给队列，消费者监听队列接收信息

Exchange：交换机。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合指定routing key 的队列

Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

代码实现

生产者更改代码逻辑为:

//        创建交换机channel.exchangeDeclare("Exchange", BuiltinExchangeType.FANOUT,false);
//        创建队列channel.queueDeclare("test01Ex",false,false,false,null);channel.queueDeclare("test02Ex",false,false,false,null);
//        队列绑定交换机channel.queueBind("test01Ex","Exchange","");channel.queueBind("test02Ex","Exchange","");
//        发布消息for (int i = 0; i < 10; i++) {channel.basicPublish("Exchange","",null,("第"+i+"次发送").getBytes());}

消费者分别从两个队列获取消息

Routing路由模式(Direct：定向)

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey(路由key）

消息的发送方在 向Exchange发送消息时，也必须指定消息的RoutingKey

Exchange不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey进行判断，只有队列的RoutingKey与消息的 RoutingKey完全一致，才会接收到消息

 channel.queueBind(queuename,exchangename,"error"); // errorchannel.queueBind(queuename2,exchangename,"error");// error info channel.queueBind(queuename2,exchangename,"info");// 发送消息//String exchange, String routingKey, BasicProperties props, byte[] bodychannel.basicPublish(exchangename,"info",null,"hello".getBytes());

Topics通配符模式

Topic与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定RoutingKey的时候使用通配符！

RoutingKey一般都是有一个或多个单词组成，多个单词之间以”.”分割

#：匹配一个或多个词 

*：匹配不多不少恰好1个词   test.* test.insert

channel.queueBind(queuename,exchangename,"order.*");channel.queueBind(queuename,exchangename,"*.error");channel.queueBind(queuename2,exchangename,"*.*");//发送消息//String exchange, String routingKey, BasicProperties props, byte[] bodychannel.basicPublish(exchangename,"goods.info",null,"hello".getBytes());

模式总结

RabbitMQ工作模式：

简单模式 HelloWorld

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

工作队列模式 Work Queue

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

发布订阅模式 Publish/subscribe

需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

路由模式 Routing

需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

通配符模式 Topic

需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

SpringBoot整合Mq

在Spring项目中，可以使用Spring-Rabbit去操作RabbitMQ

尤其是在spring boot项目中只需要引入对应的amqp启动器依赖即可，方便的使用RabbitTemplate发送消息，使用注解接收消息。

创建工程结构

添加依赖(sys-mq）

<parent><groupId>com.example</groupId><artifactId>mqdemo02</artifactId><version>0.0.1-SNAPSHOT</version></parent><artifactId>sys-mq</artifactId><packaging>pom</packaging><name>sys-mq</name><url>http://maven.apache.org</url><modules><module>mq-product</module><module>mq-consumer</module></modules><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>3.8.1</version><scope>test</scope></dependency>
<!--        rabbitmq--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency></dependencies>

sys-mq里面两个子模块的 application.yml

server

     port: 记得修改

spring:rabbitmq:username: rootpassword: roothost: 192.168.229.16virtual-host: /testport: 5672
server:port: 8086

生产者

@Configuration
public class RabbitMqConfig {
//    设置交换机的名称和队列的名字public static final String EXCHANGE_NAME="exchange_topic-test";public static final String QUEUE_NAME="queue_topic-test";public static final String QUEUE_NAME2="queue_topic-test2";
//    创建交换机 将交换机作为bean注入到spring中@Bean("topicExchange")public Exchange topicExchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(false).build();}
//    队列@Bean("topicQueue")public Queue topicQueue(){return QueueBuilder.durable(QUEUE_NAME).build();}@Bean("topicQueue2")public Queue topicQueue2(){return QueueBuilder.durable(QUEUE_NAME2).build();}
//    将队列与交换机进行绑定@Beanpublic Binding exchangeWithQueue(@Qualifier("topicExchange") Exchange exchange,@Qualifier("topicQueue") Queue queue) {return BindingBuilder.bind(queue).to(exchange).with("*.test").noargs();}@Beanpublic Binding exchangeWithQueue2(@Qualifier("topicExchange") Exchange exchange,@Qualifier("topicQueue2") Queue queue) {return BindingBuilder.bind(queue).to(exchange).with("test.#").noargs();}
}

@SpringBootTest
public class MyTest {public static final String EXCHANGE_NAME="exchange_topic-test";@Resourceprivate RabbitTemplate rabbitTemplate;@Testvoid sendMsg() {rabbitTemplate.convertAndSend(EXCHANGE_NAME,"success.test","测试整合");}
}

消费者

@Component
public class MyListener {
//    监听队列的消息@RabbitListener(queues = "queue_topic-test")public void listenQueue(Message message) {byte[] body = message.getBody();System.out.println(new String(body));}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/216989.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Redis集群(新)

Redis集群(新)

1.什么是集群 Redis集群实现了对Redis的水平扩容&#xff0c;可实现并发写操作&#xff0c;启动n个redis节点&#xff0c;将数据分别存储在不同的节点中&#xff0c;每块节点负责不同区域的插槽&#xff0c;所以Redis集群通过分区来提供一定程度的可用性。 Redis集群现采用的是…
阅读更多...
2023 年 认证杯 小美赛 ABC题 国际大学生数学建模挑战赛 |数学建模完整代码+建模过程全解全析

2023 年 认证杯 小美赛 ABC题 国际大学生数学建模挑战赛 |数学建模完整代码+建模过程全解全析

当大家面临着复杂的数学建模问题时&#xff0c;你是否曾经感到茫然无措&#xff1f;作为2022年美国大学生数学建模比赛的O奖得主&#xff0c;我为大家提供了一套优秀的解题思路&#xff0c;让你轻松应对各种难题。 cs数模团队在认证杯 小美赛前为大家提供了许多资料的内容呀&am…
阅读更多...
leetcode刷题详解二

leetcode刷题详解二

160. 相交链表 本质上是走过自己的路&#xff0c;再走过对方的路&#xff0c;这是求两个链表相交的方法 ListNode *getIntersectionNode(ListNode *headA, ListNode *headB) {//本质上是走过自己的路&#xff0c;再走过对方的路if(headA NULL|| headB NULL){return NULL;}Lis…
阅读更多...
AssembleRH850.dll未能加载或找不到

AssembleRH850.dll未能加载或找不到

AssembleRH850.dll未能加载或找不到 省流解决方案 省流 经过卸了反复重装、杀毒、系统dll修复&#xff0c;百般折腾&#xff0c;是这一堆东西在搞鬼。解决方案 下载DirectX修复工具&#xff08;增强版&#xff09;&#xff0c;专门修复上述的C问题。
阅读更多...
数字孪生技术：提升UI交互性与个性化设计

数字孪生技术：提升UI交互性与个性化设计

随着数字化时代的到来&#xff0c;数字孪生技术正在逐渐改变我们的生活和工作方式。数字孪生是一种复制现实世界系统或实体的技术&#xff0c;通过创建数字模型来模拟现实世界中的各种行为和事件。这种技术不仅为人们提供了一个全新的视角来看待和解决问题&#xff0c;同时也为…
阅读更多...
HarmonyOS简述及开发环境搭建

HarmonyOS简述及开发环境搭建

一、HarmonyOS简介 1、介绍 HarmonyOS是一款面向万物互联时代的、全新的分布式操作系统。有三大系统特性&#xff0c;分别是&#xff1a;硬件互助&#xff0c;资源共享&#xff1b;一次开发&#xff0c;多端部署&#xff1b;统一OS&#xff0c;弹性部署。 HarmonyOS通过硬件互…
阅读更多...
QT已有项目导入工程时注意事项

QT已有项目导入工程时注意事项

文章目录 从qt其他版本上开发的工程导入另一qt版本时 从qt其他版本上开发的工程导入另一qt版本时 这里以之前在qt5.12.2上开发的项目为例&#xff0c;现在到在qt6.5.3上运行。 不能直接导入IDE上&#xff0c;否则会报各种莫名奇妙的错误。 首先要把扩展名位.pro.user文件 删掉…
阅读更多...
MySQL 及 SQL 注入

MySQL 及 SQL 注入

文章目录 前言什么是sql注入防止SQL注入Like语句中的注入后言 前言 hello world欢迎来到前端的新世界 &#x1f61c;当前文章系列专栏&#xff1a;Mysql &#x1f431;‍&#x1f453;博主在前端领域还有很多知识和技术需要掌握&#xff0c;正在不断努力填补技术短板。(如果出现…
阅读更多...
Redis主从复制(新)

Redis主从复制(新)

像一些大型的项目&#xff0c;为了保证系统的稳定性&#xff0c;会有一台Redis服务器专门作为主机(master)&#xff0c;还会有多台服务器作为从机(slave)&#xff0c;主机可专门用作写数据&#xff0c;然后根据配置和策略&#xff0c;自动将数据同步到从机&#xff0c;而从机专…
阅读更多...
arp报文及使用go实现

arp报文及使用go实现

一、ARP协议报文格式及ARP表 ARP&#xff08;Address Resolution Protocal&#xff0c;地址解析协议&#xff09;是将IP地址解析为以太网的MAC地址&#xff08;或者称为物理地址&#xff09;的协议。在局域网中&#xff0c;当主机或其他网络设备有数据要发送给另一个主机或设备…
阅读更多...
2023 年 认证杯 小美赛 国际大学生数学建模挑战赛 |数学建模完整代码+建模过程全解全析

2023 年 认证杯 小美赛 国际大学生数学建模挑战赛 |数学建模完整代码+建模过程全解全析

当大家面临着复杂的数学建模问题时&#xff0c;你是否曾经感到茫然无措&#xff1f;作为2022年美国大学生数学建模比赛的O奖得主&#xff0c;我为大家提供了一套优秀的解题思路&#xff0c;让你轻松应对各种难题。 cs数模团队在认证杯 小美赛前为大家提供了许多资料的内容呀&am…
阅读更多...
小程序姓名：ssm+vue基本微信小程序的个人健康管理系统

小程序姓名：ssm+vue基本微信小程序的个人健康管理系统

项目介绍 首先,论文一开始便是清楚的论述了小程序的研究内容。其次,剖析系统需求分析,弄明白“做什么”,分析包括业务分析和业务流程的分析以及用例分析,更进一步明确系统的需求。然后在明白了小程序的需求基础上需要进一步地设计系统,主要包罗软件架构模式、整体功能模块、数…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023