RabbitMQ消息队列入门

た入场券 2021-11-11 13:04 321阅读 0赞

# 什么是消息队列？ #

解决应用程序与应用程序之间的通信。利用典型的消费者生产者模型。消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，实现了生产者和消费者的解耦。

## AMQP和JMS ##

AMQP 高级消息队列协议，规定数据传输格式和方式。垮语言垮平台。  
JMS JavaMessageService 必须使用java语言

*  ActiveMQ：基于JMS
 *  RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
 *  RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
 *  Kafka：分布式消息系统，高吞吐量

rabbitMQ 基于erlang，所以安装前需要安装erlang

--------------------

## 6种基本模型 ##

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

# 1.模型一 #

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

## 1.1 长连接的创建 ##

#### 引入依赖 ####

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

#### 工厂模式创建链接并设置ip及用户名密码 ####

1.创建ConnectionFactory工厂  
2.设置链接信息，setHost，setPort，setVirtualHost，setUsername，setPassword  
![!\[在这里插入图片描述\](https://img-blog.csdnimg.cn/20190811134012576.png?x-oss-process=image/watermark,type\_ZmFuZ3poZW5naGVpdGk,shadow\_10,text\_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8=,size\_16,color\_FFFFFF,t\_70][https_img-blog.csdnimg.cn_20190811134012576.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70]

### 1.2消息发送 ###

1.创建通道cn.createChannel  
2.声明（创建）队列，queueDeclare  
3.向指定的队列中发送消息,转成字节发送basicPublish。  
4.关闭链接  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 2]

## 1.3消息的获取 ##

1.  同样先获取链接及创建通道，声明队列。（如果被声明过会放弃声明）
2.  定义匿名类，实现接口DefaultConsumer定义队列的消费者，参数为通道。
3.  覆写handleDelivery方法。
4.  参数body即为消息内容。
5.  消费者绑定到该队列中 实时监听队列channel.basicConsume
6.  程序不会结束，会一直监听。

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

## 避免消息丢失 ##

消息如果被接收后，还未处理完消费者就宕机了？

消息确认处理机制：

*  自动ACK：消息一旦被接收，消费者自动发送ACK
 *  手动ACK：消息接收后，需要手动调用回执

如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK。

channel.basicConsume，第二个参数，bool类型。是否自动确认。  
改为false后，需要在消息handleDelivery中，  
执行手动ACK

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

--------------------

# 2.模型二 work #

多个消费者绑定同一个队列，同一个消息只会被一个消费者接收到  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 4]

### 2.1轮流发送与发送数。 ###

使用basicQos方法和prefetchCount = 1设置（预拉取数），设置在未发送ack之前，不再将对其发送新消息。能者多劳!

# 3.模式三、四、五 发布订阅模式 #

### 增加exchange 交换器。 ###

## 3.1 订阅模型-Fanout：广播 ##

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

##### 3.2.1 Fanout广播模式的特点 #####

生产者将消息发送给交换器，由交换器来进行发送至队列  
每个消费者有自己的队列。  
一条消息被多个消费者消费。  
适用于不同类型程序对同一个消息有不同的响应。

###### 3.2.2 生产者发布信息到交换器 ######

1.  新建连接，获取通道
2.  通道中声明exchange，指定类型，可以用枚举类“BuiltinExchangeType.FANOUT”
3.  将信息发送到exchange。channel.basicPublish(exchangename,"",null,message.getBytes())。
4.  关闭连接  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 6]  
    ![在这里插入图片描述][20190811170110629.png]  
    参数1为交换器名字，参数2为队列名字，如果使用fanout模式，第二项可以为null；

###### 3.2.3 消费者接收数据 ######

1.  新建连接，创建通道
2.  声明队列
3.  将队列绑定到exchange上
4.  new DefaultConsumer定义消费者消息处理方法
5.  监听队列 channel.basicConsume  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 7]  
    由消费者声明队列通道，并将通道绑定到交换器channel.queueBind。  
    交换机不保存信息，如果未绑定，则消息会丢失。

## 3.2 订阅模型-Direct：定向 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 8]  
创建exchange的时候声明direct类型。（有枚举）

*  队列与交换机绑定queueBind时候必须指定一个RoutingKey
 *  生产者往交换器发送消息的时候，也必须指定RoutingKey

## 3.3 订阅模型-Topic：通配符 ##

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

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

`Routingkey` 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： `item.insert`

通配符规则：

`#`：匹配一个或多个词

`*`：匹配不多不少恰好1个词

举例：

`audit.#`：能够匹配`audit.irs.corporate` 或者 `audit.irs`

`audit.*`：只能匹配`audit.irs`

--------------------

# RabbitMQ持久化 --解决消息丢失 #

队列及交换机默认在重启后消失，默认不会持久化  
`durable` 在交换器及队列声明时，可以配置为true。  
交换器：  
![在这里插入图片描述][20190811191352573.png]  
队列：  
![在这里插入图片描述][20190811191359696.png]

消息的持久化：在basicPublish发送时候，第三个参数props。  
![在这里插入图片描述][20190811192254135.png]  
PERSISTENT\_TEXT\_PLAIN常用，PERSISTENT表示持久。

另外：

*  手动ack也能解决消息丢失。
 *  还有生产者确认防丢失，由MQ发送回执给生产方。  
    开启生产者确认  
    `channel.confirmSelect();`  
    处理方法一：  
    `channel.waitForConfirms` 失败抛出异常，处理的时候重发。  
    处理方法二：  
    `channel.addConfirmListener` 两个参数，回调两个方法。一个是成功方法，一个是失败时候的方法。
 *  发送消息前，将消息持久到数据库，并保存消息状态。（可靠消息服务）

幂等性：同一接口被重复执行，其结果一致。

**spring有提供更好的方法！**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70]: /images/20211109/63e9ef0150774c5abe26bcbfea8c7cb6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 1]: /images/20211109/5f4eeb43530b4ced96c778e9388d7b5b.png
[https_img-blog.csdnimg.cn_20190811134012576.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70]: /images/20211109/325c9aa3032045628c0cf0f6e9e17724.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 2]: /images/20211109/3ad1885c21db4ecca92e53b87019ff0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 3]: /images/20211109/0f0d0888760c463fbb1ab4f40cea7a93.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 4]: /images/20211109/3f34a1067bcc487482389acd5141b1da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 5]: /images/20211109/c4130dc5d14045b589155f97747a9fd2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 6]: /images/20211109/e752101f75ff422497ba7b810671ae77.png
[20190811170110629.png]: /images/20211109/7670516d463c4150928e3ac0f9fcab80.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 7]: /images/20211109/d2fe31f3120340c3986af7ecf7c0aea4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 8]: /images/20211109/d62df165f5e043429ab9d628a6d164ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Zpdmlicm8_size_16_color_FFFFFF_t_70 9]: /images/20211109/e01077bd77df4edfa597d5acc1134918.png
[20190811191352573.png]: /images/20211109/3b0f4ed00c284cf7981f2ff1514680ce.png
[20190811191359696.png]: /images/20211109/273e9e9dc28e40bd8e46e71e8b060b70.png
[20190811192254135.png]: /images/20211109/7cdc1875bdeb4f46a26cbbd7fd2d23e8.png