Ruby客户端操作 rabbitmq-server 技巧

RabbitMQ 是一种消息队列，充当接收和传送消息的中间角色。实现了多应用之间相互通讯使用生产者消费者模式。其支持多平台，Linux macOS window ，同时也支持多种语言 Python, Java, Ruby, PHP。其使用场景比较多使用与现阶段讨论比较多的为服务中，把一个大的应用拆分成多个服务，其中使用 RabbitMQ 作为消息队列服务之间进行通信。在这里我们讨论的是如何使用 ruby 实现一个简单的消息队列系统，读者可举一反三。

安装 RabbitMQ

macOS 根据官方文档，跟着一步一步来即可。

Ubuntu rabbitmq-server 依赖 erlang 环境

sudo apt-get update
sudo apt-get install erlang
sudo apt-get install erlang-nox
sudo apt-get install rabbitmq-server

安装文档

开启 Web 管理页面 sudo rabbitmq-plugins enable rabbitmq_management

The following plugins have been enabled:
  mochiweb
  webmachine
  rabbitmq_web_dispatch
  amqp_client
  rabbitmq_management_agent
  rabbitmq_management

Applying plugin configuration to rabbit@i-l9ptvup7... started 6 plugins.

重启 Rabbitmq-server : cd /etc/init.d ; ./rabbitmq-server restart 安装完毕之后，你可以通过浏览器访问 http://localhost:15672/ 其中有个密码和用户名相同的默认的 guest用户可以登录管理页面，并且该用户只能通过本地 localhost 访问，远程访问是不行的。
通过 web 图形化页面可以添加用户，另外也可以通过命令行添加

sudo rabbitmqctl add_user test test
sudo rabbitmqctl set_user_tags test administrator
sudo rabbitmqctl set_permissions -p / test ".*" ".*" ".*"

关于服务的一些说明

发送消息服务，接收消息服务以及 rabbitmq-server 服务可以部署在不同的机器上，当然也可部署在同一台服务器上。有一个场景：譬如我有一个区块链钱包服务（消费者），同时区块链上交易时会发送一个回调（生产者），那么这两个服务可以通过 rabbitmq-server 进行通信。这三个服务能通过远程进行调用从而不需要部署在同一台机器上。

发送消息（生产者）

Ruby 有对应的 gem 客户端能和 rabbitmq-server 通信 bunny，安装以及快速使用在 Github 上看文档即可。

#!/usr/bin/env ruby
## encoding: utf-8
## send.rb

require "bunny"
class Send
    def self.publish(exchange, message = "hellooooo")
        x = channel.fanout("blog.#{exchange}")
        x.publish(message)
    end

    def self.channel
        @channel ||= connection.create_channel
    end

    def self.connection
        @conn = Bunny.new(:automatically_recover => false)
        @conn.start
    end
end

Send.publish("posts", "hello world ..........")

上面这段代码中，产生了一个 blog:posts 扇形(fanout)交换机。生产者的服务和 rabbitmq-server 服务部署在同一台机器，如果在不同机器，要把

@conn = Bunny.new(:automatically_recover => false)

替换成:

@conn = Bunny.new( :host => "remote_server_ip", :port => "5672", :user => "user_name", :password => "user_passowrd", :automatically_recover => false)

那么到此一个简单的发送 hello world 生产者就构建完成了。
通过访问 http://localhost:15672/#/exchanges，可以看到 blog:posts 交换机

后台队列接收消息（消费者）

如果你用过 sidekiq, 那么这里讨论的内容有点类似，只不过这里的后台队列处理的是消息队列中的信息。
在这小节中，我们需要用到另外一个 gem sneakers，它的作用和 sidekiq 非常类似，只不过它处理的是从消息队列中传过来的信息。其描述是：

A fast background processing framework for Ruby and RabbitMQ http://sneakers.io

安装以及快速使用在 Github 上查阅 readme 和 wiki 即可。

#!/usr/bin/env ruby
## encoding: utf-8
## boot.rb
require 'sneakers'
Sneakers.configure  ({
    :amqp => 'amqp://user_name:user_password@localhost:5672',
    :exchange => 'sneakers',
    :exchange_type => :direct
})
Sneakers.logger.level = Logger::INFO
class Processor
    include Sneakers::Worker
    from_queue "dashboard.posts"
    def work(msg)
        p msg
        ack!
    end
end

上面这段代码中，产生了一个 dashboard.posts 队列。访问 http://localhost:15672/#/queues 可以看到。
如果执行脚本 sneakers work Processor --require boot.rb,在管理页面可以看到

但是这个队列没有和任何交换机绑定，此刻发送者和接收者还没产生关系，如果两者要通信，就要把交换机和队列绑定。
这里的绑定有一个脚本

require "bunny"
conn = Bunny.new
conn.start
ch = conn.create_channel
# get or create exchange
x = ch.fanout("blog.posts")
# get or create queue (note the durable setting)
queue = ch.queue("dashboard.posts", durable: true)
# bind queue to exchange
queue.bind("blog.posts")
conn.close

如果接收方服务是一个 Rails 应用，那么写个 rake 任务

namespace :rabbitmq do
  desc "Setup RabbitMQ routing"
  task :setup do
    require "bunny"

    conn = Bunny.new
    conn.start

    ch = conn.create_channel

    # get or create exchange
    x = ch.fanout("blog.posts")

    # get or create queue (note the durable setting)
    queue = ch.queue("dashboard.posts", durable: true)

    # bind queue to exchange
    queue.bind("blog.posts")

    conn.close
  end
end

执行脚本或者任务(rake rabbitmq:setup) 就能把交换机和队列绑定。此时再访问http://localhost:15672/#/queues 可以看到。

绑定队列

完整的生产消费过程

做完上面所有的描述以后，咱们就可以看到 hello world 了，具体如下：

启动接收服务 sneakers work Processor --require boot.rb
绑定队列和交换机 rake rabbitmq:setup 在这里是 Rails rake 任务
发送消息 ruby send.rb