【RabbitMQ】快速入门及基本使用

这篇具有很好参考价值的文章主要介绍了【RabbitMQ】快速入门及基本使用。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、引言

1、、消息队列

Ⅰ、什么是消息队列?

        消息队列是一种进程间通信或同一进程的不同线程间的通信方式,软件的贮列用来处理一系列的输入,通常是来自用户。消息队列提供了异步的通信协议,每一个贮列中的纪录包含详细说明的数据,包含发生的时间,输入设备的种类,以及特定的输入参数。也就是说:消息的发送者和接收者不需要同时与消息队列交互。消息会保存在队列中,直到接收者取回它。

Ⅱ、Message queue 释义

        服务之间最常见的通信方式是直接调用彼此来通信,消息从一端发出后立即就可以达到另一端,称为即时消息通讯(同步通信) 消息从某一端发出后,首先进入一个容器进行临时存储,当达到某种条件后,再由这个容器发送给另一端,称为延迟消息通讯(异步通信)

Ⅲ、消息队列特点

  1. 异步性:消息队列允许发送者发送消息后继续其它任务,而不需要等待接收者的响应。这种异步性能够提高系统的吞吐量和响应速度。
  2. 解耦性:消息队列将发送者和接收者解耦,使它们不需要知道彼此的存在。发送者只需将消息发送到队列中,而不需要关心谁将接收该消息。接收者通过订阅队列来获取消息,而不需要关心消息的发送者。这种解耦性能够提高系统的可扩展性和可维护性。
  3. 可靠性:消息队列通常具有持久化机制,可以确保消息的可靠性传输。即使在发送者或接收者出现故障的情况下,消息也不会丢失。此外,消息队列还具有消息重试和消息确认等机制,确保消息能够被正确处理。
  4. 消息类型和格式:消息队列中的消息是有类型的,并且具有格式。这使得消息可以被按类型读取,并遵循一定的格式。
  5. 多进程支持:消息队列允许一个或多个进程向它写入或者读取消息。
  6. 数据持久性:当从消息队列中读出消息后,消息队列中对应的数据都会被删除,确保数据不会重复消费。
  7. 分布式特性:通过对消费者的横向扩展,降低了消息队列阻塞的风险,以及单个消费者产生单点故障的可能性。

Ⅳ、好处

  1. 高吞吐量:由于消息的传输速度比普通的文件快,所以能够实现高吞吐量。
  2. 支持异步操作:一个线程在处理完自己的任务之后,可以把结果发送到另一个线程。
  3. 支持并发操作:多个线程可以同时处理消息队列中的消息。
  4. 支持分布式系统:由于消息队列的解耦特性,分布式系统中的组件可以独立扩展。
  5. 提供数据持久化机制:消息队列可以将数据进行持久化,确保数据不会因为处理过程中的失败而丢失。
  6. 提供错误恢复机制:当系统的一部分组件失效时,消息队列可以保证即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。
  7. 提高响应速度:通过异步处理,消息队列可以显著提高系统的响应速度。
  8. 解耦:在项目启动之初来预测将来项目会碰到什么需求是极其困难的。使用消息队列可以在处理过程中间插入一个隐含的、基于数据的接口层,两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。
  9. 扩展性:由于消息队列解耦了你的处理过程,所以增大消息入队和处理的频率是很容易的,只要另外增加处理过程即可。
  10. 灵活性:使用消息队列能够使关键组件顶住增长的访问压力,而不是因为超出负荷的请求而完全崩溃。
  11. 可恢复性:当体系的一部分组件失效时,不会影响到整个系统。
  12. 送达保证:消息队列提供的冗余机制保证了消息能被实际的处理,只要一个进程读取了该队列即可。
  13. 排序保证:在许多情况下,数据处理的顺序都很重要。消息队列能保证数据会按照特定的顺序来处理。
  14. 缓冲:在任何重要的系统中,都会有需要不同的处理时间的元素。例如,加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行——写入队列的处理会尽可能的快速,而不受从队列读的预备处理的约束。该缓冲有助于控制和优化数据流经过系统的速度。
  15. 理解数据流:在一个分布式系统里,要得到一个关于用户操作会用多长时间及其原因的总体印象,是个巨大的挑战。消息系列通过消息被处理的频率,来方便的辅助确定那些表现不佳的处理过程或领域,这些地方的数据流都不够优化。
  16. 异步通信:很多时候,你不想也不需要立即处理消息。消息队列提供了异步处理机制,允许你把一个消息放入队列,但并不立即处理它。你想向队列中放入多少消息就放多少,然后在你乐意的时候再去处理它们。

