RabbitMQ

RabbitMQ是一个消息中间件：它接受并转发消息。

简介

什么是MQ？

MQ(message queue)，从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是message而已，还是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了MQ之后，消息发送上游只需要依赖MQ，不用依赖其他服务。

为什么要用MQ？

流量削峰：

  • 举个例子，如果订单系统最多能处理一万次订单，这个处理能力应付正常时段的下单时绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。使用消息队列做缓冲，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这时有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好。

应用解耦：

  • 以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，中单用户感受不到物流系统的故障，提升系统的可用性。

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled.png

异步处理：

  • 有些服务间调用是异步的，例如A调用B，B需要花费很长时间执行，但是A需要知道B什么时候可以执行完，以前一般有两种方式，A过一段时间去调用B的查询api查询。或者A提供一个callback api，B执行完之后调用api通知A服务。这两种方式都不是很优雅，使用消息总线，可以很方便解决这个问题，A调用B服务后，只需要监听B处理完成的消息，当B处理完成后，会发送一条消息给MQ，MQ会将此消息转发给A服务。这样A服务既不用循环调用B的查询api，也不用提供callback api。同样B服务也不用做这些操作。A服务还能及时的得到异步处理成功的消息。

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%201.png

四大核心概念

生产者：

  • 产生数据发送消息的程序是生产者

交换机：

  • 交换机是RabbitMQ非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定

队列：

  • 队列是RabbitMQ内部使用的一种数据结构，尽管消息流经RabbitMQ和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式

消费者：

  • 消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。

什么是中间件？

1、利用可靠的消息传递机制进行系统和系统直接的通讯
2、通过提供消息传递和消息的排队机制，他可以在分布式环境下扩展进程间的通讯

消息中间件应用场景

  1. 跨系统数据传递
  2. 高并发的流量削峰
  3. 数据的分发和异步处理
  4. 分布式事务

消息中间件的本质及设计

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

MQ消息队列：负责数据的传接受，存储和传递，所以性能要过于普通服务和技术

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%202.png

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

  1. 消息的协议
  2. 消息的持久化机制
  3. 消息的分发策略
  4. 消息的高可用，高可靠
  5. 消息的容错机制

消息中间件协议

常见消息中间件协议

  1. OpenWire
  2. AMQP
  3. MQTT
  4. Kafka
  5. OpenMessage

AMQP协议

AMQP:(全称: Advanced Message Queuing Protocol)是高级消息队列协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等。

安装RabbitMQ前需要安装Erlang的环境

特点：

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

AMQP协议的支持者：

  • RabbirMQ
  • ActiveMQ

Kafka协议

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

特点：

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

消息的分发策略

消息队列高可用和高可靠

什么是高可用机制？

高可用：是指产品在规定的条件和规定的时刻或时间内处于规定功能状态的能力。高可用常用策略是集群。

RabbitMQ的使用

安装

注意：在Linux安装Erlang使用rpm的方式安装比较方便，Erlang的rpm软件包下载地址：https://github.com/rabbitmq/erlang-rpm/releases

  1. 安装RabbitMQ运行环境Erlang
  2. yum install -y socat
  3. 解压并安装 RabbitMQ rpm -Uvh rabbitmq-server-3.8.14-1.el8.noarch.rpm
  4. 下载图形化界面插件rabbitmq-plugins enable rabbitmq_management

Docker启动方式

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

命令:

  • 启动服务:systemctl start rabbitmq-server
  • 重启服务:systemctl restart rabbitmq-server
  • 查看服务状态：systemctl status rabbitmq-server.service
  • 停止服务：systemctl stop rabbitmq-server
  • 开机启动：systemctl enable rabbitmq-server

用户管理

新增用户

  • rabbitmqctl add_user admin admin

设置用户分配操作权限

  • rabbitmqctl set_user_tags admin administrator

用户级别：

  • administrator：最高权限管理者，可以对rabbitmq进行管理
  • monitoring：监控者，登陆控制台，查看所有信息
  • policymaker：策略制定者，登陆控制台指定策略
  • managment：普通管理员，登陆控制台

为用户添加资源权限

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

其他

  • rabbitmqctl add_user 账号 密码 ：添加用户
  • rabbitmqctl add_user_tags 账号 administrator ：为用户设置角色
  • rabbitmqctl change_password username newPassword ：修改密码
  • rabbitmqctl delete_user username :删除用户
  • rabbitmqctl list_users ：查看用户列表
  • rabbitmqctl.bat set_permissions -p / 用户名 ".*" ".*" ".*" ：设置用户权限为最高级别

