zoukankan      html  css  js  c++  java
  • RabbitMQ上手记录–part 6-Shovel

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

     

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

     

    Shovel基本概念

    Shovel是RabbitMQ的一个插件,这个插件的功能就是将源节点的消息发布到目标节点,这个过程中Shovel就是一个客户端,它负责连接源节点,读取某个队列的消息,然后将消息写入到目标节点的exchange中。根据这么一个概念,其实也可以自己开发一个简单的程序,负责从一个节点读取数据然后发送到目标节点。

     

    使用Shovel的好处

    1.Shovel能在不同数据中心之间传递消息,源节点和目标节点可以使用不同的用户和vhosts,不同的RabbitMQ版本,并且不需要使用相同的cookie token(在上一part实现多服务器节点集群是,我们特意将每个主机的cookie token都设置成一样)

    2.客户端的连接允许连接断开的同时不丢失消息

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

     

    具体工作方式

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

    shovel实现了以下功能

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

    2.读取(或者说是consume)队列里的消息

    3.发布消息到目标节点(通过将消息发布到目标节点的exchange,并通过routing_key的方式发布)

     

    image(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的消息流向。

    image

     

     

     

  • 相关阅读:
    安卓adb
    图数据库学习
    分布式架构演进
    多活架构
    异地多活
    分布式CRDT
    技术架构的战略和战术原则
    分布式金融架构课笔记
    使用jackson进行json序列化时进行敏感字段脱敏(加密)或者忽略
    读书笔记《演进式架构》
  • 原文地址:https://www.cnblogs.com/shenba/p/9033423.html
Copyright © 2011-2022 走看看