2、消息队列相关

Ⅰ、AMQP

        一个提供统一消息服务的应用层标准高级消息队列协议,是一个通用的应用层协议 消息发送与接受的双方遵守这个协议可以实现异步通讯.这个协议约定了消息的格式和工作方式.

技术选型

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

3、什么是RabbitMQ

        RabbitMQ是一个开源的消息队列系统,使用Erlang语言开发,基于AMQP(高级消息队列协议)实现。它最初起源于金融系统,用于在分布式系统中存储和转发消息。RabbitMQ的主要特性包括易用性、扩展性、高可用性以及可靠性。消息队列(MQ)是一种应用程序对应用程序的通信方法,应用程序通过读写出入队列的消息来通信,而无需专用连接来链接它们。在事件驱动(发布-订阅)架构中,RabbitMQ扮演着Broker的角色。

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

  • Server(Broker):接收客户端连接,实现AMQP协议的消息队列和路由功能的进程.
  • Virtual Host:虚拟主机的概念,类似权限控制组,一个Virtual Host里可以有多个Exchange和Queue.
  • Exchange:交换机,接收生产者发送的消息,并根据Routing Key将消息路由到服务器中的队列Queue.
  • ExchangeType:交换机类型决定了路由消息行为,RabbitMQ中有三种类型Exchange,分别是fanout、direct、topic.
  • Message Queue:消息队列,用于存储还未被消费者消费的消息.
  • Message:由Header和body组成,Header是由生产者添加的各种属性的集合,包括Message是否被持久化、优先级是多少、由哪个Message Queue接收等.body是真正需要发送的数据内 容.
  • BindingKey:绑定关键字,将一个特定的Exchange和一个特定的Queue绑定起来.

二、快速入门

1、Docker安装部署RabbitMQ

【注意】获取镜像的时候要获取management版本的,不要获取last版本的,management版本的才带有管理界面

docker pull rabbitmq:management

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

--hostname:主机名(RabbitMQ的一个重要注意事项是它根据所谓的 “节点名称” 存储数据,默认为主机名)

-e:指定环境变量:
       RABBITMQ_DEFAULT_VHOST:默认虚拟机名
        RABBITMQ_DEFAULT_USER:默认的用户名
        RABBITMQ_DEFAULT_PASS:默认用户名的密码

docker run -d \
--name 容器名\
-p 5672:5672 -p 15672:15672 \
-v /home/rabbitmq:/var/lib/rabbitmq \
--hostname my-rabbitmq-host \
-e RABBITMQ_DEFAULT_VHOST=my_vhost \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin \
--restart=always \
rabbitmq:management

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

容器启动后,查看容器日志

docker logs 容器名

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

开启防火墙

systemctl start firewalld

开放端口

firewall-cmd --zone=public --add-port=15672/tcp --permanent
firewall-cmd --zone=public --add-port=5672/tcp --permanent

更新防火墙规则

firewall-cmd --reload

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

Ⅰ、配置用户

进入管理后台

密码:admin

账号:admin

http://ip:15672

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

创建用户

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

点进用户进行分配

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

2、springboot连接配置

Ⅰ、新建项目

 新建一个空项目,里面新建两个spring boot模块,并且导入依赖

接收者(publisher)消费者(consumer)

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

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

Ⅱ、配置yml

spring:
  rabbitmq:
    host: 0.0.0.0 #虚拟机开启的IP地址
    username: spring #创建的用户
    password: 123456 #用户的密码
    port: 5672
    virtual-host: my_vhost 

Ⅲ、案例一

接收一个string类型

生产者

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@SuppressWarnings("all")
public class RabbitConfig {
   @Bean
    public Queue firstQueue() {
        // 创建一个名为firstQueue的队列
        return new Queue("firstQueue");
    }
}

使用Controller 

@RestController
@SuppressWarnings("all")
public class SenderController {
    // 自动装配rabbitTemplate
    @Autowired
    private AmqpTemplate rabbitTemplate;

    // 发送消息到firstQueue队列
    @RequestMapping("/sendFirst")
    public String sendFirst() {
        // 将消息转换并发送到firstQueue队列
        rabbitTemplate.convertAndSend("firstQueue", "Hello World");
        return "🐉";
    }
}

访问http://localhost:8888/sendFirst

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

可以看到消息队列中有了一个

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

消费者