Java Client

概述：

RabbitMQ Java 客户端使用com.rabbitmq.client作为其顶级包。关键的类和接口是：

  • Channel：代表一个AMQP 0-9-1通道，提供大部分操作（协议方法）。
  • Connection：代表一个 AMQP 0-9-1 连接
  • ConnectionFactory：构造Connection实例
  • Consumer：代表一个消息消费者
  • DefaultConsumer：消费者常用的基类
  • BasicProperties：消息属性（元数据）
  • BasicProperties.Builder：建设者BasicProperties

**AMQP 0-9-1模型解释：**AMQP 0-9-1（高级消息队列协议）是一种消息传递协议，它使符合要求的客户端应用程序能够与符合要求的消息传递中间件代理进行通信。文档链接：https://www.notion.so/ikart/RabbitMQ-086e94acbe054f02b4f659afd74cc0cb#72d4a29e15c344de921b95bc92eacb97

客户端连接服务端：

port：5672用于常规连接， 5671用于使用 TLS 的连接

  • 方式一：使用给定的参数（主机名、端口号等）连接到 RabbitMQ 节点
ConnectionFactory factory =new ConnectionFactory();//默认情况下，“guest”/“guest”，寓于本地连接
factory.setUsername(userName);
factory.setPassword(password);
factory.setVirtualHost(virtualHost);//没有则省略
factory.setHost(hostName);
factory.setPort(portNumber);
  • 方式二：通过url连接
ConnectionFactory factory = new ConnectionFactory();
//连接方式参考 factory.setUri( "amqp://userName:password@hostName:portNumber/virtualHost" ); /virtualHost没有则省略
factory.setUri("amqp://ikart:sh01073938@ikart.top:5672");
Connection conn = factory.newConnection();

RabbitMQ的核心

名词介绍：

  • Broker：接收和分发消息的应用，RabbitMQ Server就是Message Broker
  • Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
  • Connection：publisher／consumer和broker之间的TCP连接
  • Channel：如果每一次访问 RabbitMQ 都建立一个Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id 帮助客户端和message broker 识别 channel，所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销
  • Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
  • Queue：消息最终被送到这里等待consumer取走
  • Binding：exchange和queue之间的虚拟连接，binding中可以包含routing key，Binding信息被保存到exchange中的查询表中，用于message的分发依据

Channel常用方法

队列声明参数表说明：

/*
生成一个队列
  1.队列名称
  2.队列里面的消息是否持久化 默认消息存储在内存中
  3.该队列是否只供一个消费者进行消费 是否进行共享 true可以多个消费者消费
  4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true自动删除
  5.其他参数
 */
channel.queueDeclare(QUEUE_NAME, false, false, false, null);

交换机声明参数列表说明：

/*
参数1：交换机名
参数2：交换机类型
*/
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

基本发布参数列表说明：

/*
发送一个消息
1.发送到那个交换机，""(空串表示默认交换机)
2.路由的key是哪个
3.其他的参数信息
4.发送消息的消息体
 */
channel.basicPublish("", QUEUE_NAME,null,message.getBytes());

基本消费参数列表说明：

/*
消费者消费消息
    1.消费哪个队列
    2.消费成功之后是否要自动应答 true代表自动应答 false手动应答
		3.消费成功回调
    4.消费者未成功消费的回调
*/
channel.basicConsume(Connect.QUEUE_NAME,true,deliverCallback,cancelCallback);

拒接消息：

//requeue设置为false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中 
channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);

简单模式：

即生产者通过队列将消息发送给消费者。

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%203.png

工作模式（Work Queues）：

工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。工作队列是轮训分发消息。

轮询分发消息：

即将按轮询分发消息给消费者。

消息应答：

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况。RabbitMQ一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费这的消息，因为它无法接收到。为了保证消息在发送过程中不丢失，rabbitmq引入消息应答机制，消息应答就是:消费者在接收到消息并且处理该消息之后，告诉rabbitmq它已经处理了，rabbitmq可以把该消息删除了。

自动应答：

消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前，消费者那边出现连接或者channel关闭，那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

不推荐使用自动应答

默认为自动应答

消息应答的方法：

  • Channel.basicAck(用于肯定确认)：RabbitMQ已知道该消息并且成功的处理消息，可以将其丢弃了
  • Channel.basicNack(用于否定确认)
  • Channel.basicReject(用于否定确认)：与Channel.basicNack相比少一个参数，不处理该消息了直接拒绝，可以将其丢弃了

Multiple的解释：

手动应答的好处是可以批量应答并且减少网络拥堵

