RabbitMQ之Shovel插件

部署背景：

• 混合部署在一个集群内，发生资源抢夺就会造成某一个rabbitmq节点high_watermark告警。
• a、multi-env    
• b、[_^strong:617eaaa5!]Stateless/Stateful 
• 资源使用不均衡
• 混合云迁移

RabbitMQ是一个开源的消息代理软件，它可以用于构建分布式系统中的消息传递架构。RabbitMQ提供了许多插件来扩展其功能，其中之一是Shovel插件。本文将详细介绍RabbitMQ的Shovel插件是什么，以及如何使用它来实现消息传递。

什么是Shovel插件

Shovel 能够可靠、持续地从一个 Broker 中的队列 （作为源端，即 source）拉取数据并转发至另一个 Broker 中的交换器（作为目的端，即 destination）。作为源端的队列和作为目的端的交换器可以同时位于同一个 Broker 上，也可以位于不同的 Broker 上。Shovel 可以翻译为“铲子”，是一种比较形象的比喻，这个“铲子”可以将消息从一 方“挖到”另一方。Shovel 的行为就像优秀的客户端应用程序能够负责连接源和目的地、负责 消息的读写及负责连接失败问题的处理。

Shovel 的主要优势在于：

• 松耦合：Shovel 可以移动位于不同管理域中的 Broker（或者集群）上的消息，这些 Broker（或者集群）可以包含不同的用户和 vhost，也可以使用不同的 RabbitMQ 和 Erlang 版本。

• 支持广域网：Shovel 插件同样基于 AMQP 协议在 Broker 之间进行通信，被设计成可以容忍时断时续的连通情形，并且能够保证消息的可靠性。

• 高度定制：当 Shovel 成功连接后，可以对其进行配置以执行相关的 AMQP 命令。

Shovel 的原理

如下图：展示的是Shovel的结构示意图。

这里一共有两个Broker：broker1 和 broker2。

• broker1 中有交换器 exchange1 和队列 queue1，且这两者通 过路由键“rk1”进行绑定；

• broker2 中有交换器 exchange2 和队列 queue2，且这两者通过路由键 “rk2”进行绑定。

在队列 queue1 和交换器 exchange2 之间配置一个 Shovel link，当一条内容为 “shovel test payload”的消息从客户端发送至交换器 exchange1 的时候，这条消息会经过下图中的数据流转最后存储在队列 queue2 中。

如果在配置 Shovel link 时设置了 add_forward_headers 参数为 true，则在消费到队列 queue2 中这条消息的时候会有特殊的 headers 属性标记。

Shovel 插件使用

Shovel 既可以部署在源端，也可以部署在目的端。

有两种方式可以部署 Shovel：

• 静态方式（static）

  指在 rabbitmq.config 配置文件中设置。

• 动态方式（dynamic）

  指通过 Runtime Parameter 设置。

rabbitmq_shovel_management 提供的 Web 管理界面进行设置。其中，有两个 broker，分别为 broker1（192.168.0.175：5772）和 broker2（192.168.0.175：5773），broker1 作为源，broker2 作为目标。broker1 上面创建 Shovel，用户在 broker2 队列上发布消息，将会自动同步到 broker1 的队列中，最后进行消费。

3.1 创建两个broker-a,broker-b

3.2 broker-a开启shovel

使用浏览器访问 http://192.168.0.175:15772/地址，点击“Add a new queue”按钮去添加一个queue，如下图：

3.3 创建shovel

使用浏览器访问 http://192.168.116.51:15672/#/dynamic-shovels 地址（“Admin”>“Shovel Management”），点击“Add a new shovel”按钮去添加一个 Shovel，如下图：

上图中，添加了一个名为“shovel-demo”的 Shovel。

其中各个参数含义如下：

• Name

  表示 Shovel 的名称。

• Source

  用来定义数据的来源。首先选择的是 AMQP 的协议版本，然后就是进行 URI 的定义。URI 定义有 6 种方式：

• URI

  定义数据来源的 AMQP URL 地址。例如：amqp://guest:guest@192.168.0.175:5772

• Prefetch count

  用来指定消息可以从源目标一次传送多少消息给目的目标，默认是1000。

• Auto-delete

  是否自动删除，默认是 never。never 表示不删除，直到明确的进行删除。after initial length transferred 表示 Shovel 启动时会检查队列的长度，它将传输消息，然后删除自己。

• Routing key

  如果选择数据源为交换器 exhcange，则这里将设置路由 key。

• Destination

  目标目的的定义，它与source定义的内容差不多，没有Prefetch count和Auto-delete，多了一个Add forwarding headers，它表示是否向被铲起的消息添加报头，指示它们从何处被铲起，以及从何处被铲起。如果没有设置，则默认为false。 

• URI

  定义目标地址 AMQP 地址，例如：amqp://guest:guest@192.168.0.175:56773

• Add forwarding headers

  是否将标头添加到已 Shovel 的消息中，以指示它们已被 Shovel 和 Shovel 的位置。如果未设置，则默认为 false。

• Routing key

  如果选择交换器 exchange 作为目标，则这里将设置路由 key。

• Reconnect delay

  一个 Shovel 节点丢失后等待多长时间进行自动重连，默认是1秒。

• Acknowledgement mode

  消息确认模式，有 on-confirm、on-publish 和 no-ack 三种方式。含义分布如下：

  1）on-confirm 默认的确认方式，它需要在目的目标消息得到确认后才进行源目标消息的删除，是最可靠的消息处理方式。不管是网络错误还是消息节点失败都不会丢失消息，这种方式处理最慢。

  2）on-publish 源目标将消息发送给目的目标消息就进行确认了，这种情况在网络错误时可以进行重发，但是在消息节点失败时会丢失消息。

  3）no-ack 不需要确认就可以进行消息删除。这种方式最不安全对于消息来说，但是却是最快的。

3.4 发布消息

broker-a上的nginx queue Publish message同时broker-b上创建好nginx queue：

3.5 登录broker-b