@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "firstQueue")
public class Receiver {
    @RabbitHandler
    public void process(String msg) {
        log.warn("接收到:" + msg);
    }
}

运行之后可以看到我们发送的消息

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

Ⅳ、案例二

我们接受一个实体类

生产者(publisher)

新建一个实体类User 实现接口Serializable

@SuppressWarnings("all")
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User implements Serializable {
private String username;
private String userpwd;
}

RabbitConfig 类里面添加一个队列

    @Bean
    public Queue secondQueue() {
        // 创建一个名为 secondQueue 的队列
        return new Queue("secondQueue");
    }

完整代码 RabbitConfig 

@Configuration
@SuppressWarnings("all")
public class RabbitConfig {
    @Bean
    public Queue firstQueue() {
        // 创建一个名为firstQueue的队列
        return new Queue("firstQueue");
    }

    @Bean
    public Queue secondQueue() {
        // 创建一个名为 secondQueue 的队列
        return new Queue("secondQueue");
    }
}

在Controller 层添加方法

        // 自动装配ObjectMapper
    @Autowired
    private ObjectMapper objectMapper;
// 发送消息到secondQueue队列
    @RequestMapping("/send2")
    public String send2() throws JsonProcessingException {
        // 创建一个User对象
        User wfzldr = new User("wfzldr", "1234567890");
        // 将User对象转换为json字符串
        String json = objectMapper.writeValueAsString(wfzldr);
        // 将消息转换并发送到firstQueue队列
        // 将消息转换并发送到secondQueue队列
        rabbitTemplate.convertAndSend("secondQueue", json);
        return "🐉";
    }

完整代码SenderController

@RestController
@SuppressWarnings("all")
public class SenderController {
   // 自动装配rabbitTemplate
    @Autowired
    private AmqpTemplate rabbitTemplate;
    // 自动装配ObjectMapper
    @Autowired
    private ObjectMapper objectMapper;

    // 发送消息到firstQueue队列
    @RequestMapping("/sendFirst")
    public String sendFirst() {
        // 将消息转换并发送到firstQueue队列
        rabbitTemplate.convertAndSend("firstQueue", "Hello World");
        return "🐉";
    }

    // 发送消息到secondQueue队列
    @RequestMapping("/send2")
    public String send2() throws JsonProcessingException {
        // 创建一个User对象
        User wfzldr = new User("wfzldr", "1234567890");
        // 将User对象转换为json字符串
        String json = objectMapper.writeValueAsString(wfzldr);
        // 将消息转换并发送到firstQueue队列
        // 将消息转换并发送到secondQueue队列
        rabbitTemplate.convertAndSend("secondQueue", json);
        return "🐉";
    }
}

消费者(consumer)

新建一个接受实体的Receiver

@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "secondQueue")
public class EntityReceiver {
    @Autowired
    private ObjectMapper objectMapper;

    @RabbitHandler
    public void process(String json) throws JsonProcessingException {
        User user = objectMapper.readValue(json, User.class);
        log.warn("接收到:" + user);
    }
}

我破门也需要在消费者里添加实体

@SuppressWarnings("all")
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User implements Serializable {
private String username;
private String userpwd;
}

运行两个方法
【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议
 

3、创建公共模块

在项目里面新建一个publi的公共模块,在里面写公共的实体类

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

在公共模块的pom.xml文件里面把打包方式war改成 jar 

【RabbitMQ】快速入门及基本使用,rabbitmq,分布式,运维,网络,网络协议

在对应的消费者或者生产者里面引入对应的模块即可

        <!--        引入公共模块-->
        <dependency>
            <groupId>org.example</groupId>
            <artifactId>public</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

我的分享就到这里,欢迎大家在评论区留言!文章来源地址https://www.toymoban.com/news/detail-808391.html