multiple的true和false代表不同意思：

  • true代表批量应答channel上未应答的消息
    比如说channel上有传送tag的消息 5,6,7,8 当前tag是8 那么此时
    5-8的这些还未应答的消息都会被确认收到消息应答
  • false同上面相比
    只会应答tag=8的消息 5,6,7这三个消息依然不会被确认收到消息应答

消息自动重新入队：

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或TCP连接丢失)，导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

ACK：应答

RabbitMQ的持久化

队列持久化：

队列没有设置持久化的情况下，RabbitMQ重启后队列消失

如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新创建一个持久化的队列，否则抛出异常。

  • 声明持久化
// args2:true表示开启持久化
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%204.png

消息持久化：

  • 消息实现持久化需要在消息生产者修改代码，MessageProperties.PERSISTENT_TEXT_PLAIN添加这个属性。
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_BASIC, message.getBytes());
  • **注意：**将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候 但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。

不公平分发：

  • 即能者多劳，将任务更多的分配给处理速度快的消费者。意思就是如果这个任务我还没有处理完或者我还没有应答你，你先别分配给我，我目前只能处理一个任务，然后rabbitmq就会把该任务分配给没有那么忙的那个空闲消费者，当然如果所有的消费者都没有完成手上任务，队列还在不停的添加新任务，队列有可能就会遇到队列被撑满的情况，这个时候就只能添加新的worker或者改变其他存储任务的策略。

预取值：

  • 本身消息的发送就是异步发送的，所以在任何时候，channel上肯定不止只有一个消息另外来自消费者的手动确认本质上也是异步的。因此这里就存在一个未确认的消息缓冲区，因此希望开发人员能限制此缓冲区的大小，以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用basic.qos方法设置“预取计数”值来完成的。该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量，RabbitMQ将停止在通道上传递更多消息，除非至少有一个未处理的消息被确认，例如，假设在通道上有未确认的消息5、6、7，8，并且通道的预取计数设置为4，此时RabbitMQ将不会在该通道上再传递任何消息，除非至少有一个未应答的消息被ack。比方说tag=6这个消息刚刚被确认ACK，RabbitMQ将会感知这个情况到并再发送一条消息。消息应答和QoS预取值对用户吞吐量有重大影响。通常，增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的RAM消耗(随机存取存储器)应该小心使用具有无限预处理的自动确认模式或手动确认模式，消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，不同的负载该值取值也不同100到300范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。预取值为1是最保守的。当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，特别是在消费者连接等待时间较长的环境中。对于大多数应用来说，稍微高一点的值将是最佳的。

发布确认

发布确认原理：

  • 生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。
    confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息。

发布确认的策略：

开启发布确认：

Channel channel = RabbitMQUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();

单个发布确认：

  • 一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布。

批量发布确认：

  • 发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题。任然是同步的，一样的会阻塞消息的发布。

异步发布确认：

  • 异步发布确认利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功。可靠性和性价比最高

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%205.png

交换机

生产者生产的消息从不会直接发送到队列，生产者只能将消息发送到交换机(exchange)，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定。

Exchange的类型：

  • 直接（direct）
  • 主题（topic）
  • 标题（headers）
  • 扇出（fanout）

临时队列：

  • 临时队列的申明：
channel.queueDeclare().getQueue();

绑定（bindings）：

  • binding其实是exchange和queue之间的桥梁，它告诉我们exchange和那个队列进行了绑定关系。

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%206.png

Fanout Exchange：

将接收到的所有消息广播到它知道的所有队列中。

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%207.png

Direct Exchange：

这种类型的工作方式是，消息只去到它绑定的routingKey队列中去。

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%208.png

Topic Exchange：

发送到类型是topic交换机的消息的routing_key不能随意写，必须满足一定的要求，它必须是一个单词列表，以点号分隔开，单词列表最多不能超过255个字节。

规则列表中的两个替换符：

  • *：可以代替一个单词
  • #：可以替代零个或多个单词

死信队列

死信，即无法被消费的消息，字面意思可以这样理解，一般来说，producer将消息投递到broker或者直接到queue里了，consumer从queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%209.png

死信处理流程图

死信来源：

  • 消息TTL过期
  • 队列达到最大长度（队列满了，无法再存入消息）
  • 消息被拒绝(basic.reject或basic.nack)并且requeue=false.

延迟队列

延时队列,队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

TTL:

TTL是RabbitMQ中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位是毫秒.

设置TTL的两种方式：

  • 设置消息TTL：
AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder()
                //设置消息过期时间为10秒，10秒后无应答转发到死信队列
                .expiration("10000")
                .build();
  • 设置队列TTL：
