rabbitMq笔记

一、理论知识

1、什么是中间件

1.1、中间件的概念

1. 中间件（Middleware）是处于操作系统和应用程序之间的软件，也有人认为它应该属于操作系统中的一部分。人们在使用中间件时，往往是一组中间件集成在一起，构成一个平台（包括开发平台和运行平台），但在这组中间件中必须要有一个通信中间件，即中间件=平台+通信，这个定义也限定了只有用于分布式系统中才能称为中间件，同时还可以把它与支撑软件和实用软件区分开来。

   举例：
   1，RMI（Remote Method Invocations, 远程调用）
   2，Load Balancing(负载均衡，将访问负荷分散到各个服务器中)
   3，Transparent Fail-over(透明的故障切换)
   4，Clustering(集群,用多个小的服务器代替大型机）
   5，Back-end-Integration(后端集成，用现有的、新开发的系统如何去集成遗留的系统)
   6，Transaction事务（全局/局部）全局事务（分布式事务）局部事务（在同一数据库联接内的事务）
   7，Dynamic Redeployment(动态重新部署,在不停止原系统的情况下，部署新的系统）
   8，System Management(系统管理)
   9，Threading(多线程处理)
   10，Message-oriented Middleware面向消息的中间件（异步的调用编程）
   11，Component Life Cycle(组件的生命周期管理)
   12，Resource pooling（资源池）
   13，Security（安全）
   14，Caching（缓存）

1.2、为什么需要使用消息

1. 中间件具体地说，中间件屏蔽了底层操作系统的复杂性，使程序开发人员面对一个简单而统一的开发环境，减少程序设计的复杂性，将注意力集中在自己的业务上，不必再为程序在不同系统软件上的移植而重复工作，从而大大减少了技术上的负担。中间件带给应用系统的，不只是开发的简便、开发周期的缩短，也减少了系统的维护、运行和管理的工作量，还减少了计算机总体费用的投入。
2. 中间件需解决的问题和作用

1.3、中间件特点

1. 为解决分布异构问题，人们提出了中间件(middleware)的概念。中间件是位于平台(硬件和操作系统)和应用之间的通用服务，如下图所示，这些服务具有标准的程序接口和协议。针对不同的操作系统和硬件平台，它们可以有符合接口和协议规范的多种实现。
2. 分布式服务：

1.4、在项目中什么时候使用中间件技术

1. 在项目的架构和重构中，使用任何技术和架构的改变我们都需要谨慎斟酌和思考，因为任何技术的融入和变化都可能人员，技术，和成本的增加，中间件的技术一般现在一些互联网公司或者项目中使用比较多，如果你仅仅还只是一个初创公司建议还是使用单体架构，最多加个缓存中间件即可，不要盲目追求新或者所谓的高性能，而追求的背后一定是业务的驱动和项目的驱动，因为一旦追求就意味着你的学习成本，公司的人员结构以及服务器成本，维护和运维的成本都会增加，所以需要谨慎选择和考虑。

2、中间件技术及架构的概述

知识图谱

2.1、学习中间件的方式和技巧

1. 理解中间件在项目架构中的作用，以及各中间件的底层实现。
2. 可以使用一些类比的生活概念去理解中间件，
3. 使用一些流程图或者脑图的方式去梳理各个中间件在架构中的作用
4. 尝试用java技术去实现中间件的远离
5. 静下来去思考中间件在项目中设计的和使用的原因
6. 如果找到对应的替代总结方案
7. 尝试编写博文总结类同中间件技术的对比和使用场景。
8. 学会查看中间件的源码以及开开源项目和博文。

2.2、学习目标

1. 什么是消息中间件
2. 什么是协议
3. 什么是持久化
4. 消息分发
5. 消息的高可用
6. 消息的集群
7. 消息的容错
8. 消息的冗余

2.3、单体架构

• 在企业开发的中，大部分的初期架构都采用的是单体架构的模式进行架构，而这种架构的典型的特点：就是把所有的业务和模块，源代码，静态资源文件等都放在一个一工程中，如果其中的一个模块升级或者迭代发生一个很小变动都会重新编译和重新部署项目。 这种的架构存在的问题就是：
  1. 耦合度太高
  2. 运维的成本过高
  3. 不易维护
  4. 服务器的成本高
  5. 以及升级架构的复杂度也会增大

2.4、分布式架构

何谓分布式系统呢：

1. 通俗一点：就是一个请求由服务器端的多个服务（服务或者系统）协同处理完成。
2. 和单体架构不同的是，单体架构是一个请求发起jvm调度线程（确切的是tomcat线程池）分配线程Thread来处理请求直到释放，而分布式是系统是：一个请求是由多个系统共同来协同完成，jvm和环境都可能是独立。
3. 如果生活中的比喻的话，单体架构就想建设一个小房子很快就能够搞定，如果你要建设一个鸟巢或者大型的建筑，你就必须是各个环节的协同和分布，这样目的也是项目发展都后期的时候要去部署和思考的问题。

分布式架构系统存在的特点和问题如下：

• 存在问题
  1. 学习成本高，技术栈过多
  2. 运维成本和服务器成本增高
  3. 人员的成本也会增高
  4. 项目的负载度也会上升
  5. 面临的错误和容错性也会成倍增加
  6. 占用的服务器端口和通讯的选择的成
  7. 本高安全性的考虑和因素逼迫可能选择RMI/MQ相关的服务器端通讯。
• 好处
  1. 服务系统的独立，占用的服务器资源减少和占用的硬件成本减少，确切的说是：可以合理的分配服务资源，不造成服务器资源的浪费
  2. 系统的独立维护和部署，耦合度降低，可插拔性。
  3. 系统的架构和技术栈的选择可以变的灵活（而不是单纯的选择java）
  4. 弹性的部署，不会造成平台因部署造成的瘫痪和停服的状态。

3、基于消息中间件的分布式系统的架构

• 从上图中可以看出来，消息中间件的是
  1. 利用可靠的消息传递机制进行系统和系统直接的通讯**(与底层系统进行交换，屏蔽了操作底层系统的复杂性)。**
  2. 通过提供消息传递和消息的排队机制，它可以在分布式系统环境下扩展进程间的通讯**(模块之间能够通过某种协议及时进行通信，会屏蔽模块之间的特性，如语言等特性)。**

3.1、消息中间件应用的场景

1. 跨系统数据传递(屏蔽的语言特性，数据能够得到及时传递)

2. 高并发的流量削峰(基于第三点，数据的分法和异步处理)

3. 数据的分发和异步处理(异步编程)

   

4. 大数据分析与传递

5. 分布式事务

6. 比如你有一个数据要进行迁移或者请求并发过多的时候，比如你有10W的并发请求下订单，我们可以在这些订单入库之前，我们可以把订单请求堆积到消息队列中，让它稳健可靠的入库和执行。

3.2、常见的消息中间件

• ActiveMQ、RabbitMQ、Kafka、RocketMQ等。

3.3、消息中间件的本质及设计

它是一种接受数据，接受请求、存储数据、发送数据等功能的技术服务。

谁来生产消息(生产者(服务器))，存储消息(MQ)和消费消息(客户端)呢？

3.4 、消息中间件的核心组成部分

1. 消息的协议**(能够依靠某种协议消息的规范和通信，可能是TCP/IP协议，也可能是基于TCP/IP协议之上开发的一套协议，MQ使用的是后者)**
2. 消息的持久化机制**(能否存盘，高可用性)**
3. 消息的分发策略**(消息的分法给服务器的机制)**
4. 消息的高可用，高可靠**(是否支持集群，宕机能否有其他中间件进行处理)**
5. 消息的容错机制

4、消息队列协议

4.1、什么是协议

我们知道消息中间件负责数据的传递，存储，和分发消费三个部分，数据的存储和分发的过程中肯定要遵循某种约定成俗的规范，你是采用底层的TCP/IP，UDP协议还是其他的自己取构建等，而这些约定成俗的规范就称之为：协议。

4.2、网络协议的三要素

1. 语法。语法是用户数据与控制信息的结构与格式,以及数据出现的顺序。
2. 语义。语义是**解释控制信息每个部分的意义。**它规定了需要发出何种控制信息,以及完成的动作与做出什么样的响应。
3. 时序。时序是对事件发生顺序的详细说明。
4. 比如我MQ发送一个信息，是以什么数据格式发送到队列中，然后每个部分的含义是什么，发送完毕以后的执行的动作，以及消费者消费消息的动作，消费完毕的响应结果和反馈是什么，然后按照对应的执行顺序进行处理。如果你还是不理解：大家每天都在接触的http请求协议:

5. 而消息中间件采用的并不是http协议，而常见的消息中间件协议有：OpenWire、AMQP、MQTT、Kafka，OpenMessage协议。如：消息中间件，消息可能有些不同的信息需要存储，但http中对这些信息存储不支持，所以得定义自己需要的协议进行一些特定数据的存储，而且中间件有高可靠、持久等特性，http这些是没有这些特性的。

4.3、AMQP协议

1. AMQP：(全称：Advanced Message Queuing Protocol) 是高级消息队列协议。由摩根大通集团联合其他公司共同设计。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang(用作于交换机)中的实现有RabbitMQ等。

2. 特性：

   • 分布式事务支持。
   • 消息的持久化支持。
   • 高性能和高可靠的消息处理优势。

3. 支持者

   

4.4、MQTT协议

1. MQTT协议：（Message Queueing Telemetry Transport）消息队列是IBM开放的一个即时通讯协议，物联网系统架构中的重要组成部分。

2. 特点：

   • 轻量

     1. 结构简单
     2. 传输快，不支持事务
     3. 没有持久化设计。

   • 应用场景：

     1. 适用于计算能力有限
     2. 低带宽
     3. 网络不稳定的场景。

   • 支持者

     

4.5、OpenMessage协议

1. 是近几年由阿里、雅虎和滴滴出行、Stremalio等公司共同参与创立的分布式消息中间件、流处理等领域的应用开发标准。

2. 特点：

   • 结构简单
   • 解析速度快
   • 支持事务和持久化设计。

3. 支持者

   

4.6、Kafka协议

1. Kafka协议是基于TCP/IP的二进制协议。消息内部是通过长度来分割，由一些基本数据类型组成。

2. 特点是：

   • 结构简单
   • 解析速度快
   • 无事务支持
   • 有持久化设计

3. 支持者

   

4.7、小结

• 协议：是在tcp/ip协议基础之上构建的一种约定成俗的规范和机制、它的主要目的可以让客户端（应用程序 java，go）进行沟通和通讯。消息中间的协议具有这种规范，也必须具有持久性，高可用，高可靠的性能。

5、消息队列持久化

5.1、持久化

• 简单来说就是将数据存入磁盘，而不是存在内存中随服务器重启断开而消失，宕机了重启，数据会从磁盘中读取之前保存的数据，数据不会丢失，使数据能够永久保存。

  

5.2、常见的持久化方式

6、消息的分发策略

6.1、消息的分发策略

1. MQ消息队列有如下几个角色
   • 生产者
   • 存储消息
   • 消费者
2. 那么生产者生成消息以后，MQ进行存储，消费者是如何获取消息的呢？一般获取数据的方式无外乎推（push）或者拉（pull）两种方式，典型的git就有推拉机制，我们发送的http请求就是一种典型的拉取数据库数据返回的过程。而消息队列MQ是一种推送的过程，而这些推机制会适用到很多的业务场景也有很多对应推机制策略。

6.2、场景分析一

• 比如我在APP上下了一个订单，我们的系统和服务很多，我们如何得知这个消息被那个系统或者那些服务或者系统进行消费，那这个时候就需要一个分发的策略。这就需要消费策略。或者称之为消费的方法论。

6.3、场景分析二

• 在发送消息的过程中可能会出现异常，或者网络的抖动，故障等等因为造成消息的无法消费，比如用户在下订单，消费MQ接受，订单系统出现故障，导致用户支付失败，那么这个时候就需要消息中间件就必须支持消息重试机制策略。也就是支持：出现问题和故障的情况下，息不丢失还可以进行重发。

6.4、消息分发策略的机制和对比

7、消息队列高可用和高可靠

7.1、什么是高可用机制

1. 所谓高可用：是指产品在规定的条件和规定的时刻或时间内处于可执行规定功能状态的能力。(服务器在任何情况下，都不会出现服务不可用的情况，有服务器宕机，其他服务器能够继续处理)
2. 当业务量增加时，请求也过大，一台消息中间件服务器的会触及硬件（CPU,内存，磁盘）的极限，一台消息服务器你已经无法满足业务的需求，所以消息中间件必须支持集群部署。来达到高可用的目的。

7.2、集群模式1 - Master-slave主从共享数据的部署方式

• 解释：生产者讲消费发送到Master节点，所有的都连接这个消息队列共享这块数据区域，Master节点负责写入，一旦Master挂掉，slave节点继续服务。从而形成高可用，

7.3、集群模式2 - Master- slave主从同步部署方式

• 解释：这种模式写入消息同样在Master主节点上，但是主节点会同步数据到slave节点形成副本，和zookeeper或者redis主从机制很类同。这样可以达到负载均衡的效果，如果消费者有多个这样就可以去不同的节点就行消费，以为消息的拷贝和同步会暂用很大的带宽和网络资源。

7.4、集群模式3 - 多主集群同步部署模式

• 解释：和上面的区别不是特别的大，但是它的写入可以往任意节点去写入。

7.5、集群模式4 - 多主集群转发部署模式

• 解释：如果你插入的数据是broker-1中，元数据信息会存储数据的相关描述和记录存放的位置（队列）。
  它会对描述信息也就是元数据信息就行同步，如果消费者在broker-2中进行消费，发现自己几点没有对应的消息，可以从对应的元数据信息中去查询，然后返回对应的消息信息，场景：比如买火车票或者黄牛买演唱会门票，比如第一个黄牛有顾客说要买的演唱会门票，但是没有但是他会去联系其他的黄牛询问，如果有就返回。

7.6、集群模式5 Master-slave与Breoker-cluster组合的方案

• 解释：实现多主多从的热备机制来完成消息的高可用以及数据的热备机制，在生产规模达到一定的阶段的时候，这种使用的频率比较高。

7.7、模式总结

1. 最终目的都是为保证：消息服务器不会挂掉，出现了故障依然可以抱着消息服务继续使用。
2. 反正终归三句话：
   • 要么消息共享，
   • 要么消息同步
   • 要么元数据共享

7.7、什么是高可靠机制

1. 所谓高可用是指：是指系统可以无故障低持续运行，比如一个系统突然崩溃，报错，异常等等并不影响线上业务的正常运行，出错的几率极低，就称之为：高可靠。
2. 在高并发的业务场景中，如果不能保证系统的高可靠，那造成的隐患和损失是非常严重的。
3. 如何保证中间件消息的可靠性呢？可以从两个方面考虑：
   • 消息的传输：通过协议来保证系统间数据解析的正确性。
   • 消息的存储可靠：通过持久化来保证消息的可靠性。

二、实操

1、RabbitMQ入门及安装

1.1、概述

1. 官网：https://www.rabbitmq.com/
2. 什么是RabbitMQ,官方给出来这样的解释：

3. 简单概述：RabbitMQ是一个开源的遵循AMQP协议实现的基于Erlang语言编写，支持多种客户端（语言）。用于在分布式系统中存储消息，转发消息，具有高可用，高可扩性，易用性等特征。

1.2、安装RabbitMQ

1. 下载地址：https://www.rabbitmq.com/download.html

2. 环境准备：CentOS7.x+ / Erlang

3. RabbitMQ是采用Erlang语言开发的，所以系统环境必须提供Erlang环境，第一步就是安装Erlang。

4. erlang和RabbitMQ版本的按照比较: https://www.rabbitmq.com/which-erlang.html，必须对应版本去下载。

   

1.3、 Erlang安装

安装下载

1. 参考地址：https://www.erlang-solutions.com/downloads/

   

2. 通过下载安装包后，通过xftp进行传输,解压

   

3. 解压时如果抛出如下缺少依赖错误，安装对应的依赖：

rpm -ivh esl-erlang_23.2.1-1_centos_7_amd64.rpm
安装抛出一下错误：
警告：esl-erlang_23.2.1-1_centos_7_amd64.rpm: 头V4 RSA/SHA256 Signature, 密钥 ID a14f4fca: NOKEY
错误：依赖检测失败：
		libodbc.so.2()(64bit) 被 esl-erlang-23.2.1-1.x86_64 需要
安装依赖：
yum install unixODBC

注意：如果安装时抛出以上错误，去改网址下载对应的依赖即可，如果有其他依赖不存在，搜索依赖，安装即可。

​ https://centos.pkgs.org/7/centos-aarch64/unixODBC-2.3.1-14.el7.aarch64.rpm.html

​ 再次解压，即可解，安装即可成功：

yum install -y erlang
在进行验证：
erl -v

1.4、安装socat

1.5、安装rabbitmq

1. 解压，安装

2. 运行rabbitMq

1.6、RabbitMQ的配置

• 相关端口

• 一定要注意：RabbitMQ 在安装完毕以后，会绑定一些端口，如果你购买的是阿里云或者腾讯云相关的服务器一定要在安全组中把对应的端口添加到防火墙。

• 设置开机自启动

  

2、RabbitMQWeb管理界面及授权操作

1、RabbitMQ管理界面

• 默认情况下，rabbitmq是没有安装web端的客户端插件，需要安装才可以生效(启动之后进行安装)

rabbitmq-plugins enable rabbitmq_management

​

• 说明：rabbitmq有一个默认账号和密码是：guest 默认情况只能在localhost本机下访问，所以需要添加一个远程登录的用户。

  

1.1、安装完毕以后，重启服务即可

systemctl restart rabbitmq-server

1.2、在浏览器访问

• http://ip:15672/

  

• 如果是阿里云账号，需要将该端口号打开，设置成安全组，否则无法访问。

2、授权账号和密码

2.1、新增用户

rabbitmqctl add_user admin admin

2.1、设置用户分配操作权限

rabbitmqctl set_user_tags admin administrator

• 用户级别：

  1. administrator 可以登录控制台、查看所有信息、可以对rabbitmq进行管理
  2. monitoring 监控者 登录控制台，查看所有信息
  3. policymaker 策略制定者 登录控制台,指定策略
  4. managment 普通管理员 登录控制台

  

2.3、为用户添加资源权限

• 添加了administrator角色 就可以不用设置这一步。

rabbitmqctl.bat set_permissions -p / admin ".*" ".*" ".*"

3、输入添加的账号密码

4、小结

rabbitmqctl add_user 账号 密码
rabbitmqctl set_user_tags 账号 administrator
rabbitmqctl change_password Username Newpassword 修改密码
rabbitmqctl delete_user Username 删除用户
rabbitmqctl list_users 查看用户清单
rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*" 为用户设置administrator角色
rabbitmqctl set_permissions -p / root ".*" ".*" ".*"

3、RabbitMQ之Docker安装

docker容器技术可查看此文章：https://blog.csdn.net/weixin_43947102/article/details/120428266

1、Docker安装RabbitMQ

1.1、虚拟化容器技术—Docker的安装

（1）yum 包更新到最新
> yum update
（2）安装需要的软件包， yum-util 提供yum-config-manager功能，另外两个是devicemapper驱动依赖的
> yum install -y yum-utils device-mapper-persistent-data lvm2
（3）设置yum源为阿里云
> yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo
（4）安装docker
> yum install docker-ce -y
（5）安装后查看docker版本
> docker -v
 (6) 安装加速镜像
 sudo mkdir -p /etc/docker
 sudo tee /etc/docker/daemon.json <<-'EOF'
 {
  "registry-mirrors": ["https://0wrdwnn6.mirror.aliyuncs.com"]
 }
 EOF
 sudo systemctl daemon-reload
 sudo systemctl restart docker

1.2、docker的相关命令

# 启动docker：
systemctl start docker
# 停止docker：
systemctl stop docker
# 重启docker：
systemctl restart docker
# 查看docker状态：	
systemctl status docker
# 开机启动：  
systemctl enable dockersy
stemctl unenable docker
# 查看docker概要信息
docker info
# 查看docker帮助文档
docker --help

1.3、安装rabbitmq

参考网站：

1. https://www.rabbitmq.com/download.html

   

   

2. https://registry.hub.docker.com/_/rabbitmq/

   

1.4、获取rabbit镜像

docker pull rabbitmq:management

​

1.5、创建并运行容器

1. 运行容器

   如果直接运行，没有容器也会自动去下载

   

docker run -di --name=myrabbit -p 15672:15672 rabbitmq:management
—hostname：指定容器主机名称
—name:指定容器名称
-p:将mq端口号映射到本地或者运行时设置用户和密码 宿主机端口号:容器端口号(如果启动多台，容器端口号不要变)
- RABBITMQ_DEFAULT_USER、RABBITMQ_DEFAULT_PASS 创建容器时，创建的账号和密码
- rabbitmq:management ：图形化界面
docker run -di --name myrabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883 rabbitmq:management

docker run -di --hostname rabbit_host1 --name rabbitmq1 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15673:15672 -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie' rabbitmq:management

docker run -di --hostname rabbit_host2 --name rabbitmq2 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15674:15672 --link rabbitmq1:rabbit_host1 -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie' rabbitmq:management


docker run -d --hostname rabbit_host1 --name rabbitmq1 -p 15673:15672 -p 5672:5672 -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie' rabbitmq:management

docker run -d --hostname rabbit_host2 --name rabbitmq2 -p 5674:5672 --link rabbitmq1:rabbit_host1 -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie' rabbitmq:management

2. 注意：是否启动成功

   

3. 启动多台rabbitMq服务

   

   

4. 查看日志

docker logs -f myrabbit

1.6、容器运行正常

• 使用 http://你的IP地址:15672 访问rabbit控制台

  

2、额外Linux相关排查命令

> more xxx.log  查看日记信息
> netstat -naop | grep 5672 查看端口是否被占用
> ps -ef | grep 5672  查看进程
> systemctl stop 服务

4、RabbitMQ的角色分类

1、none

• 不能访问management plugin，无法进行rabbitMqWeb界面的登陆。

2、management：查看自己相关节点信息

• 只能查看自己的创建的交换机或者队列等。
• 列出自己可以通过AMQP登入的虚拟机
• 查看自己的虚拟机节点 virtual hosts的queues,exchanges和bindings信息
• 查看和关闭自己的channels和connections
• 查看有关自己的虚拟机节点virtual hosts的统计信息。包括其他用户在这个节点virtual hosts中的活动信息。

3、Policymaker

• 包含management所有权限
• 查看和创建和删除自己的virtual hosts所属的policies和parameters信息。

4、Monitoring

• 包含management所有权限
• 罗列出所有的virtual hosts，包括不能登录的virtual hosts。
• 查看其他用户的connections和channels信息
• 查看节点级别的数据如clustering和memory使用情况
• 查看所有的virtual hosts的全局统计信息。
• 不可对其他virtual hosts进行删除修改

5、Administrator

• 最高权限
• 可以创建和删除virtual hosts
• 可以查看，创建和删除users
• 查看创建permisssions
• 关闭所有用户的connections

6、具体操作的界面

• 退出登陆，使用创建的用户进行登陆

  

5、rabbitMq的工作模式

• 官网介绍：https://www.rabbitmq.com/getstarted.html

  

  

6、RabbitMQ入门案例 - Simple 简单模式

1、实现步骤

1. jdk1.8
2. 构建一个maven工程
3. 导入rabbitmq的maven依赖
4. 启动rabbitmq-server服务
5. 定义生产者
6. 定义消费者
7. 观察消息的在rabbitmq-server服务中的过程

2、构建一个maven工程

3、导入rabbitmq的maven依赖

3.1、Java原生依赖

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

3.2、spring依赖

<dependency>
    <groupId>org.springframework.amqp</groupId>   
    <artifactId>spring-amqp</artifactId>    
    <version>2.2.5.RELEASE</version>
</dependency>
<dependency>    
    <groupId>org.springframework.amqp</groupId>  
    <artifactId>spring-rabbit</artifactId>  
    <version>2.2.5.RELEASE</version>
</dependency>

3.3、springboot依赖

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

3.4、上面根据自己的项目环境进行选择即可。

4、启动rabbitmq-server服务

systemctl start rabbitmq-server
或者
docker start myrabbit

5、定义生产者

1. 执行发送，这个时候可以在web控制台查看到这个队列queue的信息

   /**
    * 简单模式的生产者
    * 一个生产者--->队列--->消费者
    */
   public class Producer {
       public static void main(String[] args) {
           //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
           //1、创建连接工厂
           ConnectionFactory connectionFactory = new ConnectionFactory();
           connectionFactory.setHost("169.254.118.178");
           connectionFactory.setPort(5672);
           connectionFactory.setUsername("admin");
           connectionFactory.setPassword("admin");
           //设置访问的虚拟机，/根目录
           connectionFactory.setVirtualHost("/");
           //2、创建连接的connection
           Connection connection = null;
           Channel channel = null;
           try {
               //设置连接名称
               connection = connectionFactory.newConnection("生产者");
               //3、通过连接获取通道
               channel = connection.createChannel();
               //4、创建队列
               String queueName = "queue1";
               /*
                   参数1：队列名
                   参数2：是否持久化durable=false 持久化消息是否存盘，如果false 非持久化 true持久化  非持久化会存盘嘛？
                   参数3：是否有排他性  是否独占独立，排他队列
                   参数4：是否自动删除 随着最后一个消费者消费消息以后是否把队列删除
                   参数5：是否携带额外参数，可以给消费者获取到
                */
               //非持久队列
               channel.queueDeclare(queueName, false, false, false, null);
               //持久化队列
               //channel.queueDeclare(queueName, true, false, false, null);
               //5、准备消息内容
               String message = "Hello Word";
               //6、发送消息给队列queue
                /*
                	Rabbitmq不允许创建两个相同的队列名称，否则会报错。
                   参数1：交换机名称
                   参数2：队列名称
                   参数3:消息的控制状态
                   参数4：消息的字节数
                */
               channel.basicPublish("", queueName, null, message.getBytes());
               System.out.println("消息发送成功~~");
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               //7、关闭通道
               if (channel != null && connection.isOpen()) {
                   try {
                       channel.close();
                   } catch (Exception e) {
                       e.printStackTrace();
                   }
                   //7、关闭连接
                   if (channel != null && connection.isOpen()) {
                       try {
                           connection.close();
                       } catch (IOException e) {
                           e.printStackTrace();
                       }
                   }
               }
           }
       }
   }
   

2. 出现错误：

   

3. 错误解决：

   

4. 我们可以进行对队列的消息进行预览和测试如下：通debug，查看每一步产生的结果。

   

5. 手动创建持久化队列

   

6. 进行预览和获取消息进行测试

   

7. 面试题：

   • 问：是否持久化durable=false 持久化消息是否存盘，如果false 非持久化 true持久化 非持久化会存盘嘛？
   • 答：会存盘，但是会随着重启服务会进行删除。
   • 问：可以存在没有交换机的队列嘛？
   • 答：不可能，虽然没有指定交换机，但是一定存在一个默认的交换机。

6、定义消费者

/**
 * 简单模式的消费者
 */
public class Consumer {
    public static void main(String[] args) {
        //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
        //1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("169.254.118.178");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        //设置访问的虚拟机，/根目录
        connectionFactory.setVirtualHost("/");
        //2、创建连接的connection
        Connection connection = null;
        Channel channel = null;
        try {
            //设置连接名称
            connection = connectionFactory.newConnection("生产者");
            //3、通过连接获取通道
            channel = connection.createChannel();
            //4、通过交换机，声明队列，绑定关系，路由key，发送消息，接收消息
            String queueName = "queue1";
            /*
                参数1：从哪个队列取
                参数2：是否验证
                参数3：收到消息后成功处理
                参数4：收到消息失败如何处理
             */
            channel.basicConsume(queueName, true, (s,delivery)-> {
                    System.out.println("收到的消息是：" + new String(delivery.getBody(), "utf-8"));
            }, (s) -> {
                System.out.println("接收失败。。。");
            });
            System.out.println("开始接收消息");
            //阻塞 可查看消费消息之后的队列 如果停止 队列queue1也会被删除 因为未做持久化
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //7、关闭通道
            if (channel != null && connection.isOpen()) {
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                //7、关闭连接
                if (channel != null && connection.isOpen()) {
                    try {
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

            }
        }
    }
}

7、观察消息的在rabbitmq-server服务中的过程

1. 消费者消费了一条消息，在观察队列中的情况，发现队列中的消息被消费掉了

   

   

2. 因为queue1队列是非持久化的队列，对rabbitMq服务进行重启之后，队列会自动删除，但是手动创建的队列，是持久化的队列，重启rabbitMq服务不会进行重启。

   

   

7、AMQP

• AMQP全称：Advanced Message Queuing Protocol(高级消息队列协议)。是应用层协议的一个开发标准，为面向消息的中间件设计。

• 面试题：

  问：rabbitMq为什么基于channel通道处理，而不是连接connection处理？

  答：因为connection会经过三次握手四次挥手的过程，会比较耗时间，且改连接也是长连接(不需要每一次请求都重新建立一次新连接，支持请求流水线的操作)，连接中就会有很多信道，该信道就速度很快，很方便。

1、AMQP生产者流转过程

2、AMQP消费者流转过程

8、RabbitMQ的核心组成部分

1、RabbitMQ的核心部分

• 核心概念：

  1. Server：又称Broker ,接受客户端的连接，实现AMQP实体服务。 安装rabbitmq-server

  2. Connection：连接，应用程序与Broker的网络连接 TCP/IP/ 三次握手和四次挥手

  3. Channel：网络信道，几乎所有的操作都在Channel中进行(比连接connection操作快很多)，Channel是进行消息读写的通道，客户端可以建立对各Channel，每个Channel代表一个会话任务。

  4. Message :消息：服务与应用程序之间传送的数据，由Properties和body组成，Properties可是对消息进行修饰，比如消息的优先级，延迟等高级特性，Body则就是消息体的内容。

  5. Virtual Host 虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机理由可以有若干个Exhange和Queueu，同一个虚拟主机里面不能有相同名字的Exchange

  6. Exchange：交换机，接受消息，根据路由键发送消息到绑定的队列。(不具备消息存储的能力且创建的交换机默认是路由模式)

     

  7. Bindings：Exchange和Queue之间的虚拟连接，binding中可以保护多个routing key.

  8. Routing key：是一个路由规则，虚拟机可以用它来确定如何路由一个特定消息。

  9. Queue：队列：也成为Message Queue,消息队列，保存消息并将它们转发给消费者。

2、RabbitMQ整体架构是什么样子的？

3、RabbitMQ的运行流程

4、RabbitMQ支持消息的模式

参考官网：https://www.rabbitmq.com/getstarted.html

4.1、web界面操作rabbitMq

1. 操作队列

   • 创建队列

     

   • 队列详情页面的操作

     

   • 手动发送消息

     

   • 手动获取消息，注意，如果线上的环境，请注意不要选择消费消息

     

2. 操作交换机

   • 创建交换机

     

   • 交换机详情页面

     

   • 交换机向队列中发送消息

     

3. 页面多久刷新一次

   

4.2、简单模式 Simple

1. 生成者生产消息，投递到默认的交换机中，再由交换机投递到队列中，消费者通过订阅等方式进行绑定，队列就会向绑定的消费者投递消息(图中虽然没有交换机，但是存在默认的交换机)

   

2. 手动操作web界面流程

   • 交换机向队列中发送消息

     

   • 可在队列详情中查看消息，预览消息

     

   • 消费消息

     

4.3、工作模式 Work

• 由一个生产者，和多个消费者组成，生产者投递消息到队列(中间也有默认交换机)，生产的队列通过某些分发机制分发到各个消费者。

​

• web操作查看视频

  展示效果不明显，后面以代码方式进行讲解。

• 类型：无

• 特点：分发机制

4.4、发布订阅模式 Fanout

• 生产者投递消息到交换机，交换机发送消息到绑定了当前交换机的队列，在由消费者消进行费消

  

• web操作

  1. 创建交换机

     

  2. 创建队列

     

  3. 绑定队列(如果指定了路由key，是不会生效的)

     

     

  4. 模拟发送消息，交换机发送消息

     

  5. 绑定了当前交换机的所有队列收到消息

     

• 类型：fanout

• 特点：Fanout—发布与订阅模式，是一种广播机制，它是没有路由key的模式。

4.5、路由模式 Direct

• 生产者投递消息到交换机，交换机发送消息，通过路由key进行匹配**（完全匹配）**，发送到匹配上的队列中，在由消费者消进行费消(如：发送邮件只需要通知到微信和qq，不需要通过短信的方式进行通知，即可通过路由key进行匹配通知)

  

• web操作

  1. 创建路由模式的交换机

     

  2. 绑定队列和队列的路由key

     

  3. 进行消息的发送，且指定路由key

     

  4. 查看发送的队列的消息

     

• 类型：direct

• 特点：有routing-key的匹配模式

4.6、主题模式 Topic

• 相对于路由模式而言，进行了模糊匹配的方式进行消息的分发。

  

• 匹配规则

  #：0级或者多级(一级至少有一个字符串)

  *：一级(仅此只能有一级)

  总结：*只能表示一个，#号表示一个或多个

• web操作

  1. 创建主题模式的交换机

     

  2. 绑定队列，指定路由key的匹配规则

     

  3. 发送消息，指定模糊匹配的路由key

     

  4. 查看消息，队列能够模糊匹配上的路由key收到了消息

     

• 类型：topic

• 特点：模糊的routing-key的匹配模式

4.7、参数模式 headers

• 交换机发送消息时，携带参数，能够匹配上(参数名和值都能完全匹配上)队列绑定的参数的队列能够收到消息

• web操作

  1. 创建参数模式的交换机

     

  2. 绑定队列和队列的参数

     

  3. 发送消息，指定参数

     

  4. 结果

     

• 类型：headers

• 特点：参数匹配模式

5、小结

• rabbitmq发送消息一定有一个交换机(没有指定交换机的队列会默认的指定一个交换机)

9、RabbitMQ入门案例 - fanout模式

1、生产者

1. 以fanout_exchange交换机为例(已经已图形化界面的形式创建好了，绑定的队列也以图形化界面的形式绑定好了)

   

2. 生产者的代码

   /**
    * fanout发布订阅模式的生产者
    */
   public class Producer {
       public static void main(String[] args) {
           //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
           //1、创建连接工厂
           ConnectionFactory connectionFactory = new ConnectionFactory();
           connectionFactory.setHost("169.254.118.178");
           connectionFactory.setPort(5672);
           connectionFactory.setUsername("admin");
           connectionFactory.setPassword("admin");
           //设置访问的虚拟机，/根目录
           connectionFactory.setVirtualHost("/");
           //2、创建连接的connection
           Connection connection = null;
           Channel channel = null;
           try {
               //设置连接名称
               connection = connectionFactory.newConnection("生产者");
               //3、通过连接获取通道
               channel = connection.createChannel();
               //4、准备消息内容
               String message = "Hello Word fanout_exchange";
               //5、准备交换机(代码中未展示交换机和队列之间的绑定，已经以图形化界面的形式进行了绑定，节省了代码的编写)
               String exchangName = "fanout_exchange";
               //6、定义路由key 为空 因为fanout模式不需要指定路由key
               String routeKey = "";
               //7、发送消息给队列queue
                /*
                   参数1：交换机名称
                   参数2：路由key
                   参数3:消息的控制状态
                   参数4：消息的字节数
                */
               channel.basicPublish(exchangName, routeKey, null, message.getBytes());
               System.out.println("消息发送成功~~");
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               //8关闭通道
               if (channel != null && connection.isOpen()) {
                   try {
                       channel.close();
                   } catch (Exception e) {
                       e.printStackTrace();
                   }
                   //9、关闭连接
                   if (channel != null && connection.isOpen()) {
                       try {
                           connection.close();
                       } catch (IOException e) {
                           e.printStackTrace();
                       }
                   }
               }
           }
       }
   }
   

3. 运行结果

   

2、消费者

1. 消费者代码

   /**
    * fanout发布订阅模式的消费者
    */
   public class Consumer {
   
       private static Runnable runnable = new Runnable() {
           @Override
           public void run() {
               //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
               //1、创建连接工厂
               ConnectionFactory connectionFactory = new ConnectionFactory();
               connectionFactory.setHost("169.254.118.178");
               connectionFactory.setPort(5672);
               connectionFactory.setUsername("admin");
               connectionFactory.setPassword("admin");
               //设置访问的虚拟机，/根目录
               connectionFactory.setVirtualHost("/");
               //2、创建连接的connection
               Connection connection = null;
               Channel channel = null;
               try {
                   //设置连接名称
                   connection = connectionFactory.newConnection("生产者");
                   //3、通过连接获取通道
                   channel = connection.createChannel();
                   //4、通过交换机，声明队列，绑定关系，路由key，发送消息，接收消息
                   //当前线程为取出的队列名
                   String queueName = Thread.currentThread().getName();
               /*
                   会把队列中所有的消息全部接收掉
                   参数1：从哪个队列取
                   参数2：是否验证
                   参数3：收到消息后成功处理
                   参数4：收到消息失败如何处理
                */
                   channel.basicConsume(queueName, true, (s, delivery) -> {
                       System.out.println(queueName + "收到的消息是：" + new String(delivery.getBody(), "utf-8"));
                   }, (s) -> {
                       System.out.println("接收失败。。。");
                   });
                   System.out.println("开始接收消息");
                    //消费者不停止 一直监听队列
                   System.in.read();
               } catch (Exception e) {
                   e.printStackTrace();
               } finally {
                   //7、关闭通道
                   if (channel != null && connection.isOpen()) {
                       try {
                           channel.close();
                       } catch (Exception e) {
                           e.printStackTrace();
                       }
                       //7、关闭连接
                       if (channel != null && connection.isOpen()) {
                           try {
                               connection.close();
                           } catch (IOException e) {
                               e.printStackTrace();
                           }
                       }
   
                   }
               }
           }
       };
   
       public static void main(String[] args) {
           //以多线程的形式对不同队列进行读取
           new Thread(runnable, "queue1").start();
           new Thread(runnable, "queue2").start();
           new Thread(runnable, "queue3").start();
           new Thread(runnable, "queue4").start();
       }
   }
   

2. 运行结果

   、

3、注意事项

1. 如果监听的队列不存在，启动消费者会抛出异常，不存在的队列不会自动创建，需要自己手动创建。
2. 如果绑定的交换机不存在，也会抛出异常，不会为你自动创建，需要通过代码的方式创建。

10、RabbitMQ入门案例 - Direct模式

1、生产者

1. 交换机与队列绑定关系和队列绑定路由key的关系

   

2. 消费者代码(只更改了交换机的名称、内容和路由key，其余相同)

   /**
    * direct路由key模式的生产者
    */
   public class Producer {
       public static void main(String[] args) {
           //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
           //1、创建连接工厂
           ConnectionFactory connectionFactory = new ConnectionFactory();
           connectionFactory.setHost("169.254.118.178");
           connectionFactory.setPort(5672);
           connectionFactory.setUsername("admin");
           connectionFactory.setPassword("admin");
           //设置访问的虚拟机，/根目录
           connectionFactory.setVirtualHost("/");
           //2、创建连接的connection
           Connection connection = null;
           Channel channel = null;
           try {
               //设置连接名称
               connection = connectionFactory.newConnection("生产者");
               //3、通过连接获取通道
               channel = connection.createChannel();
               //4、准备消息内容
               String message = "Hello Word direct_exchange";
               //5、准备交换机(代码中未展示交换机和队列之间的绑定，已经以图形化界面的形式进行了绑定，节省了代码的编写)
               String exchangName = "direct_exchange";
               //6、定义路由key 指定需要匹配的路由key
               String routeKey = "email";
               //7、发送消息给队列queue
                /*
                   参数1：交换机名称
                   参数2：路由key
                   参数3:消息的控制状态
                   参数4：消息的字节数
                */
               channel.basicPublish(exchangName, routeKey, null, message.getBytes());
               System.out.println("消息发送成功~~");
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               //8关闭通道
               if (channel != null && connection.isOpen()) {
                   try {
                       channel.close();
                   } catch (Exception e) {
                       e.printStackTrace();
                   }
                   //9、关闭连接
                   if (channel != null && connection.isOpen()) {
                       try {
                           connection.close();
                       } catch (IOException e) {
                           e.printStackTrace();
                       }
                   }
               }
           }
       }
   }
   

3. 发送消息后的结果

   

2、消费者

1. 消费者代码与上面一样，无需更改

   /**
    * direct路由key模式的消费者
    */
   public class Consumer {
   
       private static Runnable runnable = new Runnable() {
           @Override
           public void run() {
               //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
               //1、创建连接工厂
               ConnectionFactory connectionFactory = new ConnectionFactory();
               connectionFactory.setHost("169.254.118.178");
               connectionFactory.setPort(5672);
               connectionFactory.setUsername("admin");
               connectionFactory.setPassword("admin");
               //设置访问的虚拟机，/根目录
               connectionFactory.setVirtualHost("/");
               //2、创建连接的connection
               Connection connection = null;
               Channel channel = null;
               try {
                   //设置连接名称
                   connection = connectionFactory.newConnection("生产者");
                   //3、通过连接获取通道
                   channel = connection.createChannel();
                   //4、通过交换机，声明队列，绑定关系，路由key，发送消息，接收消息
                   // 当前线程为取出的队列名
                   String queueName = Thread.currentThread().getName();
               /*
                   会把队列中所有的消息全部接收掉
                   参数1：从哪个队列取
                   参数2：是否验证
                   参数3：收到消息后成功处理
                   参数4：收到消息失败如何处理
                */
                   channel.basicConsume(queueName, true, (s, delivery) -> {
                       System.out.println(queueName + "收到的消息是：" + new String(delivery.getBody(), "utf-8"));
                   }, (s) -> {
                       System.out.println("接收失败。。。");
                   });
                   System.out.println("开始接收消息");
                   //消费者不停止 一直监听队列
                   System.in.read();
               } catch (Exception e) {
                   e.printStackTrace();
               } finally {
                   //7、关闭通道
                   if (channel != null && connection.isOpen()) {
                       try {
                           channel.close();
                       } catch (Exception e) {
                           e.printStackTrace();
                       }
                       //7、关闭连接
                       if (channel != null && connection.isOpen()) {
                           try {
                               connection.close();
                           } catch (IOException e) {
                               e.printStackTrace();
                           }
                       }
   
                   }
               }
           }
       };
   
       public static void main(String[] args) {
           //以多线程的形式对不同队列进行读取
           new Thread(runnable, "queue1").start();
           new Thread(runnable, "queue2").start();
           new Thread(runnable, "queue3").start();
           new Thread(runnable, "queue4").start();
       }
   }
   

2. 结果

   

11、RabbitMQ入门案例 - Topic模式

1、生产者

1. topic_exchange交换机绑定的队列以及队列绑定的路由key

   

2. 消费者代码(只更改了交换机的名称、内容和路由key，其余相同)

   /**
    * topic路由key模式的生产者
    */
   public class Producer {
       public static void main(String[] args) {
           //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
           //1、创建连接工厂
           ConnectionFactory connectionFactory = new ConnectionFactory();
           connectionFactory.setHost("169.254.118.178");
           connectionFactory.setPort(5672);
           connectionFactory.setUsername("admin");
           connectionFactory.setPassword("admin");
           //设置访问的虚拟机，/根目录
           connectionFactory.setVirtualHost("/");
           //2、创建连接的connection
           Connection connection = null;
           Channel channel = null;
           try {
               //设置连接名称
               connection = connectionFactory.newConnection("生产者");
               //3、通过连接获取通道
               channel = connection.createChannel();
               //4、准备消息内容
               String message = "Hello Word topic_exchange";
               //5、准备交换机(代码中未展示交换机和队列之间的绑定，已经以图形化界面的形式进行了绑定，节省了代码的编写)
               String exchangName = "topic_exchange";
               //6、定义路由key 指定需要模糊匹配的路由key
               String routeKey = "com.order.test.xxx";
               //7、发送消息给队列queue
                /*
                   参数1：交换机名称
                   参数2：路由key
                   参数3:消息的控制状态
                   参数4：消息的字节数
                */
               channel.basicPublish(exchangName, routeKey, null, message.getBytes());
               System.out.println("消息发送成功~~");
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               //8关闭通道
               if (channel != null && connection.isOpen()) {
                   try {
                       channel.close();
                   } catch (Exception e) {
                       e.printStackTrace();
                   }
                   //9、关闭连接
                   if (channel != null && connection.isOpen()) {
                       try {
                           connection.close();
                       } catch (IOException e) {
                           e.printStackTrace();
                       }
                   }
               }
           }
       }
   }
   

3. 运行结果

   [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3hyvyNaG-1642383413223)(C:\Users\彭俊\Desktop\NOTE_JVM-main\rabbitMQhttps://gitee.com/pjgitee/note/raw/master/rabbitMq%E7%AC%94%E8%AE%B0.assets/image-20220109174801054.png)]

2、消费者

1. 消费者代码与上面一样，无需更改

   /**
    * direct路由key模式的消费者
    */
   public class Consumer {
   
       private static Runnable runnable = new Runnable() {
           @Override
           public void run() {
               //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
               //1、创建连接工厂
               ConnectionFactory connectionFactory = new ConnectionFactory();
               connectionFactory.setHost("169.254.118.178");
               connectionFactory.setPort(5672);
               connectionFactory.setUsername("admin");
               connectionFactory.setPassword("admin");
               //设置访问的虚拟机，/根目录
               connectionFactory.setVirtualHost("/");
               //2、创建连接的connection
               Connection connection = null;
               Channel channel = null;
               try {
                   //设置连接名称
                   connection = connectionFactory.newConnection("生产者");
                   //3、通过连接获取通道
                   channel = connection.createChannel();
                   //4、通过交换机，声明队列，绑定关系，路由key，发送消息，接收消息
                   // 当前线程为取出的队列名
                   String queueName = Thread.currentThread().getName();
               /*
                   会把队列中所有的消息全部接收掉
                   参数1：从哪个队列取
                   参数2：是否验证
                   参数3：收到消息后成功处理
                   参数4：收到消息失败如何处理
                */
                   channel.basicConsume(queueName, true, (s, delivery) -> {
                       System.out.println(queueName + "收到的消息是：" + new String(delivery.getBody(), "utf-8"));
                   }, (s) -> {
                       System.out.println("接收失败。。。");
                   });
                   System.out.println("开始接收消息");
                   //消费者不停止 一直监听队列
                   System.in.read();
               } catch (Exception e) {
                   e.printStackTrace();
               } finally {
                   //7、关闭通道
                   if (channel != null && connection.isOpen()) {
                       try {
                           channel.close();
                       } catch (Exception e) {
                           e.printStackTrace();
                       }
                       //7、关闭连接
                       if (channel != null && connection.isOpen()) {
                           try {
                               connection.close();
                           } catch (IOException e) {
                               e.printStackTrace();
                           }
                       }
   
                   }
               }
           }
       };
       public static void main(String[] args) {
           //以多线程的形式对不同队列进行读取
           new Thread(runnable, "queue1").start();
           new Thread(runnable, "queue2").start();
           new Thread(runnable, "queue3").start();
           new Thread(runnable, "queue4").start();
       }
   }
   

2. 结果

   

12、代码方式创建交换机和队列

1. 创建交换机和队列代码

   /**
    * 创建换机和队列
    */
   public class Producer {
       public static void main(String[] args) {
           //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
           //1、创建连接工厂
           ConnectionFactory connectionFactory = new ConnectionFactory();
           connectionFactory.setHost("169.254.118.178");
           connectionFactory.setPort(5672);
           connectionFactory.setUsername("admin");
           connectionFactory.setPassword("admin");
           //设置访问的虚拟机，/根目录
           connectionFactory.setVirtualHost("/");
           //2、创建连接的connection
           Connection connection = null;
           Channel channel = null;
           try {
               //设置连接名称
               connection = connectionFactory.newConnection("生产者");
               //3、通过连接获取通道
               channel = connection.createChannel();
               //4、准备消息内容
               String message = "Hello Word create_exchange";
               //5、准备交换机(代码中未展示交换机和队列之间的绑定，已经以图形化界面的形式进行了绑定，节省了代码的编写)
               String exchangeName = "create_exchange";
               //6、定义路由key 指定需要匹配的路由key
               String routeKey = "order";
               //7、定义交换机的类型
               String exchangeType = "direct";
               //创建交换机  参数1：交换机名称  参数2：交换机类型  参数3：是否持久化
               channel.exchangeDeclare(exchangeName, exchangeType, true);
               /**
                * Rabbitmq不允许创建两个相同的队列名称，否则会报错。
                *                 参数1：队列名
                *                 参数2：是否持久化durable=false 持久化消息是否存盘，如果false 非持久化 true持久化  非持久化会存盘嘛？
                *                 参数3：是否有排他性  是否独占独立，排他队列
                *                 参数4：是否自动删除 随着最后一个消费者消费消息以后是否把队列删除
                *                 参数5：是否携带额外参数   如果携带了参数 就是参数模式 headers的队列
                */
               //8、创建队列
               channel.queueDeclare("queue5", true, false, false, null);
               channel.queueDeclare("queue6", true, false, false, null);
               channel.queueDeclare("queue7", true, false, false, null);
               //9、绑定交换机和队列之间的关系 参数1：队列名  参数2：交换机名  参数3：路由key
               channel.queueBind("queue5", exchangeName, "order");
               channel.queueBind("queue6", exchangeName, "order");
               channel.queueBind("queue7", exchangeName, "course");
   
               //10、发送消息给队列queue
                /*
                   参数1：交换机名称
                   参数2：路由key
                   参数3:消息的控制状态
                   参数4：消息的字节数
                */
               channel.basicPublish(exchangeName, routeKey, null, message.getBytes());
               System.out.println("消息发送成功~~");
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               //11、关闭通道
               if (channel != null && connection.isOpen()) {
                   try {
                       channel.close();
                   } catch (Exception e) {
                       e.printStackTrace();
                   }
                   //12、关闭连接
                   if (channel != null && connection.isOpen()) {
                       try {
                           connection.close();
                       } catch (IOException e) {
                           e.printStackTrace();
                       }
                   }
               }
           }
       }
   
   

2. 运行结果

   

   

3. 消费者进行消费

   public class Consumer {
       private static Runnable runnable = new Runnable() {
           @Override
           public void run() {
               //所有的中间件技术都是基于tcp/ip协议基础之上构建的新型的协议规范，只不过rabbitMq遵循的是amqp  都需要配置 ip port
               //1、创建连接工厂
               ConnectionFactory connectionFactory = new ConnectionFactory();
               connectionFactory.setHost("169.254.118.178");
               connectionFactory.setPort(5672);
               connectionFactory.setUsername("admin");
               connectionFactory.setPassword("admin");
               //设置访问的虚拟机，/根目录
               connectionFactory.setVirtualHost("/");
               //2、创建连接的connection
               Connection connection = null;
               Channel channel = null;
               try {
                   //设置连接名称
                   connection = connectionFactory.newConnection("生产者");
                   //3、通过连接获取通道
                   channel = connection.createChannel();
                   //4、通过交换机，声明队列，绑定关系，路由key，发送消息，接收消息
                   // 当前线程为取出的队列名
                   String queueName = Thread.currentThread().getName();
               /*
                   会把队列中所有的消息全部接收掉
                   参数1：从哪个队列取
                   参数2：是否验证
                   参数3：收到消息后成功处理
                   参数4：收到消息失败如何处理
                */
                   channel.basicConsume(queueName, true, (s, delivery) -> {
                       System.out.println(queueName + "收到的消息是：" + new String(delivery.getBody(), "utf-8"));
                   }, (s) -> {
                       System.out.println("接收失败。。。");
                   });
                   System.out.println("开始接收消息");
                   //消费者不停止 一直监听队列
                   System.in.read();
               } catch (Exception e) {
                   e.printStackTrace();
               } finally {
                   //7、关闭通道
                   if (channel != null && connection.isOpen()) {
                       try {
                           channel.close();
                       } catch (Exception e) {
                           e.printStackTrace();
                       }
                       //7、关闭连接
                       if (channel != null && connection.isOpen()) {
                           try {
                               connection.close();
                           } catch (IOException e) {
                               e.printStackTrace();
                           }
                       }
   
                   }
               }
           }
       };
   
       public static void main(String[] args) {
           //以多线程的形式对不同队列进行读取
           new Thread(runnable, "queue5").start();
           new Thread(runnable, "queue6").start();
           new Thread(runnable, "queue7").start();
       }
   }
   

   

4. 注意：交换机的创建、绑定队列，队列的创建等，可以在消费者进行，也可以在生产者进行，两者皆可，但是没有的交换机去发送消息或者没有的队列取监听消息，都会抛出异常。

13、RabbitMQ入门案例 - Work模式

1、、Work模式

1. 图解：一个生产者发送消息，由多个消费者消费，但消息不会被重复消费。

   

2. 当有多个消费者时，我们的消息会被哪个消费者消费呢，我们又该如何均衡消费者消费信息的多少呢?

3. 主要有两种模式：

   • 轮询模式的分发：一个消费者一条，按均分配；
   • 公平分发：根据消费者的消费能力进行公平分发，处理快的处理的多，处理慢的处理的少；按劳分配；

2、Work模式 - 轮询模式（Round-Robin）

1. 生产者代码

   /**
    * work工作模式的轮训策略的生产者
    * @author shkstart
    * @date 2022/1/15 - 12:03
    */
   public class Producer {
       public static void main(String[] args) {
           //1、创建连接工厂
           ConnectionFactory connectionFactory = new ConnectionFactory();
           //2、设置连接属性
           connectionFactory.setHost("169.254.118.178");
           connectionFactory.setPort(5672);
           connectionFactory.setVirtualHost("/");
           connectionFactory.setUsername("admin");
           connectionFactory.setPassword("admin");
   
           Connection connection = null;
           Channel channel = null;
           try {
               //3、从连接工厂获取连接
               connection = connectionFactory.newConnection();
               //4、从连接中获取通道
               channel = connection.createChannel();
               //5、准备发送20条消息
               for (int i = 1; i <= 20; i++) {
                   //消息内容
                   String msg = "hello "+i;
                   //7、发送消息
                   // @params1: 交换机exchange  没指定交换机有一个默认的交换机
                   // @params2: 队列名称/routingkey
                   // @params3: 属性配置
                   // @params4: 发送消息的内容
                   channel.basicPublish("", "queue1", null, msg.getBytes());
                   //1ms发送一条
                   Thread.sleep(100);
               }
               System.out.println("发送消息成功");
   
           }catch (Exception e){
               e.printStackTrace();
               System.out.println("发送消息是失败");
           }finally {
               // 7: 释放连接关闭通道
               if (channel != null && channel.isOpen()) {
                   try {
                       channel.close();
                   } catch (Exception ex) {
                       ex.printStackTrace();
                   }
               }
               if (connection != null) {
                   try {
                       connection.close();
                   } catch (Exception ex) {
                       ex.printStackTrace();
                   }
               }
           }
       }
   }
   

2. 模拟两个生产者，代码(复制一份，改个名称)

   /**
    * work工作模式的轮训策略的消费者
    *
    * @author shkstart
    * @date 2022/1/15 - 12:03
    */
   public class Work1 {
       public static void main(String[] args) {
           //1、创建连接工厂
           ConnectionFactory connectionFactory = new ConnectionFactory();
           //2、设置连接属性
           connectionFactory.setHost("169.254.118.178");
           connectionFactory.setVirtualHost("/");
           connectionFactory.setUsername("admin");
           connectionFactory.setPassword("admin");
   
           Connection connection = null;
           Channel channel = null;
   
           try {
               //3、从连接工厂获取连接
               connection = connectionFactory.newConnection();
               //4、从连接中获取通道
               channel = connection.createChannel();
               //5、声明队列queue消费消息
               /*
                *  如果队列不存在，则会创建
                *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
                *
                *  @params1： queue 队列的名称
                *  @params2： durable 队列是否持久化
                *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
                *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
                *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
                * */
               // 这里如果queue已经被创建过一次了，可以不需要定义
               //channel.queueDeclare("queue1", false, false, false, null);
   
               // 同一时刻，服务器只会推送一条消息给消费者
               // 6、 定义接受消息的回调
               Channel finalChannel = channel;
               //参数2：应答机制 true自动应答(可能会有死循环的问题存在)  false手动应答
               finalChannel.basicConsume("queue1", true, new DeliverCallback() {
                   @Override
                   public void handle(String s, Delivery delivery) throws IOException {
                       try {
                           //队列名称
                           //System.out.println("s = " + s);
                           System.out.println("Work-1收到消息" + new String(delivery.getBody(), "utf-8"));
                           //休息2s 模拟业务理速度  可以在消费者2时时间改短点  模拟业务处理快慢
                           Thread.sleep(2000);
                       } catch (Exception e) {
                           e.printStackTrace();
                       } finally {
   
                       }
                   }
               }, new CancelCallback() {
                   @Override
                   public void handle(String s) throws IOException {
   
                   }
               });
               System.out.println("Work1-开始接受消息");
               //一直监听
               System.in.read();
   
           } catch (Exception e) {
               e.printStackTrace();
               System.out.println("发送消息出现异常...");
           } finally {
               // 7: 释放连接关闭通道
               if (channel != null && channel.isOpen()) {
                   try {
                       channel.close();
                   } catch (Exception ex) {
                       ex.printStackTrace();
                   }
               }
               if (connection != null && connection.isOpen()) {
                   try {
                       connection.close();
                   } catch (Exception ex) {
                       ex.printStackTrace();
                   }
               }
           }
       }
   }
   

3. 一次启动两个生产者，在启动消费者，查看控制台结果

   

4. 并不会因为处理速度而分法不均匀，work1和work2的消息处理能力不同，但是最后处理的消息条数相同，是“按均分配”。

   

3、Work模式 - 公平分发（Fair Dispatch）

1. 特点：由于消息接收者处理消息的能力不同，存在处理快慢的问题，我们就需要能者多劳，处理快的多处理，处理慢的少处理

2. 生产者代码跟上面一样

3. 消费者代码(work2雷同)

   /**
    * work工作模式的公平策略的消费者
    *
    * @author shkstart
    * @date 2022/1/15 - 12:03
    */
   public class Work1 {
       public static void main(String[] args) {
           //1、创建连接工厂
           ConnectionFactory connectionFactory = new ConnectionFactory();
           //2、设置连接属性
           connectionFactory.setHost("169.254.118.178");
           connectionFactory.setVirtualHost("/");
           connectionFactory.setUsername("admin");
           connectionFactory.setPassword("admin");
   
           Connection connection = null;
           Channel channel = null;
   
           try {
               //3、从连接工厂获取连接
               connection = connectionFactory.newConnection();
               //4、从连接中获取通道
               channel = connection.createChannel();
               //5、声明队列queue消费消息
               /*
                *  如果队列不存在，则会创建
                *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
                *
                *  @params1： queue 队列的名称
                *  @params2： durable 队列是否持久化
                *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
                *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
                *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
                * */
               // 这里如果queue已经被创建过一次了，可以不需要定义
               //channel.queueDeclare("queue1", false, false, false, null);
   
               // 同一时刻，服务器只会推送一条消息给消费者
               // 6、 定义接受消息的回调
               Channel finalChannel = channel;
               //公平分法的必须设置的一项  默认为null 指该消费者在接收到队列里的消息但没有返回确认结果之前,它不会将新的消息分发给它。  同一时刻服务器只会发一条消息给消费者(能者多劳模式)，空闲多的消费者，消费更多的消息  根据自己服务器内存、磁盘大小情况去设置大小，不要太大
               finalChannel.basicQos(1);
               //参数2：应答机制 true自动应答(可能会有死循环的问题存在)  false手动应答 公平分法必须为false  设置为false需要手动进行应答
               finalChannel.basicConsume("queue1", false, new DeliverCallback() {
                   @Override
                   public void handle(String s, Delivery delivery) throws IOException {
                       try {
                           //队列名称
                           System.out.println("s = " + s);
                           System.out.println("Work-1收到消息" + new String(delivery.getBody(), "utf-8"));
                           //手动应答
                           // channel.basicQos(1);和channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);是配套使用，只有在channel.basicQos被使用的时候channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false)才起到作用。
                           finalChannel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                           //休息2s 模拟业务理速度
                           Thread.sleep(2000);
                       } catch (Exception e) {
                           e.printStackTrace();
                       } finally {
   
                       }
                   }
               }, new CancelCallback() {
                   @Override
                   public void handle(String s) throws IOException {
   
                   }
               });
               System.out.println("Work1-开始接受消息");
               //一直监听
               System.in.read();
           } catch (Exception e) {
               e.printStackTrace();
               System.out.println("发送消息出现异常...");
           } finally {
               // 7: 释放连接关闭通道
               if (channel != null && channel.isOpen()) {
                   try {
                       channel.close();
                   } catch (Exception ex) {
                       ex.printStackTrace();
                   }
               }
               if (connection != null && connection.isOpen()) {
                   try {
                       connection.close();
                   } catch (Exception ex) {
                       ex.printStackTrace();
                   }
               }
           }
       }
   }
   

4. 消费者接收消息的速度会根据服务器处理速度的不同而接收的消息条数不同

   

5. 总结

   1. 从结果可以看到，消费者2在相同时间内，处理了更多的消息；以上代码我们实现了公平分发模式；
   2. 消费者一次接收一条消息，代码channel.BasicQos(0, 1, false);
   3. 公平分发需要消费者开启手动应答，关闭自动应答
   4. 关闭自动应答代码channel.BasicConsume(“queue_test”, false, consumer);
   5. 消费者开启手动应答代码：channel.BasicAck(ea.DeliveryTag, false);

4、总结

1. 当队列里消息较多时，我们通常会开启多个消费者处理消息；公平分发和轮询分发都是我们经常使用的模式。
2. 轮询分发的主要思想是“按均分配”，不考虑消费者的处理能力，所有消费者均分；这种情况下，处理能力弱的服务器，一直都在处理消息，而处理能力强的服务器，在处理完消息后，处于空闲状态；
3. 公平分发的主要思想是”能者多劳”，按需分配，能力强的干的多。

14、RabbitMQ使用场景

面试题某某技术的使用场景模块：

​ 在某公司进行入职时，当时的业务员不是很庞大，使用的是单体架构，但是随着业务的增大，需求的增多，需要开始对项目进行拆分，变成了一个分布式架构，就有个业务，需要与另一个业务协同进行，所以需要选用一个消息队列，有很多种消息队列，但最后选择了rabbitMq，给我的感受就是异步的，多线程、分法机制，因为异步，所以处理的速度更加高效、稳健。

1、解耦、削峰、异步

1、同步异步的问题（串行）

• 串行方式：将订单信息写入数据库成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端

  

public void makeOrder(){
    // 1 :保存订单 
    orderService.saveOrder();
    // 2： 发送短信服务
    messageService.sendSMS("order");//1-2 s
    // 3： 发送email服务
    emailService.sendEmail("order");//1-2 s
    // 4： 发送APP服务
    appService.sendApp("order");    
}

2、并行方式 异步线程池

• 并行方式：将订单信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间

  

  public void makeOrder(){
      // 1 :保存订单 
      orderService.saveOrder();
     // 相关发送
     relationMessage();
  }
  public void relationMessage(){
      // 异步
       theadpool.submit(new Callable<Object>{
           public Object call(){
               // 2： 发送短信服务  
               messageService.sendSMS("order");
           }
       })
      // 异步
       theadpool.submit(new Callable<Object>{
           public Object call(){
                // 3： 发送email服务
              emailService.sendEmail("order");
           }
       })
        // 异步
       theadpool.submit(new Callable<Object>{
           public Object call(){
               // 4： 发送短信服务
               appService.sendApp("order");
           }
       })
        // 异步
           theadpool.submit(new Callable<Object>{
           public Object call(){
               // 4： 发送短信服务
               appService.sendApp("order");
           }
       })
  }
  

  • 存在问题：
    1. 耦合度高
    2. 需要自己写线程池自己维护成本太高
    3. 出现了消息可能会丢失，需要你自己做消息补偿
    4. 如何保证消息的可靠性你自己写
    5. 如果服务器承载不了，你需要自己去写高可用

3、异步消息队列的方式

• 好处

  1. 完全解耦，用MQ建立桥接

  2. 有独立的线程池和运行模型

  3. 出现了消息可能会丢失，MQ有持久化功能

  4. 如何保证消息的可靠性，死信队列和消息转移的等

  5. 如果服务器承载不了，你需要自己去写高可用，HA镜像模型高可用。

  6. 按照以上约定，用户的响应时间相当于是订单信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了两倍

     public void makeOrder(){
         // 1 :保存订单 
         orderService.saveOrder();   
         rabbitTemplate.convertSend("ex","2","消息内容");
     }
     

2、高内聚，低耦合

3、流量的削峰

4、其他场景

1. 分布式事务的可靠消费和可靠生产
2. 索引、缓存、静态化处理的数据同步
3. 流量监控
4. 日志监控（ELK）：通过日志文件，将日志记录在消息队列中，可能线上的日志文件不可查看，也不可通过控制台去查看，导致日志很难看清楚明白。
5. 下单、订单分发、抢票

15、RabbitMQ-SpringBoot案例 -fanout模式

1、创建springboot工程,导入依赖

<!--rabbitMq的依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、整体核心

3、目标

• 使用springboot完成rabbitmq的消费模式-Fanout

  

4、实现步骤

1、定义生产者

1. 导入依赖，编写application.yaml配置

   server:
     port: 8080
   
   #配置rabbitMq的基本设置
   spring:
     rabbitmq:
       host: 169.254.118.178
       username: admin
       password: admin
       virtual-host: /
   

2. 编写模拟下订单的功能，进行消息发送

   /**
    * 订单生成，通知各个服务
    * @author shkstart
    * @date 2022/1/15 - 14:38
    */
   @Service
   public class OrderService {
       //处理rabbitMq的对象
       @Autowired
       private RabbitTemplate rabbitTemplate;
       //交换机
       String exchangeName = "fanout_order_exchange";
       //路由key fanout模式没有路由key
       String routeKey = "";
   
       //模拟用户下单
       public void makeOrder(String  id , String  productId,int num){
           //根据商品id查询库存是否充足
           //保存订单
           String orderId = UUID.randomUUID().toString();
           System.out.println("订单生成成功："+orderId);
           //通过MQ来完成消息的分法
           //参数1：交换机名称 参数2：路由key/queue队列的名称  参数3：消息的内容
           rabbitTemplate.convertAndSend(exchangeName, routeKey,orderId);
   
       }
   }
   

3. 编写配置类，创建队列和交换机，并绑定关系

   /**
    * rabbitMq配置类 绑定交换机和队列之间的关系
    * @author shkstart
    * @date 2022/1/15 - 14:48
    */
   @Configuration
   public class RabbitMqConfig {
       /**
        * 创建一个fanout模式的交换机且是持久化的
        * @return
        */
       @Bean
       public FanoutExchange fanoutExchange(){
           return new FanoutExchange("fanout_order_exchange", true, false);
       }
       /**
        * 创建一个持久化的队列
        * @return
        */
       @Bean
       public Queue smsQueue(){
           /**
            durable:是否持久化,默认是false, 持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
            exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
            autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
            return new Queue("TestDirectQueue",true,true,false);
           一般设置一下队列的持久化就好,其余两个就是默认false
            */
           return new Queue("sms.fanout.queue", true);
       }
   
       /**
        * 创建一个持久化的队列
        * @return
        */
       @Bean
       public Queue emailQueue(){
           return new Queue("email.fanout.queue", true);
       }
   
       /**
        * 创建一个持久化的队列
        * @return
        */
       @Bean
       public Queue shortMessageQueue(){
           return new Queue("shortMessage.fanout.queue", true);
       }
   
       /**
        * 交换机绑定对应的队列
        * @return
        */
       @Bean
       public Binding smsBindingBuilder(){
           //绑定队列到交换机上
           return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
       }
       /**
        * 交换机绑定对应的队列
        * @return
        */
       @Bean
       public Binding eamilBindingBuilder(){
           //绑定队列到交换机上
           return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
       }
       /**
        * 交换机绑定对应的队列
        * @return
        */
       @Bean
       public Binding shortMessageBuilder(){
           //绑定队列到交换机上
           return BindingBuilder.bind(shortMessageQueue()).to(fanoutExchange());
       }
   }
   
   

4. 编写测试类

   @Autowired
   private OrderService orderService;
   
   @Test
   void contextLoads() {
       orderService.makeOrder("1", "1", 12);
   }
   

5. 运行测试类，查看和观察web控制台的状况配置类

   

   

   

2、定义消费者

1. 导入rabbitMq依赖和web场景依赖，编写application.yaml配置

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

   server:
     port: 8081
   
   #配置rabbitMq的基本设置
   spring:
     rabbitmq:
       host: 169.254.118.178
       username: admin
       password: admin
       virtual-host: /
   

2. 编写接收消息方法 (其余两个消费者雷同)

   /**
    * 邮件的消费者
    */
   @RabbitListener(queues = "email.fanout.queue")  //指定监听的交换机
   @Component //注入到容器中
   public class EmailConsumer {
   
       @RabbitHandler //标识为了一个监听消息的方法  监听到的消息会自动注入到message中
       public void reviceMessage(String message){
           System.out.println("邮件的fonout----->接收到的订单信息是："+message);
       }
   }
   

3. 启动服务SpringbootRabbitmqFanoutConsumerApplication，查看效果

   

16、RabbitMQ-SpringBoot案例 -direct模式

1、定义生产者

1. 创建交换机和队列，绑定关系

   /**
    * rabbitMq配置类 绑定交换机和队列之间的关系
    * @author shkstart
    * @date 2022/1/15 - 14:48
    */
   @Configuration
   public class DirectRabbitMqConfig {
       /**
        * 创建一个Direct模式的交换机且是持久化的
        * @return
        */
       @Bean
       public DirectExchange directExchange(){
           return new DirectExchange("direct_order_exchange", true, false);
       }
       /**
        * 创建一个持久化的队列
        * @return
        */
       @Bean
       public Queue smsDirectQueue(){
           /**
            durable:是否持久化,默认是false, 持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
            exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
            autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
            return new Queue("TestDirectQueue",true,true,false);
           一般设置一下队列的持久化就好,其余两个就是默认false
            */
           return new Queue("sms.direct.queue", true);
       }
   
       /**
        * 创建一个持久化的队列
        * @return
        */
       @Bean
       public Queue emailDirectQueue(){
           return new Queue("email.direct.queue", true);
       }
   
       /**
        * 创建一个持久化的队列
        * @return
        */
       @Bean
       public Queue shortMessageDirectQueue(){
           return new Queue("shortMessage.direct.queue", true);
       }
   
       /**
        * 交换机绑定对应的队列
        * @return
        */
       @Bean
       public Binding smsDirectBindingBuilder(){
           //绑定队列到交换机上   with方法绑定路由key
           return BindingBuilder.bind(smsDirectQueue()).to(directExchange()).with("sms");
       }
       /**
        * 交换机绑定对应的队列
        * @return
        */
       @Bean
       public Binding emailDirectBindingBuilder(){
           //绑定队列到交换机上  with方法绑定路由key
           return BindingBuilder.bind(emailDirectQueue()).to(directExchange()).with("email");
       }
       /**
        * 交换机绑定对应的队列
        * @return
        */
       @Bean
       public Binding shortMessageDirectBuilder(){
           //绑定队列到交换机上   with方法绑定路由key
           return BindingBuilder.bind(shortMessageDirectQueue()).to(directExchange()).with("shortMessage");
       }
   }
   

2. 编写业务代码，进行消息的发送

   //模拟用户下单  direct模式，根据路由key分法消息
   public void makeDirectOrder(String  id , String  productId,int num){
       //根据商品id查询库存是否充足
       //保存订单
       String orderId = UUID.randomUUID().toString();
       System.out.println("订单生成成功："+orderId);
       //交换机
       String exchangeName = "direct_order_exchange";
       String routeKey = "";
       //通过MQ来完成消息的分法
       //参数1：交换机名称 参数2：路由key/queue队列的名称    参数3：消息的内容
       rabbitTemplate.convertAndSend(exchangeName, "email",orderId);
       rabbitTemplate.convertAndSend(exchangeName, "sms",orderId);
   }
   

3. 编写测试方法

   //测试direct模式
   @Test
   void testDirect() {
       orderService.makeDirectOrder("1", "1", 12);
   }
   

2、定义消费者

1. 编写接收消息的方法(注意更改类名和交换机名称)，其他两个雷同

   /**
    * 邮件的消费者
    * @author shkstart
    * @date 2022/1/15 - 15:13
    */
   @RabbitListener(queues = "email.direct.queue")  //指定监听的交换机
   @Component //注入到容器中
   public class DirectEmailConsumer {
       @RabbitHandler //标识为了一个监听消息的方法  监听到的消息会自动注入到message中
       public void reviceMessage(String message){
           System.out.println("邮件的direct----->接收到的订单信息是："+message);
       }
   }
   

2. 启动SpringbootRabbitmqFanoutConsumerApplication，在运行生产者的测试方法，查看效果

   

   

3、创建交换机和队列以及绑定的配置文件放在那边

1. 注意：如果没有创交换机和队列，且没有绑定关系，启动消费者去监听会产生抛出异常
2. 消费者和生产者都可以进行交换机和队列，且绑定之间的关系的操作，只需要把消费者的配置文件复制一份给消费者即可。
3. 交换机和队列可以通过消费者，可以通过生产者，可以通过图形化界面的方式进行创建和绑定。但最好通过代码，最好在消费者进行声明，因为如果消费者先启动，没有寻找到对应的交换机和队列，会抛出异常。

17、 RabbitMQ-SpringBoot案例 -topic模式(注解的方式)

1、定义生产者

1. 业务处理完，进行消息的发送

   //模拟用户下单 topic模式，根据路由key分法消息
   public void makeTopicOrder(String  id , String  productId,int num){
       //根据商品id查询库存是否充足
       //保存订单
       String orderId = UUID.randomUUID().toString();
       System.out.println("订单生成成功："+orderId);
       //交换机
       String exchangeName = "topic_order_exchange";
       //通过MQ来完成消息的分法
       //参数1：交换机名称 参数2：路由key/queue队列的名称    参数3：消息的内容
       //#.email.#
       //com.shortMessage.#
       //*.sms.#
       String routeKey = "com.shortMessage.email.xxx"; //给email 和 shortMessage 消费者发送消息
       rabbitTemplate.convertAndSend(exchangeName, routeKey,orderId);
   }
   

2. 测试代码

   //测试topic模式
   @Test
   void testTopic() {
       orderService.makeTopicOrder("1", "1", 12);
   }
   

2、定义消费者

1. 编写接收消息的方法(注意更改类名和交换机名称)，其他两个雷同

   /**
    * 注解的方式创建交换机和队列，在进行绑定 就可以不用写配置类了
    * 邮件的消费者
    * @author shkstart
    * @date 2022/1/15 - 15:13
    */
   @RabbitListener(bindings = @QueueBinding(
           //持久化队列的创建
           value = @Queue(value = "email.topic.queue",durable = "true",autoDelete = "false"),
           //topic模式的交换机的创建
           exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
           //指定路由key
           key = "#.email.#"
   ))  //注解形式就行绑定
   @Component //注入到容器中
   public class TopicEmailConsumer {
   
       @RabbitHandler //标识为了一个监听消息的方法  监听到的消息会自动注入到message中
       public void reviceMessage(String message){
           System.out.println("邮件的direct----->接收到的订单信息是："+message);
       }
   }
   

2. 启动SpringbootRabbitmqFanoutConsumerApplication，在运行生产者的测试方法，查看效果

   

   

18、RabbitMQ高级-过期时间TTL

1、概述和实操

1. 过期时间TTL表示可以对消息设置预期的时间，在这个时间内都可以被消费者接收获取；过了之后默认消息将自动被删除，也可以设置到死信队列中。RabbitMQ可以对消息和队列设置TTL。目前有两种方法可以设置。

   • 第一种方法是通过队列属性设置，队列中所有消息都有相同的过期时间。可将删除的队列投入到死信队列中。

     1. 交换机和队列的创建，创建队列设置ttl（x-message-ttl）参数，在绑定关系。

        /**
         * ttl参数的设置，设置给整个队列
         */
        @Configuration
        public class TTLRabbitMqConfig {
            @Bean
            public DirectExchange ttlDirectExchange() {
                return new DirectExchange("ttl_direct_exchange", true, false);
            }
            @Bean
            public Queue ttlQueue() {
                HashMap<String, Object> map = new HashMap<>();
                //参数 x-message-ttl 的值 必须是非负 32 位整数 (0 <= n <= 2^32-1) ，以毫秒为单位表示 TTL 的值。这样，值 6000 表示存在于 队列 中的当前 消息 将最多只存活 6 秒钟。
                map.put("x-message-ttl", 5000);//这里一定是int类型
                return new Queue("ttl.Queue", true, false, false, map);
            }
            @Bean
            public Binding ttlBinding() {
                return BindingBuilder.bind(ttlQueue()).to(ttlDirectExchange()).with("ttl");
            }
        }
        

        • web创建队列时添加ttl参数

          

     2. 业务处理发送消息

        // ttl 设置队列中的消息自动删除的
        public void makeTtlOrder(String  id , String  productId,int num){
            //根据商品id查询库存是否充足
            //保存订单
            String orderId = UUID.randomUUID().toString();
            System.out.println("订单生成成功："+orderId);
            //交换机
            String exchangeName = "ttl_direct_exchange";
            //通过MQ来完成消息的分法
            //参数1：交换机名称 参数2：路由key/queue队列的名称    参数3：消息的内容
            String routeKey = "ttl"; //给ttl 消费者发送消息
            rabbitTemplate.convertAndSend(exchangeName, routeKey,orderId);
        }
        

     3. 测试代码

        //测试ttl自动过期时间
        @Test
        void testTTL() {
            orderService.makeTtlOrder("1", "1", 12);
        }
        

     4. 图形化查看结果

        

        

   • 第二种方法是对消息进行单独设置，每条消息TTL可以不同。这时候的消息无法投入到死信队列中，会直接给删除。

     1. 交换机和队列的创建，绑定关系。

        //ttl参数的设置，给单条消息设置时间
        @Bean
        public Queue ttlMessageQueue() {
            return new Queue("ttlMessage.Queue", true);
        }
        @Bean
        public Binding ttlMessageBinding() {
            return BindingBuilder.bind(ttlMessageQueue()).to(ttlDirectExchange()).with("ttlMessage");
        }
        

     2. 业务处理发送消息,且设置消息的过期时间

        // ttl 设置某条消息的自动删除的
        public void makeTtlMessageOrder(String  id , String  productId,int num){
            //根据商品id查询库存是否充足
            //保存订单
            String orderId = UUID.randomUUID().toString();
            System.out.println("订单生成成功："+orderId);
            //交换机
            String exchangeName = "ttl_direct_exchange";
            //通过MQ来完成消息的分法
            //参数1：交换机名称 参数2：路由key/queue队列的名称    参数3：消息的内容
            String routeKey = "ttlMessage"; //给ttl 消费者发送消息
            MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    //设置过期时间
                    message.getMessageProperties().setExpiration("5000");
                    //设置消息的编码方式
                    message.getMessageProperties().setContentEncoding("utf-8");
                    return message;
                }
            };
            rabbitTemplate.convertAndSend(exchangeName, routeKey,orderId,messagePostProcessor);
        }
        

     3. 测试代码

        //测试ttl给单条消息自动过期时间
        @Test
        void testTTLMessage() {
            orderService.makeTtlMessageOrder("1", "1", 12);
        }
        

     4. 图形化查看结果

        

        

2. 如果上述两种方法同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就称为dead message被投递到死信队列， 消费者将无法再收到该消息。

2、 总结消息TTL

1. expiration 字段以毫秒为单位表示 TTL 值。且与 x-message-ttl 具有相同的约束条件。因为 expiration 字段必须为字符串类型，broker 将只会接受以字符串形式表达的数字。
2. 当同时指定了 queue 和 message 的 TTL 值，则两者中较小的那个才会起作用。

19、RabbitMQ高级-消息确认机制的配置

NONE值是禁用发布确认模式，是默认值
CORRELATED值是发布消息成功到交换器后会触发回调方法，如1示例
SIMPLE值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;

18、RabbitMQ高级-死信队列

1、概念

1. DLX，全称为Dead-Letter-Exchange , 可以称之为死信交换机，也有人称之为死信邮箱。当消息在一个队列中变成死信(dead message 就是设置ttl时间过期之后)之后，它能被重新发送到另一个交换机中，这个交换机就是DLX ，绑定DLX的队列就称之为死信队列。消息变成死信，可能是由于以下的原因：
   • 消息被拒绝
   • 消息过期
   • 队列达到最大长度
   • 面试题：什么样的消息会进入死信队列中。
2. DLX也是一个正常的交换机，和一般的交换机没有区别，它能在任何的队列上被指定，实际上就是设置某一个队列的属性。当这个队列中存在死信时，Rabbitmq就会自动地将这个消息重新发布到设置的DLX上去，进而被路由到另一个队列，即死信队列。要想使用死信队列，只需要在定义队列的时候设置队列参数 x-dead-letter-exchange 指定交换机即可。

2、在rabbitMQ管理界面中设置队列消息过期后进入的死信队列

3、代码步骤

1. 创建一个存放死信队列的交换机和队列，并绑定上关系

   /**
    * 死信队列
    */
   @Configuration
   public class DeadRabbitMqConfig {
       @Bean
       public DirectExchange deadDirectExchange() {
           return new DirectExchange("dead_direct_exchange", true, false);
       }
   
       @Bean
       public Queue deadQueue() {
           return new Queue("dead.Queue", true);
       }
   
       @Bean
       public Binding deadBinding() {
           return BindingBuilder.bind(deadQueue()).to(deadDirectExchange()).with("dead");
       }
   }
   

2. 设置死信队列，消息过期之后发送到死信队列中，在创建队列时就将其绑定(队列设置参数x-dead-letter-exchange、x-dead-letter-routing-key)，注意：如果之前已经产生了的队列，再去修改这个队列的参数去添加队列会抛出异常，需要手动删除之后在重新设置。

   @Bean
   public Queue ttlDeadQueue() {
       HashMap<String, Object> map = new HashMap<>();
       //5秒后消息过期，设置到死信队列中
       map.put("x-message-ttl", 5000);//这里一定是int类型
   
       //设置死信队列的参数  绑定死信队列的交换机  如果之前已经产生了的队列，再去修改这个队列的参数去添加队列会抛出异常
       map.put("x-dead-letter-exchange", "dead_direct_exchange");
       //绑定路由key  由路由key发送到匹配的队列上
       map.put("x-dead-letter-routing-key", "dead");//绑定死信队列的路由key
       return new Queue("ttlDead.Queue", true, false, false, map);
   }
   @Bean
   public Binding ttlDeadBinding() {
       return BindingBuilder.bind(ttlDeadQueue()).to(ttlDirectExchange()).with("ttlDead");
   }
   

3. 业务逻辑处理，发送消息

   // 设置队列中的消息自动删除的后注入到死信队列中
   public void makeTtlDeadOrder(String  id , String  productId,int num){
       //根据商品id查询库存是否充足
       //保存订单
       String orderId = UUID.randomUUID().toString();
       System.out.println("订单生成成功："+orderId);
       //交换机
       String exchangeName = "ttl_direct_exchange";
       //通过MQ来完成消息的分法
       //参数1：交换机名称 参数2：路由key/queue队列的名称    参数3：消息的内容
       String routeKey = "ttlDead"; //给ttlDead 消费者发送消息  fonout模式不需要指定路由key
       rabbitTemplate.convertAndSend(exchangeName, routeKey,orderId);
   }
   

4. 测试代码

   //测试ttl整个队列的自动过期时间,g过期的消息注入到死信队列中
   @Test
   void testTTLDead() {
       orderService.makeTtlDeadOrder("1", "1", 12);
   }
   

5. 结果

   

4、扩展：消息超出指定队列能容量的消息长度，其余消息进入到死信队列

1. 设置死信队列，消息超出指定队列能容量的消息长度时，在创建队列时就将其绑定(给队列设置参数x-max-length)

   @Bean
   public Queue ttlDeadMaxLengthQueue() {
       HashMap<String, Object> map = new HashMap<>();
       //map.put("x-message-ttl", 5000);//这里一定是int类型  如果设置了这个参数，这个队列中的5个消息在5秒没有被消费也会被打入到死信队列中
   
       //设置死信队列的参数  绑定死信队列的交换机  如果之前已经产生了的队列，再去修改这个队列的参数去添加队列会抛出异常
       map.put("x-dead-letter-exchange", "dead_direct_exchange");
       //绑定路由key  由路由key发送到匹配的队列上
       map.put("x-dead-letter-routing-key", "dead");//绑定死信队列的路由key
   
       //设置队列中最多能存放多少条消息 通过参数x-max-length设置  超过5条，其他都进入死信队列
       map.put("x-max-length", 5);
       return new Queue("ttlDeadMaxLength.Queue", true, false, false, map);
   }
   
   @Bean
   public Binding ttlDeadMaxLengthBinding() {
       return BindingBuilder.bind(ttlDeadMaxLengthQueue()).to(ttlDirectExchange()).with("ttlDeadMaxLength");
   }
   

2. 业务逻辑处理，发送消息

   // 设置消息超出指定队列能容量的消息长度时注入到死信队列中
   public void makeTtlDeadMaxLengthOrder(String  id , String  productId,int num){
       //根据商品id查询库存是否充足
       //保存订单
       String orderId = UUID.randomUUID().toString();
       System.out.println("订单生成成功："+orderId);
       //交换机
       String exchangeName = "ttl_direct_exchange";
       //通过MQ来完成消息的分法
       //参数1：交换机名称 参数2：路由key/queue队列的名称    参数3：消息的内容
       String routeKey = "ttlDeadMaxLength"; //给ttlDeadMaxLength 消费者发送消息
       rabbitTemplate.convertAndSend(exchangeName, routeKey,orderId);
   }
   

3. 测试代码

   //测试 消息超出指定队列能容量的消息长度，其余消息进入到死信队列
   @Test
   void testTTLMaxLengthDead() {
       //发送10条消息
       for (int i = 0; i < 10; i++) {
           orderService.makeTtlDeadMaxLengthOrder("1", "1", 12);
       }
   }
   

4. 结果

   

19、RabbitMQ运维-持久化机制和内存磁盘的监控

1、RibbitMQ持久化和消息的持久化

• 持久化就把信息写入到磁盘的过程，重启rabbitMQ时，会读入写入磁盘的数据，从而达到消息不丢失。、

• 把消息默认放在内存中是为了加快传输和消费的速度，存入磁盘是保证消息数据的持久化。

2、RabbitMQ非持久化消息

• 非持久消息：是指当内存不够用的时候，会把消息和数据转移到磁盘，但是重启以后非持久化队列消息就丢失。

3、RabbitMQ持久化分类

• RabbitMQ的持久化队列分为：
  1. 队列持久化
  2. 消息持久化
  3. 交换机持久化
  4. 不论是持久化的消息还是非持久化的消息都可以写入到磁盘中，只不过非持久的是等内存不足的情况下才会被写入到磁盘中。

4、RabbitMQ队列持久化的代码实现

• 队列的持久化是定义队列时的durable参数来实现的，Durable为true时，队列才会持久化。

• 其中参数2：设置为true,就代表的是持久化的含义。即durable=true。持久化的队列在web控制台中有一个D 的标

  //部分代码  具体可查看上面的代码
  String exchangeName = "create_exchange";
  //6、定义路由key 指定需要匹配的路由key
  String routeKey = "order";
  //7、定义交换机的类型
  String exchangeType = "direct";
  //创建交换机  参数1：交换机名称  参数2：交换机类型  参数3：是否持久化
  channel.exchangeDeclare(exchangeName, exchangeType, true);
  /**
   * Rabbitmq不允许创建两个相同的队列名称，否则会报错。
   *                 参数1：队列名
   *                 参数2：是否持久化durable=false 持久化消息是否存盘，如果false 非持久化 true持久化  非持久化会存盘嘛？
   *                 参数3：是否有排他性  是否独占独立，排他队列
   *                 参数4：是否自动删除 随着最后一个消费者消费消息以后是否把队列删除
   *                 参数5：是否携带额外参数   如果携带了参数 就是参数模式 headers的队列
   */
  //8、创建队列
  channel.queueDeclare("queue5", true, false, false, null);
  channel.queueDeclare("queue6", true, false, false, null);
  channel.queueDeclare("queue7", true, false, false, null);
  

5、RabbitMQ消息持久化

• 消息持久化是通过消息的属性deliveryMode来设置是否持久化，在发送消息时通过basicPublish的参数传入。

• DeliveryMode 属性：2代表持久化，1代表非持久化

  //部分代码
  BasicProperties properties = new BasicProperties();
  properties.builder().deliveryMode(2);
  //发送消息
  channel.basicPublish(exchangeName, routeKey, properties, message.getBytes());
  

6、RabbitMQ交换机持久化

• 和队列一样，交换机也需要在定义的时候设置持久化的标识，否则在rabbit-server服务重启以后将丢失。

  //创建交换机  参数1：交换机名称  参数2：交换机类型  参数3：是否持久化
  channel.exchangeDeclare(exchangeName, exchangeType, true);
  

20、RabbitMQ运维-内存磁盘的监控

1、RabbitMQ的内存警告

1. 当内存使用超过配置的阈值或者磁盘空间剩余空间对于配置的阈值时，RabbitMQ会暂时阻塞客户端的连接，并且停止接收从客户端发来的消息，以此避免服务器的崩溃，客户端与服务端的心态检测机制也会失效。

2. 如下图：02、RabbitMQ的内存控制

   

   

3. 当出现blocking或blocked话说明到达了阈值和以及高负荷运行了。

2、RabbitMQ的内存控制

1. 参考帮助文档：https://www.rabbitmq.com/configure.html
2. 当出现警告的时候，可以通过配置去修改和调整

2.1、命令的方式

rabbitmqctl set_vm_memory_high_watermark <fraction>
rabbitmqctl set_vm_memory_high_watermark absolute 50MB

• fraction/value 为内存阈值。默认情况是：0.4/2GB，代表的含义是：当RabbitMQ的内存超过40%时，就会产生警告并且阻塞所有生产者的连接。通过此命令修改阈值在Broker重启以后将会失效，通过修改配置文件方式设置的阈值则不会随着重启而消失，但修改了配置文件一样要重启broker才会生效。

• 修改

  rabbitmqctl set_vm_memory_high_watermark absolute 50MB
  

  

  

2.2、配置文件方式 rabbitmq.conf

• 当前配置文件：/etc/rabbitmq/rabbitmq.conf

  #默认
  #vm_memory_high_watermark.relative = 0.4
  # 使用relative相对值进行设置fraction,建议取值在04~0.7之间，不建议超过0.7.
  vm_memory_high_watermark.relative = 0.6
  # 使用absolute的绝对值的方式，但是是KB,MB,GB对应的命令如下
  vm_memory_high_watermark.absolute = 2GB
  

3、RabbitMQ的内存换页

1. 在某个Broker节点及内存阻塞生产者之前，它会尝试将队列中的消息换页到磁盘以释放内存空间，持久化和非持久化的消息都会写入磁盘中，其中持久化的消息本身就在磁盘中有一个副本，所以在转移的过程中持久化的消息会先从内存中清除掉。

2. 例子：比如有1000MB内存，当内存的使用率达到了400MB,已经达到了极限，但是因为配置的换页内存0.5，这个时候会在达到极限400mb之前，会把内存中的200MB进行转移到磁盘中。从而达到稳健的运行。

3. 可以通过设置 vm_memory_high_watermark_paging_ratio 来进行调整

   vm_memory_high_watermark.relative = 0.4
   vm_memory_high_watermark_paging_ratio = 0.7（设置小于1的值）
   

4. 为什么设置小于1，以为你如果你设置为1的阈值。内存都已经达到了极限了。你在去换页意义不是很大了。

4、RabbitMQ的磁盘预警

1. 当磁盘的剩余空间低于确定的阈值时(如：磁盘的剩余情况低于100M时，就会出现预警)，RabbitMQ同样会阻塞生产者，这样可以避免因非持久化的消息持续换页而耗尽磁盘空间导致服务器崩溃。

   

   

2. 通过命令方式修改如下：

   rabbitmqctl set_disk_free_limit  <disk_limit>
   rabbitmqctl set_disk_free_limit memory_limit  <fraction>
   disk_limit：固定单位 KB MB GB
   fraction ：是相对阈值，建议范围在:1.0~2.0之间。（相对于内存）
   

3. 通过配置文件配置如下：

   disk_free_limit.relative = 3.0
   disk_free_limit.absolute = 50mb
   

e, false, null);
channel.queueDeclare(“queue6”, true, false, false, null);
channel.queueDeclare(“queue7”, true, false, false, null);

##### 5、RabbitMQ消息持久化

- 消息持久化是通过消息的属性deliveryMode来设置是否持久化，在发送消息时通过basicPublish的参数传入。

- DeliveryMode 属性：2代表持久化，1代表非持久化

```java
//部分代码
BasicProperties properties = new BasicProperties();
properties.builder().deliveryMode(2);
//发送消息
channel.basicPublish(exchangeName, routeKey, properties, message.getBytes());

6、RabbitMQ交换机持久化

• 和队列一样，交换机也需要在定义的时候设置持久化的标识，否则在rabbit-server服务重启以后将丢失。

  //创建交换机  参数1：交换机名称  参数2：交换机类型  参数3：是否持久化
  channel.exchangeDeclare(exchangeName, exchangeType, true);
  

20、RabbitMQ运维-内存磁盘的监控

1、RabbitMQ的内存警告

1. 当内存使用超过配置的阈值或者磁盘空间剩余空间对于配置的阈值时，RabbitMQ会暂时阻塞客户端的连接，并且停止接收从客户端发来的消息，以此避免服务器的崩溃，客户端与服务端的心态检测机制也会失效。

2. 如下图：02、RabbitMQ的内存控制

   [外链图片转存中…(img-riYhEFis-1642383413263)]

   [外链图片转存中…(img-Mynzpm37-1642383413264)]

3. 当出现blocking或blocked话说明到达了阈值和以及高负荷运行了。

2、RabbitMQ的内存控制

1. 参考帮助文档：https://www.rabbitmq.com/configure.html
2. 当出现警告的时候，可以通过配置去修改和调整

2.1、命令的方式

rabbitmqctl set_vm_memory_high_watermark <fraction>
rabbitmqctl set_vm_memory_high_watermark absolute 50MB

• fraction/value 为内存阈值。默认情况是：0.4/2GB，代表的含义是：当RabbitMQ的内存超过40%时，就会产生警告并且阻塞所有生产者的连接。通过此命令修改阈值在Broker重启以后将会失效，通过修改配置文件方式设置的阈值则不会随着重启而消失，但修改了配置文件一样要重启broker才会生效。

• 修改

  rabbitmqctl set_vm_memory_high_watermark absolute 50MB
  

  [外链图片转存中…(img-O9k9Odw2-1642383413265)]

  [外链图片转存中…(img-6Zs9sZZF-1642383413267)]

2.2、配置文件方式 rabbitmq.conf

• 当前配置文件：/etc/rabbitmq/rabbitmq.conf

  #默认
  #vm_memory_high_watermark.relative = 0.4
  # 使用relative相对值进行设置fraction,建议取值在04~0.7之间，不建议超过0.7.
  vm_memory_high_watermark.relative = 0.6
  # 使用absolute的绝对值的方式，但是是KB,MB,GB对应的命令如下
  vm_memory_high_watermark.absolute = 2GB
  

3、RabbitMQ的内存换页

1. 在某个Broker节点及内存阻塞生产者之前，它会尝试将队列中的消息换页到磁盘以释放内存空间，持久化和非持久化的消息都会写入磁盘中，其中持久化的消息本身就在磁盘中有一个副本，所以在转移的过程中持久化的消息会先从内存中清除掉。

2. 例子：比如有1000MB内存，当内存的使用率达到了400MB,已经达到了极限，但是因为配置的换页内存0.5，这个时候会在达到极限400mb之前，会把内存中的200MB进行转移到磁盘中。从而达到稳健的运行。

3. 可以通过设置 vm_memory_high_watermark_paging_ratio 来进行调整

   vm_memory_high_watermark.relative = 0.4
   vm_memory_high_watermark_paging_ratio = 0.7（设置小于1的值）
   

4. 为什么设置小于1，以为你如果你设置为1的阈值。内存都已经达到了极限了。你在去换页意义不是很大了。

4、RabbitMQ的磁盘预警

1. 当磁盘的剩余空间低于确定的阈值时(如：磁盘的剩余情况低于100M时，就会出现预警)，RabbitMQ同样会阻塞生产者，这样可以避免因非持久化的消息持续换页而耗尽磁盘空间导致服务器崩溃。

   [外链图片转存中…(img-QEKhT8q8-1642383413268)]

   [外链图片转存中…(img-u48cdX4D-1642383413268)]

2. 通过命令方式修改如下：

   rabbitmqctl set_disk_free_limit  <disk_limit>
   rabbitmqctl set_disk_free_limit memory_limit  <fraction>
   disk_limit：固定单位 KB MB GB
   fraction ：是相对阈值，建议范围在:1.0~2.0之间。（相对于内存）
   

3. 通过配置文件配置如下：

   disk_free_limit.relative = 3.0
   disk_free_limit.absolute = 50mb