RabbitMQ上手记录–part 6-Shovel

时间 2019-11-08

原文原文链接

上一part《RabbitMQ上手记录–part 5-节点集群高可用(多服务器)》讲到了经过多个服务器来搭建RabbitMQ的节点集群，示例当中提到的服务器都是在同一个局域网中的（其实是一个机器上的多个不一样虚拟机而已），这种使用方式适用于在同一个数据中心的状况。互联网里经常提到异地多活、多数据中心来实现更高级别的高可用。个人理解是当数据或者访问量超过当个数据中心规模时，经过更多的数据中心来提供更多的访问量支持，同时当某地数据中心出问题时，也不会让数据由于都放在同一个数据中心而致使整个系统宕机。html

RabbitMQ经过Shovel插件实现节点集群跨多数据中心的需求。下面来简单了解一下Shovel的一些基本概念。python

Shovel基本概念git

Shovel是RabbitMQ的一个插件，这个插件的功能就是将源节点的消息发布到目标节点，这个过程当中Shovel就是一个客户端，它负责链接源节点，读取某个队列的消息，而后将消息写入到目标节点的exchange中。根据这么一个概念，其实也能够本身开发一个简单的程序，负责从一个节点读取数据而后发送到目标节点。github

使用Shovel的好处web

1.Shovel能在不一样数据中心之间传递消息，源节点和目标节点可使用不一样的用户和vhosts，不一样的RabbitMQ版本，而且不须要使用相同的cookie token（在上一part实现多服务器节点集群是，咱们特地将每一个主机的cookie token都设置成同样）ubuntu

2.客户端的链接容许链接断开的同时不丢失消息安全

3.支持多个版本的AMQP协议服务器

具体工做方式cookie

Shovel插件经过定义一个或多个shovel来实现消息的传递。测试

shovel实现了如下功能

1.链接源节点和目标节点

2.读取（或者说是consume）队列里的消息

3.发布消息到目标节点（经过将消息发布到目标节点的exchange，并经过routing_key的方式发布）

（Shovel的工做过程简要描述）

使用Shovel

梳理完理论以后，接下来将使用两个不一样地域云主机来实践一下（有点小成本，须要自行租用云主机）。

1.准备云主机

1.须要有两台云主机，我这里两个云主机分别来自vulrtr和阿里云（来源不重要，只要是云主机而且分布在不一样的地区），练习用最低配的就够了。

云主机信息

名称	角色	主机提供方	操做系统	IP	端口	备注
主机1	来源节点	vultr	ubuntu1604	45.32.250.47	5672
主机2	目标节点	aliyun	ubuntu1604	47.106.179.208	5672	阿里云须要在安全策略组中单独开放5672端口

而后在各个主机安装好RabbitMQ，而且确认5672端口号可被外部访问到。

注意这里咱们并无同步两个机器的cookie token，是为了证实在使用shovel时不须要依赖于cookie token。

2.安装Shovel

Shovel是RabbitMQ的一个插件，在已经安装好RabbitMQ的基础上，把相关的插件启用便可。

咱们只须要在主机1，也就是来源节点启用shovel插件便可。

执行以下命令启用Shovel插件

rabbitmq-plugins enable rabbitmq_shovel

看到以下输出即代表启用成功

The following plugins have been enabled:
amqp_client
rabbitmq_shovel

Applying plugin configuration to rabbit@vultr... started 2 plugins.

3.配置和运行Shovel

shovel分红两种

静态shovel：在配置文件中定了源节点和目标节点信息，修改配置后须要重启

动态shovel：经过运行时参数指定，可在运行时建立或删除

这里我使用静态shovel，在配置里定义shovel配置。

首先要找到RabbitMQ使用的配置，默认状况下是没有建立的，咱们能够经过启动日志查看目前是否有指定的配置文件。

一般log文件在/var/log/rabbitmq下的rabbit@{hostname}.log文件中，打开文件发现

config file(s) : /etc/rabbitmq/rabbitmq.config (not found)

说明配置文件没有建立。

从新建立一个rabbitmq.config文件太麻烦了，咱们从基于官方提供的example配置文件来修改会简单一些。

打开目录/usr/share/doc/rabbitmq-server/

cd /usr/share/doc/rabbitmq-server/

而后找到rabbitmq.config.example.gz，解压后复制到/etc/rabbitmq目录下

gzip -dk rabbitmq.config.example.gz

mv rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

这时候的rabbitmq.config文件里全部配置都是注释的，这里咱们如今只关注shovel部分的配置。

a.准备配置信息

在主机1和主机2上分别建立一个RabbitMQ用户，用户名是shovel_user，密码是123456，并设置受权

sudo rabbitmqctl add_user shovel_user 123456

sudo rabbitmqctl set_user_tags shovel_user administrator

sudo rabbitmqctl set_permissions -p / shovel_user ".*" ".*" ".*"

b.配置shovel

在主机1上打开rabbitmq.config文件，修改shovel部分的配置为以下内容