params.put("x-message-ttl",10000);
channel.queueDeclare(normalQueue, false, false, false, params);

  • 两者的区别：

    如果是设置了队列的TTL属性，那么一旦消息过期，就会被队列丢弃(如果配置了死信队列被丢到死信队列中)，而设置消息的TTL属性，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间；另外，还需要注意的一点是，如果不设置TTL，表示消息永远不会过期，如果将TTL设置为0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

队列TTL：

RabbitMQ只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行。

abbitmq插件实现延迟队列：

使用延迟队列插件后短的会优先执行。

  • 在官网上下载https://www.rabbitmq.com/community-plugins.html，下载 rabbitmq_delayed_message_exchange插件，然后解压放置到RabbitMQ的插件目录。进入RabbitMQ的安装目录下的plgins目录，执行下面命令让该插件生效，然后重启RabbitMQ
  • 插件目录：/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
  • 开启插件：rabbitmq-plugins enable rabbitmq_delayed_message_exchange

发布确认高级

问题：在生产环境中由于一些不明原因，导致rabbitmq重启，在RabbitMQ重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行RabbitMQ的消息可靠投递呢？特别是在这样比较极端的情况，RabbitMQ集群不可用的时候，无法投递的消息该如何处理呢？
解决方案：

  • 确认机制方案：

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%2010.png

开启发布确认：

  • 开启发布确认需要添加配置：
spring:
  rabbitmq:
    publisher-confirm-type: correlated

  • confirm-type的三种类型：

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

    RabbitMQ其他知识点

    幂等性

    用户对于同一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生了副作用。

    产生原因：

    例如：用户购买商品后支付，支付扣款成功，但是返回结果的时候网络异常，此时钱已经扣了，用户再次点击按钮，此时会进行第二次扣款，返回结果成功，用户查询余额发现多扣钱了，流水记录也变成了两条。在以前的单应用系统中，我们只需要把数据操作放入事务中即可，发生错误立即回滚，但是再响应客户端的时候也有可能出现网络中断或者异常等等

    消息重复消费:

    消费者在消费MQ中的消息时，MQ已把消息发送给消费者，消费者在给MQ返回ack时网络中断，故MQ未收到确认信息，该条消息会重新发给其他的消费者，或者在网络重连后再次发送给该消费者，但实际上该消费者已成功消费了该条消息，造成消费者消费了重复的消息。

    解决思路：

    MQ消费者的幂等性的解决一般使用全局ID 或者写个唯一标识比如时间戳 或者UUID 或者订单消费者消费MQ中的消息也可利用MQ的该id来判断，或者可按自己的规则生成一个全局唯一id，每次消费消息时用该id先判断该消息是否已消费过。

    消费端的幂等性保障：

    在海量订单生成的业务高峰期，生产端有可能就会重复发生了消息，这时候消费端就要实现幂等性，这就意味着我们的消息永远不会被消费多次，即使我们收到了一样的消息。业界主流的幂等性有两种操作:a.唯一ID+指纹码机制,利用数据库主键去重, b.利用redis的原子性去实现

    • 唯一ID+指纹码机制：

      指纹码:我们的一些规则或者时间戳加别的服务给到的唯一信息码,它并不一定是我们系统生成的，基本都是由我们的业务规则拼接而来，但是一定要保证唯一性，然后就利用查询语句进行判断这个id是否存在数据库中,优势就是实现简单就一个拼接，然后查询判断是否重复；劣势就是在高并发时，如果是单个数据库就会有写入性能瓶颈当然也可以采用分库分表提升性能，但也不是我们最推荐的方式。

    • Redis原子性：

      利用redis执行setnx命令，天然具有幂等性。从而实现不重复消费

优先级队列

即按消息设置的优先级对消息进行重新排序，优先级越大越先处理。优先级范围0~255，不建议将优先级设置过大，影响服务器性能。

如何设置优先级❓

注意：要让队列实现优先级需要做的事情有如下事情：1、队列需要设置为优先级队列。2、消息需要设置消息的优先级。3、消费者需要等待消息已经发送到队列中才去消费因为，这样才有机会对消息进行排序（生产者将消息发送完毕后再启动消费者）

  • 控制台添加：

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%2011.png

  • 通过代码添加队列的优先级：
Map<String, Object> params = new HashMap(); 
params.put("x-max-priority", 10); 
channel.queueDeclare("hello", true, false, false, params);
  • 通过代码添加消息的优先级：
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();

惰性队列