到了这里,关于【RabbitMQ】快速入门及基本使用的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Rabbitmq----分布式场景下的应用

    如果单机模式忘记也可以看看这个快速回顾rabbitmq,在做学习 消息队列在使用过程中,面临着很多实际问题需要思考: 消息从发送,到消费者接收,会经理多个过程: 其中的每一步都可能导致消息丢失,常见的丢失原因包括: 发送时丢失: 生产者发送的消息未送达exchange 消

    2024年02月08日
    浏览(47)
  • RabbitMQ——解决分布式事务问题,RabbitMQ的重要作用之一!!!通过可靠生产和可靠消费来完美解决!

    分布式事务是指涉及多个独立的计算机系统(也称为节点或参与者)之间的事务处理。在分布式系统中,每个节点可能各自拥有自己的数据存储和事务管理机制。分布式事务的目标是保证在跨多个节点执行的一系列操作可以以一致和可靠的方式执行和提交,即使在面对故障或

    2024年04月23日
    浏览(46)
  • 微服务学习:SpringCloud+RabbitMQ+Docker+Redis+搜索+分布式

    目录 一、高级篇 二、面试篇 ==============实用篇============== day05-Elasticsearch01 1.初识elasticsearch 1.4.安装es、kibana 1.4.1.部署单点es 1.4.2.部署kibana 1.4.3.安装IK分词器 1.4.4.总结 2.索引库操作 2.1.mapping映射属性 2.2.索引库的CRUD 2.2.1.创建索引库和映射 2.2.2.查询索引库 2.2.3.修改索引库 2.

    2024年02月02日
    浏览(54)
  • Python爬虫分布式架构 - Redis/RabbitMQ工作流程介绍

    在大规模数据采集和处理任务中,使用分布式架构可以提高效率和可扩展性。本文将介绍Python爬虫分布式架构中常用的消息队列工具Redis和RabbitMQ的工作流程,帮助你理解分布式爬虫的原理和应用。 为什么需要分布式架构? 在数据采集任务中,单机爬虫可能面临性能瓶颈和资

    2024年02月11日
    浏览(40)
  • 分布式消息队列:Kafka vs RabbitMQ vs ActiveMQ

    在现代分布式系统中,消息队列是一种常见的异步通信模式,它可以帮助系统处理高并发、高可用性以及容错等问题。在这篇文章中,我们将深入探讨三种流行的分布式消息队列:Apache Kafka、RabbitMQ和ActiveMQ。我们将讨论它们的核心概念、算法原理、特点以及使用场景。 随着

    2024年02月02日
    浏览(60)
  • (黑马出品_07)SpringCloud+RabbitMQ+Docker+Redis+搜索+分布式

    [此文档是在心向阳光的天域的博客加了一些有助于自己的知识体系,也欢迎大家关注这个大佬的博客](https://blog.csdn.net/sinat_38316216/category_12263516.html) [是这个视频](https://www.bilibili.com/video/BV1LQ4y127n4/?p=5spm_id_from=pageDrivervd_source=9beb0a2f0cec6f01c2433a881b54152c) 聚合 可以让我们极其方便

    2024年03月12日
    浏览(52)
  • 微服务技术栈SpringCloud+RabbitMQ+Docker+Redis+搜索+分布式(五):分布式搜索 ES-下

    聚合(aggregations)可以实现对文档数据的统计、分析、运算。聚合常见的有三类: 桶(Bucket)聚合:用来对文档做分组 TermAggregation:按照文档字段值分组 Date Histogram:按照日期阶梯分组,例如一周为一组,或者一月为一组 度量(Metric)聚合:用以计算一些值,比如:最大值

    2024年03月26日
    浏览(61)
  • 分布式搜索引擎(Elastic Search)+消息队列(RabbitMQ)部署(商城4)

    1、全文搜索 Elastic search可以用于实现全文搜索功能,例如商城中对商品搜索、搜索、分类搜索、订单搜索、客户搜索等。它支持复杂的查询语句、中文分词、近似搜索等功能,可以快速地搜索并返回匹配的结果。 2、日志分析 Elastic search可以用于实现实时日志分析,例

    2024年02月04日
    浏览(48)
  • 【运维知识进阶篇】zabbix5.0稳定版详解7(zabbix分布式监控:使用场景+功能详解+快速部署+基本使用)

    如果你有几百上千台客户端的数据需要上报给zabbix服务端,即便是你做了主动注册,监控项目主动式,那服务端压力还是会很大,所以我们可以考虑zabbix分布式监控。 zabbix proxy可以代替zabbix server收集性能和可用性数据,然后把数据汇报给zabbix server,并且在一定程序分担zab

    2024年02月11日
    浏览(56)
  • 分布式 SpringCloudAlibaba、Feign与RabbitMQ实现MySQL到ES数据同步

    本文参考黑马 分布式Elastic search Elasticsearch是一款非常强大的开源搜索引擎,具备非常多强大功能,可以帮助我们从海量数据中快速找到需要的内容 同步调用 方案一:同步调用 基本步骤如下: hotel-demo对外提供接口,用来修改elasticsearch中的数据 酒店管理服务在完成数据库操

    2024年04月11日
    浏览(45)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包