RabbitMQ 重试与集群笔记

记录 RabbitMQ 重试、延迟队列和集群相关配置与学习资料。

阅读说明：下面的 XML、插件及集群示例沿用原笔记版本。示例口令与地址已替换，实际部署请核对 RabbitMQ、Spring AMQP 与插件版本。

RabbitMQ消息重试机制

RabbitMQ重试机制_LiZhen798的博客-CSDN博客_rabbitmq生产者重试

除了使用客户端重试，还可以使用延迟队列实现延迟重试，架构如下：

RabbitMQ相关配图 1

RabbitMQ相关配图 2

延迟重试也有两种方式：

基于死信队列：上图1就是基于死信队列实现架构图，有一个缺点：可能会导致消息消费不及时，就是重试的消息会重新投递到实际消费队列的最后面，此时，可能在重试的等待期间，中间队列中又新增很多消息，导致重试不及时。
基于延时插件：上图2就是基于延时插件实现架构图，这种实现方式，可用于业务的过期时间是不一样的，因为只有等到到期时间到了到会将消息投到队列，所以，不存在后面过期时间短的消息被阻塞，推荐使用这种

自己处理重试计数问题

消费者端处理这消息失败，但是确认这条消息，然后把这条消息发布到nanit.users.retry1交换机。由于我们没有reject这条消息，RabbitMQ并不会帮我们保存x-death header，因此我们需要自己处理重试计数问题。这个可以很容易做到，只需要给消息增加一个自定义的header，比如说x-retries。还有，我们需要自己处理TTL：因为等待队列层面没有设置TTL，我们在发布消息的时候，需要设置一个消息级别的TTL。我们可以新增一个expiration的header，其值就是base-retry-delay和当前重试次数的乘积。这种方式，重试的延迟会随着重试次数增加而增加。最后提一下：我们发布消息的时候，消息的routing key要设成队列的名字。当前这个例子就是，mailman.users.created。

原文: https://www.liangzl.com/get-article-detail-236543.html

设置限流机制

/**
 *  设置限流机制
 *  param1: prefetchSize，消息本身的大小 如果设置为0  那么表示对消息本身的大小不限制
 *  param2: prefetchCount，告诉rabbitmq不要一次性给消费者推送大于N个消息
 *  param3：global，是否将上面的设置应用于整个通道，false表示只应用于当前消费者
 */
channel.basicQos(0, 5, false);

小结

限流的核心代码就是channel.basicQos();
限流情况 ack 不能设置自动签收，一定要手动签收

channel.basicQos()

/**
 * @param prefetchSize maximum amount of content (measured in
 * octets) that the server will deliver, 0 if unlimited
 * @param prefetchCount maximum number of messages that the server
 * will deliver, 0 if unlimited
 * @param global true if the settings should be applied to the
 * entire channel rather than each consumer
 */
void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

该方法的作用是：进行消费端的限流

param1：prefetchSize，消息本身的大小如果设置为0 那么表示对消息本身的大小不限制

param2：prefetchCount，告诉rabbitmq不要一次性给消费者推送大于N个消息

param3：global，是否将上面的设置应用于整个通道

false：表示只应用于当前消费者

true：表示当前通道的所有消费者都应用这个限流策略

RabbitMQ 与 SpringBoot2.X 整合

https://juejin.cn/post/6844903778395635726

RabbitMQ负载均衡

https://blog.csdn.net/cd_yourheart/article/details/108487697

RabbitMQ集群模式有哪些?

RabbitMQ的集群模式有两种：

普通模式：消息实体在其中一个节点上，消费时同步

默认模式，以两个节点（rabbit01、rabbit02）为例来进行说明。

对于Queue来说，消息实体只存在于其中一个节点rabbit01（或者rabbit02），rabbit01和rabbit02两个节点仅有相同的元数据，即队列的结构。

当消息进入rabbit01节点的Queue后，consumer从rabbit02节点消费时，RabbitMQ会临时在rabbit01、rabbit02间进行消息传输，把A中的消息实体取出并经过B发送给consumer。

所以consumer应尽量连接每一个节点，从中取消息。即对于同一个逻辑队列，要在多个节点建立物理Queue。否则无论consumer连rabbit01或rabbit02，出口总在rabbit01，会产生瓶颈。

当rabbit01节点故障后，rabbit02节点无法取到rabbit01节点中还未消费的消息实体。如果做了消息持久化，那么得等rabbit01节点恢复，然后才可被消费；如果没有持久化的话，就会产生消息丢失的现象。

镜像模式：消息实体会同步到所有节点上

把需要的队列做成镜像队列，存在与多个节点属于RabbitMQ的HA方案。

该模式解决了普通模式中的问题，其实质和普通模式不同之处在于，消息实体会主动在镜像节点间同步，而不是在客户端取数据时临时拉取。

该模式带来的副作用也很明显，除了降低系统性能外，如果镜像队列数量过多，加之大量的消息进入，集群内部的网络带宽将会被这种同步通讯大大消耗掉。

所以在对可靠性要求较高的场合中适用。

RabbitMQ镜像队列模式集群搭建

SpringBoot启动时实际只会连接集群中的第一个MQ节点，所以需要使用外部工具来实现负载均衡，可看RabbitProperties.java中源码

RabbitMQ 配置 Nginx 负载均衡

// 编译时 加上 tcp 模块
// 修改 nginx.conf 文件，加入
stream {
  upstream rabbitmq {
    server 192.0.2.2:5672;
    server 192.0.2.3:5672;
  }

  server {
    listen 5673;
    proxy_connect_timeout 3s;
    proxy_timeout 10s;
    proxy_pass rabbitmq;
  }
}

建议将 proxy_timeout 的时间设置长一点 以免 tcp 连接中断重连

spring-data 应用 nginx 负载
<rabbit:connection-factory id="rabbitConnectionFactory" addresses="192.0.2.2:5673" username="admin" password="CHANGE_ME"
publisher-confirms="true"/>

addreses 或者 host + port 属性配上 nginx 的 ip 和监听端口

RabbitMQ 配置 HAProxy 负载均衡

https://www.cnblogs.com/GYY2046/articles/15770394.html

RabbitMQ相关配图 3

RabbitMQ相关配图 4

RabbitMQ 常用命令

参考：windows下安装 rabbitMQ 及操作常用命令 - Eric-Lee - 博客园

RabbitMQ修改默认端口配置

参考：RabbitMQ修改默认端口配置（常用）_毕邺的博客-CSDN博客_rabbitmq默认端口号

x-delayed-message的安装

下载插件：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange ，下载 rabbitmq_delayed_message_exchange-3.8.0.ez。
将下载的包放到 /RABBIT_HOME/plugins下。
执行命令 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 去启用
重启rabbitmq。

SpringBoot整合rabbitMQ 启动服务后便自动创建交换机，队列，绑定关系

参考：https://blog.csdn.net/qq_41712271/article/details/115675092