惰性队列会尽可能的将消息存入磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，它的一个重要的设计目标是能够支持更长的队列，即支持更多的消息存储。当消费者由于各种各样的原因(比如消费者下线、宕机亦或者是由于维护而关闭等)而致使长时间内不能消费消息造成堆积时，惰性队列就很有必要了。

使用场景：

默认情况下，当生产者将消息发送到RabbitMQ的时候，队列中的消息会尽可能的存储在内存之中，这样可以更加快速的将消息发送给消费者。即使是持久化的消息，在被写入磁盘的同时也会在内存中驻留一份备份。当RabbitMQ需要释放内存的时候，会将内存中的消息换页至磁盘中，这个操作会耗费较长的时间，也会阻塞队列的操作，进而无法接收新的消息。虽然RabbitMQ的开发者们一直在升级相关的算法，但是效果始终不太理想，尤其是在消息量特别大的时候。

队列的两种模式：

**队列具备两种模式：default和lazy。**默认的为default模式，在3.6.0之前的版本无需做任何变更。lazy模式即为惰性队列的模式，可以通过调用channel.queueDeclare方法的时候在参数中设置，也可以通过Policy的方式设置，如果一个队列同时使用这两种方式设置的话，那么Policy的方式具备更高的优先级。如果要通过声明的方式改变已有队列的模式的话，那么只能先删除队列，然后再重新声明一个新的。

  • 惰性队列的声明
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-queue-mode", "lazy");
channel.queueDeclare("myqueue", false, false, false, args);

内存开销对比：

  • 在发送1百万条消息，每条消息大概占1KB的情况下，普通队列占用内存是1.2GB，而惰性队列仅仅占用1.5MB

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%2012.png

RabbitMQ集群

搭建步骤：

  1. 确保各个节点上的.erlang.cookie文件一致

    • 找到.erlang.cookie文件路径
    [root@iZbp13bwx64spr1p933f1yZ rabbitmq]# cd /var/lib/rabbitmq/
[root@iZbp13bwx64spr1p933f1yZ rabbitmq]# ls -a
.  ..  .erlang.cookie  mnesia
[root@iZbp13bwx64spr1p933f1yZ rabbitmq]# pwd
/var/lib/rabbitmq
    • 节点主机复制主机.erlang.cookie文件
    [root@iZbp1g2g2zubvtsxtk7s0qZ rabbitmq]# scp root@116.62.163.67:/var/lib/rabbitmq/.erlang.cookie /var/lib/rabbitmq/
root@116.62.163.67's password: 
.erlang.cookie                                                                                                                                           100%   20     5.3KB/s   00:00    
[root@iZbp1g2g2zubvtsxtk7s0qZ rabbitmq]#

  2. 启动RabbitMQ服务,顺带启动Erlang虚拟机和RbbitMQ应用服务(在三台节点上分别执行以下命令)

    • rabbitmq-server -detached

  3. 在节点2执行

    • rabbitmqctl stop_app
      (rabbitmqctl stop会将Erlang虚拟机关闭，rabbitmqctl stop_app只关闭RabbitMQ服务)
    • rabbitmqctl reset
    • rabbitmqctl join_cluster rabbit@ip
    • rabbitmqctl start_app(只启动应用服务)

  4. 在节点3执行

    • rabbitmqctl stop_app
    • rabbitmqctl reset
    • rabbitmqctl join_cluster rabbit@node2
    • rabbitmqctl start_app

  5. 集群状态

    • rabbitmqctl cluster_status

  6. 需要重新设置用户

    • 创建账号：rabbitmqctl add_user admin 123
    • 设置用户角色：rabbitmqctl set_user_tags admin administrator
    • 设置用户权限：rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

  7. 解除集群节点(node2和node3机器分别执行)

    • rabbitmqctl stop_app
    • rabbitmqctl reset
    • rabbitmqctl start_app
    • rabbitmqctl cluster_status
    • rabbitmqctl forget_cluster_node rabbit@node2(node1机器上执行)

注意项?:

  • 客户端连接服务端：

RabbitMQ%200b062fd4d96e4f03924e7ff06df0d4fb/Untitled%2013.png

  • 如果修改了队列声明方法queueDeclare()后，在RabbitMQ控制面板将将之前的队列删除，重新运行即可?。

参考文档：

  • Client端API文档

RabbitMQ Java Client 5.12.0 API

  • 文档

Documentation: Table of Contents

  • 关于队列的x-###-###参数

RabbitMQ的各个参数

  • RabbitMQ与Springboot整合的POM依赖

  • RabbitMQ插件下载

Community Plugins

Q.E.D.

在等花开，等春天来.