rabbitmq-01 rabbitmq概念介绍

待我称王封你为后i 2022-02-09 14:11 393阅读 0赞

# 1.RabbitMQ简介 #

[RabbitMQ][]是一个由[erlang][]开发的AMQP（Advanced Message Queue ）的开源实现。AMQP 的出现其实也是应了广大人民群众的需求，虽然在同步消息通讯的世界里有很多公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 MSMQ ，IBM 的 Websphere MQ 等），因此，在 2006 年的 6 月，Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。

RabbitMQ是由RabbitMQ Technologies Ltd开发并且提供商业支持的。该公司在2010年4月被SpringSource（VMWare的一个部门）收购。在2013年5月被并入Pivotal。其实VMWare，Pivotal和EMC本质上是一家的。不同的是VMWare是独立上市子公司，而Pivotal是整合了EMC的某些资源，现在并没有上市。

注意:RabbitMQ是采用erlang语言开发的，所以必须有erlang环境才可以运行

## 1.1 erlang语言简介 ##

[Erlang][]编程语言最初目的是进行大型电信交换设备的软件开发，是一种适用于大规模并行处理环境的高可靠性编程语言。随着多核处理器技术的日渐普及，以及互联网、云计算等技术的发展，该语言的应用范围也有逐渐扩大之势。

初衷理念实现抗高并发语言

# 2.几种常见消息中间件 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDYzOTk5_size_16_color_FFFFFF_t_70]

# 3.RabbitMQ环境安装 #

## 3.1 windows下安装 ##