{rabbitmq_shovel,
   [{shovels,
     [%% A named shovel worker.
      {my_test_shovel,
       [

    % List the source broker(s) from which to consume.

        {sources,
         [%% URI(s) and pre-declarations for all source broker(s).
         {brokers, ["amqp://shovel_user:123456@45.32.250.47:5672"]},
          {declarations, [
              {'exchange.declare',
                 [ {exchange, <<"shovel_exchange">>},
                 {type, <<"direct">>},
                 durable
                 ]},
             {'queue.declare',
                 [{queue,    <<"shovel_outcome_queue">>},durable]},
             {'queue.bind',
                 [ {exchange, <<"shovel_exchange">>},
                 {queue,    <<"shovel_outcome_queue">>},
                 {routing_key, <<"shovel_key">>}
                 ]}
          ]}
         ]},

    % List the destination broker(s) to publish to.
        {destinations,
         [%% A singular version of the 'brokers' element.
         {broker, "amqp://shovel_user:123456@47.106.179.208:5672"},
          {declarations, [{'exchange.declare',
                             [ {exchange, <<"shovel_exchange">>},
                             {type, <<"direct">>},
                             durable
                             ]},
                         {'queue.declare',
                             [{queue,    <<"shovel_income_queue">>},durable]},
                         {'queue.bind',
                             [ {exchange, <<"shovel_exchange">>},
                             {queue,    <<"shovel_income_queue">>},
                             {routing_key, <<"shovel_key">>}
                             ]}]}
         ]},

    % Name of the queue to shovel messages from.

      {queue, <<"shovel_outcome_queue">>},

    % Optional prefetch count.

      {prefetch_count, 10},

    % when to acknowledge messages:
     % - no_ack: never (auto)
     % - on_publish: after each message is republished
     % - on_confirm: when the destination broker confirms receipt

      {ack_mode, no_ack},

    % Overwrite fields of the outbound basic.publish.

      {publish_fields, [{exchange,    <<"shovel_exchange">>},
                        {routing_key, <<"shovel_key">>}]},

    % Static list of basic.properties to set on re-publication.

      {publish_properties, [{delivery_mode, 2}]},

    % The number of seconds to wait before attempting to
     % reconnect in the event of a connection failure.

      {reconnect_delay, 2.5}

     ]} %% End of my_first_shovel
     ]}
    %% Rather than specifying some values per-shovel, you can specify
    %% them for all shovels here.
    %%
    %% {defaults, [{prefetch_count,     0},
    %%             {ack_mode,           on_confirm},
    %%             {publish_fields,     []},
    %%             {publish_properties, [{delivery_mode, 2}]},
    %%             {reconnect_delay,    2.5}]}
   ]}

RabbitMQ的官网的shovel配置示例不可用，这里使用配置的是RabbitMQ在github提供的配置示例基础上修改的（https://github.com/rabbitmq/rabbitmq-server/blob/master/docs/rabbitmq.config.example），而后结合官网文档的说明本身摸索配置出来。

配置说明

简单介绍一下上述配置中的关键部分

shovels以后接着可定义多个shovel，这里只定义了一个shovel，名称是my_test_shovel。

sources:定义了消息的来源

brokers

须要给出来源服务的地址，一般格式为amqp://用户名:密码@主机名(IP):端口号/vhost名称。

以前定义的shovel_user这时候就能够用上了，配置中咱们使用的是默认的vhost，因此没有设置vhost名称

amqp://shovel_user:123456@45.32.250.47:5672

declarations:

declarations里面的内容就是执行一些amqp的命令，这些命令跟使用API调用的过程相似，

好比声明队列，Exchange和绑定信息等。

destinations:定义了消息的去向，里面的内容跟sources相似，实际上就是定义接收的exchange和队列

queue：这里单独配置了一个queue，是表示从哪一个队列读取消息，这里跟sources里声明的队列一致。

其余一些可选的配置就不详细介绍了，具体能够查看官网文档http://www.rabbitmq.com/configure.html。

咱们这里的配置表示从主机1的shovel_outcome_queue队列获取消息，而后转发到主机2的shovel_income_queue队列，两边使用的exchange名称都是shovel_exchange，而且routing_key的值都是shovel_key。

完整的配置请参考

https://github.com/shenba2014/RabbitMQ/blob/master/shovel/rabbitmq.config。

c.验证配置

配置完成以后，重启主机1的RabbitMQ服务

service rabbitmq-server restart

而后在主机1查看shovel的状态

rabbitmqctl eval 'rabbit_shovel_status:status().'
能够看到相似以下输出信息,看到running字样代表shovel服务已正常运行

[{my_test_shovel,static,
                  {running,[{src_uri,<<"amqp://45.32.250.47:5672">>},
                            {dest_uri,<<"amqp://47.106.179.208:5672">>}]},
                  {{2018,5,13},{12,4,48}}}]

输出信息中列出了源节点和目标节点信息，最后一行是时间戳。

4.使用Shovel

接下来咱们使用代码来测试Shovel是否可用，咱们的代码跟链接普通的RabbitMQ服务相似，只是具体链接的服务器地址不一样。

测试的代码包含consumer和producer两部分，consumer将链接主机2，producer将链接到主机1。

分别运行consumer和producer的代码，producer在运行以后会发送消息到队列shovel_outcome_queue，而后consumer会接收到消息。

具体consumer和producer的代码不贴出来，完整代码请参考

https://github.com/shenba2014/RabbitMQ/tree/master/shovel

咱们来直接看看运行代码的效果

运行消费者的代码，参数依次是：主机IP，端口，RabbitMQ用户名和密码，这里的咱们链接的是主机2（目标节点）。

运行

python shovel_consumer.py 47.106.179.208 5672 shovel_user 123456

输出

Ready for orders!

而后新开一个控制台运行生产者的代码，参数同样，可是生产者链接的是主机1（源节点）。

运行

python shovel_producer.py 45.32.250.47 5672 shovel_user 123456

输出

Sent order message.

而后在消费者的那个控制台应该会看到相似以下输出

Received order 92 for test type.

数字是随机生成的，因此最终会结果可能不一样。

好了，到目前为止，整个演练就完成了，前面的准备工做较多，测试的代码很简单，主要是演示shovel的跨数据中心的消息传递功能。

结合实际的IP地址，exchange和queue，最后用一个图来讲明使用shovel的消息流向。