*  下载并安装erlang,下载地址：[http://www.erlang.org/download][http_www.erlang.org_download]
 *  配置erlang环境变量信息
    
     *  新增环境变量ERLANG\_HOME=erlang的安装地址
     *  将%ERLANG\_HOME%\\bin加入到path中
 *  下载并安装RabbitMQ，下载地址：[http://www.rabbitmq.com/download.html][http_www.rabbitmq.com_download.html]

注意: RabbitMQ 它依赖于Erlang,需要先安装Erlang。

## 3.2 docker安装 ##

*  拉取镜像

docker pull rabbitmq:3-management

*  启动镜像（默认用户名密码）,默认guest 用户，密码也是 guest

docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3-management

*  启动镜像（设置用户名密码）

docker run -d --hostname my-rabbit --name rabbit -e RABBITMQ_DEFAULT_USER=user -e RABBITMQ_DEFAULT_PASS=password -p 15672:15672 -p 5672:5672 rabbitmq:3-management

# 4.RabbitMQ管理平台 #

*  RabbitMQ 管理平台地址:[http://127.0.0.1:15672][http_127.0.0.1_15672]
 *  默认账号:guest/guest  
    用户可以自己创建新的账号

## 4.1 Virtual Hosts ##

像mysql有数据库的概念并且可以指定用户对库和表等操作的权限。那RabbitMQ呢？RabbitMQ也有类似的权限管理。在RabbitMQ中可以虚拟消息服务器VirtualHost，每个VirtualHost相当月一个相对独立的RabbitMQ服务器，每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDYzOTk5_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDYzOTk5_size_16_color_FFFFFF_t_70 2]

# 5.RabbitMQ关键名词 #

> AMQP（高级消息队列协议）是一个异步消息传递所使用应用层协议规范，为面向消息中间件设计，基于此协议的客户端与消息中间件可以无视消息来源传递消息，不受客户端、消息中间件、不同的开发语言环境等条件的限制；

涉及概念解释：

*  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绑定起来。

## 5.1 RabbitMQ交换机的作用 ##

![在这里插入图片描述][20190503152318775.png]  
生产者发送消息不会向传统方式直接将消息投递到队列中，而是先将消息投递到交换机中，在由交换机转发到具体的队列，队列在将消息以推送或者拉取方式给消费者进行消费，这和我们之前学习Nginx有点类似。

交换机的作用根据具体的路由策略分发到不同的队列中，交换机有四种类型。

*  Direct exchange（直连交换机）是根据消息携带的路由键（routing key）将消息投递给对应队列的
 *  Fanout exchange（扇型交换机）将消息路由给绑定到它身上的所有队列
 *  Topic exchange（主题交换机）队列通过路由键绑定到交换机上，然后，交换机根据消息里的路由值，将消息路由给一个或多个绑定队列
 *  Headers exchange（头交换机）类似主题交换机，但是头交换机使用多个消息属性来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。 (很少使用)

## 5.2 消息队列RabbitMQ应答模式 ##

为了确保消息不会丢失，RabbitMQ支持消息应答。消费者发送一个消息应答，告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ就可以删除它了。  
如果一个消费者挂掉却没有发送应答，RabbitMQ会理解为这个消息没有处理完全，然后交给另一个消费者去重新处理。这样，你就可以确认即使消费者偶尔挂掉也不会丢失任何消息了。  
没有任何消息超时限制；只有当消费者挂掉时，RabbitMQ才会重新投递。即使处理一条消息会花费很长的时间。  
消息应答是默认打开的。我们通过显示的设置autoAsk=true关闭这种机制。现即自动应答开，一旦我们完成任务，消费者会自动发送应答。通知RabbitMQ消息已被处理，可以从内存删除。如果消费者因宕机或链接失败等原因没有发送ACK（不同于ActiveMQ，在RabbitMQ里，消息没有过期的概念），则RabbitMQ会将消息重新发送给其他监听在队列的下一个消费者。

### 5.2.1 案例 ###

生产者端代码不变，消费者端代码这部分就是用于开启手动应答模式的。

channel.basicConsume(QUEUE_NAME, false, defaultConsumer);

注：第二个参数值为false代表关闭RabbitMQ的自动应答机制，改为手动应答。  
在处理完消息时，返回应答状态，true表示为自动应答模式。

channel.basicAck(envelope.getDeliveryTag(), false);

## 5.3 RabbitMQ的公平转发 ##

目前消息转发机制是平均分配，这样就会出现俩个消费者，奇数的任务很耗时，偶数的任何工作量很小，造成的原因就是近当消息到达队列进行转发消息。并不在乎有多少任务消费者并未传递一个应答给RabbitMQ。仅仅盲目转发所有的奇数给一个消费者，偶数给另一个消费者。  
为了解决这样的问题，我们可以使用basicQos方法，传递参数为prefetchCount= 1。这样告诉RabbitMQ不要在同一时间给一个消费者超过一条消息。  
换句话说，只有在消费者空闲的时候会发送下一条信息。调度分发消息的方式，也就是告诉RabbitMQ每次只给消费者处理一条消息，也就是等待消费者处理完毕并自己对刚刚处理的消息进行确认之后，才发送下一条消息，防止消费者太过于忙碌，也防止它太过去清闲。  
通过设置channel.basicQos(1);  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDYzOTk5_size_16_color_FFFFFF_t_70 3]

## 5.4 RabbitMQ消息确认机制 ##

生产者发送消息出去之后，不知道到底有没有发送到RabbitMQ服务器， 默认是不知道的。而且有的时候我们在发送消息之后，后面的逻辑出问题了，我们不想要发送之前的消息了，需要撤回该怎么做。

### 5.4.1 解决方案 ###

*  AMQP 事务机制
 *  Confirm 模式

### 5.4.2 事务模式 ###

*  txSelect 将当前channel设置为transaction模式
 *  txCommit 提交当前事务
 *  txRollback 事务回滚

#### 5.4.2.1 代码实现 ####

*  [AmqpProducer][]

### 5.4.3 Confirm 模式 ###

*  [ConfirmProducer][]

[RabbitMQ]: http://www.rabbitmq.com
[erlang]: https://baike.baidu.com/item/AMQP/8354716?fr=aladdin
[Erlang]: https://baike.baidu.com/item/Erlang%E8%AF%AD%E8%A8%80/20864044?fr=aladdin
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDYzOTk5_size_16_color_FFFFFF_t_70]: /images/20220206/5487e3a56f724d1aa96c6f3f13576361.png
[http_www.erlang.org_download]: http://www.erlang.org/download
[http_www.rabbitmq.com_download.html]: http://www.rabbitmq.com/download.html
[http_127.0.0.1_15672]: http://127.0.0.1:15672
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDYzOTk5_size_16_color_FFFFFF_t_70 1]: /images/20220206/1587e40c68eb45cfb3e83cb3081ee2c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDYzOTk5_size_16_color_FFFFFF_t_70 2]: /images/20220206/d64d7e1ba3714f0aa452a59622054bce.png
[20190503152318775.png]: /images/20220206/2e9335c44b6949ceac5f0fc527d7d8bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDYzOTk5_size_16_color_FFFFFF_t_70 3]: /images/20220206/76b95ee39148452f9d82ad00a9b9d538.png
[AmqpProducer]: https://gitee.com/itmyx/rabbitmq-demo/blob/master/rabbitmq-queue/src/main/java/com/grape/rabbitmq/queue/ack/amqp/AmqpProducer.java
[ConfirmProducer]: https://gitee.com/itmyx/rabbitmq-demo/blob/master/rabbitmq-queue/src/main/java/com/grape/rabbitmq/queue/ack/confirm/ConfirmProducer.java