> CentOS > CentOS教程 >

在CentOS上安装RabbitMQ流程

时间:2015-01-12 13:55  来源:linux.it.net.cn   作者:IT

1. 需求

   由于项目中要用到消息队列，经过ActiveMQ与RabbitMQ的比较，最终选择了RabbbitMQ做为我们的消息系统，但是ActiveMQ在效率和可扩展性上都不错，只是网上很多人反应它会时常崩溃，而且随着消息并发数的增加，时常会出现连接很慢的情况。
   目前我测试的服务器系统信息如下：
  1. LSB Version:    :core-3.1-amd64:core-3.1-ia32:core-3.1-noarch:graphics-3.1-amd64:graphics-3.1-ia32:graphics-3.1-noarch  
  2. Distributor ID: CentOS  
  3. Description:    CentOS release 5.4 (Final)  
  4. Release:    5.4  
  5. Codename:   Final  
   由于ActiveMQ是用erlang写的，这里简单介绍一下这个家伙是什么，Erlang 是由爱立信公司开发的一种平台式语言，可以说是一种自带了操作系统平台的编程语言，而且在这个平台上实现了并发机制、 进程调度、内存管理、分布式计算、网络通讯等功能，这些功能都是完全独立于用户的操作系统的，它采用的是类似于Java一样的虚拟机的方式来实现对操作系统的独立性的。
　　
　　一面是它的特点：　　
  1. 并发性：Erlang的轻量级进程可以支持极高的并发性，而且在高并发的情况下内存使用相当的少。Erlang的并发性并不会受到宿主操作系统并发性的限制。 
  2. 分布式：最开始Erlang的设计目标就是实现分布式环境，一个Erlang的虚拟机就是一个Erlang网络上的节点。一个 Erlang节点可以在另一个Erlang节点上创建自己的并发进程，而子进程所在的节点可能是运行其他的操作系统的服务器。不同节点的之间的可以进行极为高效而又精确的通信，就像它们运行在同一个节点一样。 
  3. 鲁棒形：Erlang内部建设有多种错误检测原语。我们可以通过这些原语来架设高容错性的系统。例如，一个进程可以监视其他进程的状态和活动，即使那些被监控的进程处于其他节点。在分布式状态下，我们可以把系统配置成具有Fail-over功能的分布式系统。当有其他节点出错的时候，系统会把他的运行场景自动快速的切换备份节点上。Erlang支持9个9的级别的故障率，一年只有几分钟的故障时间。 
  4. 软实时：Erlang是一个“软”实时系统（Soft Real Time），它可以提供毫秒级别的响应。

2. AMQP的一些概念 

   AMQP当中有四个概念非常重要：虚拟主机（virtual host），交换机（exchange），队列（queue）和绑定（binding）。一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，用户只能在虚拟主机的粒度进行权限控制。因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。如果这就够了，那现在就可以开始了。

交换机，队列，还有绑定……天哪！
刚开始我思维的列车就是在这里脱轨的…… 这些鬼东西怎么结合起来的？

队列（Queues）是你的消息（messages）的终点，可以理解成装消息的容器。消息就一直在里面，直到有客户端（也就是消费者，Consumer）连接到这个队列并且将其取走为止。不过。你可以将一个队列配置成这样的：一旦消息进入这个队列，biu~，它就烟消云散了。这个有点跑题了……

需要记住的是，队列是由消费者（Consumer）通过程序建立的，不是通过配置文件或者命令行工具。这没什么问题，如果一个消费者试图创建一个已经存在的队列，RabbitMQ就会起来拍拍他的脑袋，笑一笑，然后忽略这个请求。因此你可以将消息队列的配置写在应用程序的代码里面。这个概念不错。

OK，你已经创建并且连接到了你的队列，你的消费者程序正在百无聊赖的敲着手指等待消息的到来，敲啊，敲啊…… 没有消息。发生了什么？你当然需要先把一个消息放进队列才行。不过要做这个，你需要一个交换机（Exchange）……

交换机可以理解成具有路由表的路由程序，仅此而已。每个消息都有一个称为路由键（routing key）的属性，就是一个简单的字符串。交换机当中有一系列的绑定（binding），即路由规则（routes），例如，指明具有路由键 “X” 的消息要到名为timbuku的队列当中去。先不讨论这个，我们有点超前了。

你的消费者程序要负责创建你的交换机们（复数）。啥？你是说你可以有多个交换机？是的，这个可以有，不过为啥？很简单，每个交换机在自己独立的进程当中执行，因此增加多个交换机就是增加多个进程，可以充分利用服务器上的CPU核以便达到更高的效率。例如，在一个8核的服务器上，可以创建5个交换机来用5个核，另外3个核留下来做消息处理。类似的，在RabbitMQ的集群当中，你可以用类似的思路来扩展交换机一边获取更高的吞吐量。

OK，你已经创建了一个交换机。但是他并不知道要把消息送到哪个队列。你需要路由规则，即绑定（binding）。一个绑定就是一个类似这样的规则：将交换机“desert（沙漠）”当中具有路由键“阿里巴巴”的消息送到队列“hideout（山洞）”里面去。换句话说，一个绑定就是一个基于路由键将交换机和队列连接起来的路由规则。例如，具有路由键“audit”的消息需要被送到两个队列，“log-forever”和“alert-the-big-dude”。要做到这个，就需要创建两个绑定，每个都连接一个交换机和一个队列，两者都是由“audit”路由键触发。在这种情况下，交换机会复制一份消息并且把它们分别发送到两个队列当中。交换机不过就是一个由绑定构成的路由表。

现在复杂的东西来了：交换机有多种类型。他们都是做路由的，不过接受不同类型的绑定。为什么不创建一种交换机来处理所有类型的路由规则呢？因为每种规则用来做匹配分子的CPU开销是不同的。例如，一个“topic”类型的交换机试图将消息的路由键与类似“dogs.*”的模式进行匹配。匹配这种末端的通配符比直接将路由键与“dogs”比较（“direct”类型的交换机）要消耗更多的CPU。如果你不需要“topic”类型的交换机带来的灵活性，你可以通过使用“direct”类型的交换机获取更高的处理效率。那么有哪些类型，他们又是怎么处理的呢？

Fanout Exchange – 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。

Direct Exchange – 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 “dog”，则只有被标记为“dog”的消息才被转发，不会转发dog.puppy，也不会转发dog.guard，只会转发dog。

Topic Exchange – 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.*” 只会匹配到“audit.irs”。我在RedHat的朋友做了一张不错的图，来表明topic交换机是如何工作的：

3. 安装流程

   好了，回到我们的安装依赖部分，由于我们测试时装的是rabbitMQ 2.7.0版本，它依赖于erlang R12B-5以上的版本，你可以在http://www.erlang.org/download.html下载到相应的erlang的源代码，下载解析以后直接./configure，它会列出你机器上没有安装的依赖包，如我这边有crypto，ssl,ssh，wxWidget都没有安装，由于wxWidget是可选安装的，所以我这边没有安装，你可以通过yum install来安装它们，安装完以后再./configure一下，成功以后make, sudo make install就安装好了。安装好以后就可以通过在shell中输入erl命令来验证安装的成功。
   
   用wget http://www.rabbitmq.com/releases/rabbitmq-server/v2.7.0/rabbitmq-server-generic-unix-2.7.0.tar.gz下载最新的源代码，解压，在编译这前要先设置一些环境变量，一般配置方法如下
   export TARGET_DIR=/opt/rabbitmq
   export SBIN_DIR=/opt/rabbitmq/sbin
   export MAN_DIR=/opt/rabbitmq/man

   如果rabbitMQ服务只是当前用户来用的话，可以用 chown -R myuser /opt/rabbitmq 命令来改变其目录权限

   下来是运行make与make install命令来安装。

   最后还要设置一下日志与消息持久化目录，命令如下，其中的myuser是你当前的用户名
 
  1. mkdir /var/log/rabbitmq  
  2. chown myuser /var/log/rabbitmq  
  3. mkdir /var/lib/rabbitmq  
  4. chown myuser /var/lib/rabbitmq  
  现在RabbitMQ已经安装好了，现在要启动它了，是不是好马，遛不才知道啊，它的启动也很简单，运行命令
 
  1. cd /opt/rabbitmq/sbin  
  2. ./rabbitmq-server  
   当然你会为启动方便，可以把/opt/rabbitmq/sbin下的命令在/usr/local/bin下做一个链接，命令如下
   sudo ln -s /opt/rabbitmq/sbin/rabbitmq-server /usr/local/bin/rabbitmq-server
   其它命令也一样，这样你就可以在任何地方使用rabbitmq-server命令了。

   这样就启动了，是不是很简单，默认是监听当前的5672端口的，而且默认也是不需要配置的，当然你也可以进行相应的配置，具体可以参考如下：     http://www.rabbitmq.com/configure.html
   
   启动以后会打印出如下信息 
  1.   Activating RabbitMQ plugins ...  
  2. 0 plugins activated:  
  3. +---+   +---+  
  4. |   |   |   |  
  5. |   |   |   |  
  6. |   |   |   |  
  7. |   +---+   +-------+  
  8. |                   |  
  9. | RabbitMQ  +---+   |  
  10. |           |   |   |  
  11. |   v2.7.0  +---+   |  
  12. |                   |  
  13. +-------------------+  
  14. AMQP 0-9-1 / 0-9 / 0-8  
  15. Copyright (C) 2007-2011 VMware, Inc.  
  16. Licensed under the MPL.  See http://www.rabbitmq.com/  
  17.   
  18.   
  19. node           : rabbit@xunuu62  
  20. app descriptor : /opt/rabbitmq/sbin/../ebin/rabbit.app  
  21. home dir       : /home/xunuu  
  22. config file(s) : (none)  
  23. cookie hash    : 4OkU6/5RZ9ck4GPo4zyaKw==  
  24. log            : /var/log/rabbitmq/rabbit@xunuu62.log  
  25. sasl log       : /var/log/rabbitmq/rabbit@xunuu62-sasl.log  
  26. database dir   : /var/lib/rabbitmq/mnesia/rabbit@xunuu62  
  27. erlang version : 5.7.4  
  28.   
  29.   
  30. -- rabbit boot start  
  31. starting file handle cache server                                     ...done  
  32. starting worker pool                                                  ...done  
  33. starting database                                                     ...done  
  34. starting codec correctness check                                      ...done  
    RabbitMQ的管理插件的安装：
    你可以用如下命令安装RabbitMQ的管理插件  
 
  1. rabbitmq-plugins enable rabbitmq_management  
    我这边输出如下： 
  1. [www.codesky.net@linuxidc rabbitmq-server-2.7.0]$ rabbitmq-plugins enable rabbitmq_management  
  2. The following plugins have been enabled:  
  3.   mochiweb  
  4.   webmachine  
  5.   rabbitmq_mochiweb  
  6.   amqp_client  
  7.   rabbitmq_management_agent  
  8.   rabbitmq_management  
  9.   
  10. Plugin configuration has changed. Restart RabbitMQ for changes to take effect.  
    装好以后要重启rabbitMQ服务，关闭服务命令如下 
  1. rabbitmqctl stop  
    再运行启动命令： 
  1. rabbit-server &  
    这样你就可以用浏览器访问 http://server-name:55672/mgmt/ 来进行相应的管理 ，这里的的默认用户名密码都是guest

4. 测试 

    这里有两种测试方法，可以在刚才安装的管理里面进行测试，也可以通过客户端来进行测试，RabbitMQ支持丰富的客户端语言

    这里有相应的工具下载
    http://www.rabbitmq.com/devtools.html

    我们这里使用python来做一下测试，因为其代码比较简洁。

    首先，我们需要一个Python的AMQP库。有两个可选：

py-amqplib – 通用的AMQP
txAMQP – 使用 Twisted 框架的AMQP库，因此允许异步I/O。
根据你的需求，py-amqplib或者txAMQP都是可以的。因为是基于Twisted的，txAMQP可以保证用异步IO构建超高性能的AMQP程序。但是Twisted编程本身就是一个很大的主题……因此清晰起见，我们打算用 py-amqplib。更新：请参见Esteve Fernandez关于txAMQP的使用和代码样例的回复。

AMQP支持在一个TCP连接上启用多个MQ通信channel，每个channel都可以被应用作为通信流。每个AMQP程序至少要有一个连接和一个channel。 
  1. from amqplib import client_0_8 as amqp  
  2. conn = amqp.Connection(host="localhost:5672 ", userid="guest",  
  3. password="guest", virtual_host="/", insist=False)  
  4. chan = conn.channel()  
每个channel都被分配了一个整数标识，自动由Connection()类的.channel()方法维护。或者，你可以使用.channel(x)来指定channel标识，其中x是你想要使用的channel标识。通常情况下，推荐使用.channel()方法来自动分配channel标识，以便防止冲突。

现在我们已经有了一个可以用的连接和channel。现在，我们的代码将分成两个应用，生产者（producer）和消费者（consumer）。我们先创建一个消费者程序，他会创建一个叫做“po_box”的队列和一个叫“sorting_room”的交换机： 
  1. chan.queue_declare(queue="po_box", durable=True,  
  2. exclusive=False, auto_delete=False)  
  3. chan.exchange_declare(exchange="sorting_room", type="direct", durable=True,  
  4. auto_delete=False,)  
这段代码干了啥？首先，它创建了一个名叫“po_box”的队列，它是durable的（重启之后会重新建立），并且最后一个消费者断开的时候不会自动删除（auto_delete=False）。在创建durable的队列（或者交换机）的时候，将auto_delete设置成false是很重要的，否则队列将会在最后一个消费者断开的时候消失，与durable与否无关。如果将durable和auto_delete都设置成True，只有尚有消费者活动的队列可以在RabbitMQ意外崩溃的时候自动恢复。

（你可以注意到了另一个标志，称为“exclusive”。如果设置成True，只有创建这个队列的消费者程序才允许连接到该队列。这种队列对于这个消费者程序是私有的）。

还有另一个交换机声明，创建了一个名字叫“sorting_room”的交换机。auto_delete和durable的含义和队列是一样的。但是，.excange_declare() 还有另外一个参数叫做type，用来指定要创建的交换机的类型（如前面列出的）： fanout, direct 和 topic.

到此为止，你已经有了一个可以接收消息的队列和一个可以发送消息的交换机。不过我们需要创建一个绑定，把它们连接起来。 
  1. chan.queue_bind(queue=”po_box”, exchange=”sorting_room”,  
  2. routing_key=”jason”)  
这个绑定的过程非常直接。任何送到交换机“sorting_room”的具有路由键“jason” 的消息都被路由到名为“po_box” 的队列。

现在，你有两种方法从队列当中取出消息。第一个是调用chan.basic_get()，主动从队列当中拉出下一个消息（如果队列当中没有消息，chan.basic_get()会返回None， 因此下面代码当中print msg.body 会在没有消息的时候崩掉）： 
  1. msg = chan.basic_get("po_box")  
  2. print msg.body  
  3. chan.basic_ack(msg.delivery_tag)  
但是如果你想要应用程序在消息到达的时候立即得到通知怎么办？这种情况下不能使用chan.basic_get()，你需要用chan.basic_consume()注册一个新消息到达的回调。 
  1. def recv_callback(msg):  
  2.     print 'Received: ' + msg.body  
  3. chan.basic_consume(queue='po_box', no_ack=True,  
  4. callback=recv_callback, consumer_tag="testtag")  
  5. while True:  
  6.     chan.wait()  
  7. chan.basic_cancel("testtag")  
chan.wait() 放在一个无限循环里面，这个函数会等待在队列上，直到下一个消息到达队列。chan.basic_cancel() 用来注销该回调函数。参数consumer_tag 当中指定的字符串和chan.basic_consume() 注册的一直。在这个例子当中chan.basic_cancel() 不会被调用到，因为上面是个无限循环…… 不过你需要知道这个调用，所以我把它放在了代码里。

需要注意的另一个东西是no_ack参数。这个参数可以传给chan.basic_get()和chan.basic_consume()，默认是false。当从队列当中取出一个消息的时候，RabbitMQ需要应用显式地回馈说已经获取到了该消息。如果一段时间内不回馈，RabbitMQ会将该消息重新分配给另外一个绑定在该队列上的消费者。另一种情况是消费者断开连接，但是获取到的消息没有回馈，则RabbitMQ同样重新分配。如果将no_ack 参数设置为true，则py-amqplib会为下一个AMQP请求添加一个no_ack属性，告诉AMQP服务器不需要等待回馈。但是，大多数时候，你也许想要自己手工发送回馈，例如，需要在回馈之前将消息存入数据库。回馈通常是通过调用chan.basic_ack()方法，使用消息的delivery_tag属性作为参数。参见chan.basic_get() 的实例代码。

好了，这就是消费者的全部代码。（下载：amqp_consumer.py http://blogs.digitar.com/jjww/code-samples/amqp_consumer.py）

不过没有人发送消息的话，要消费者何用？所以需要一个生产者。下面的代码示例表明如何将一个简单消息发送到交换区“sorting_room”，并且标记为路由键“jason” ： 
  1. msg = amqp.Message("Test message!")  
  2. msg.properties["delivery_mode"] = 2  
  3. chan.basic_publish(msg,exchange="sorting_room",routing_key="jason")  
你也许注意到我们设置消息的delivery_mode属性为2，因为队列和交换机都设置为durable的，这个设置将保证消息能够持久化，也就是说，当它还没有送达消费者之前如果RabbitMQ重启则它能够被恢复。

剩下的最后一件事情（生产者和消费者都需要调用的）是关闭channel和连接： 
  1. chan.close()  
  2. conn.close()  
很简单吧。（下载：amqp_publisher.py http://blogs.digitar.com/jjww/code-samples/amqp_publisher.py）

来真实地跑一下吧……
现在我们已经写好了生产者和消费者，让他们跑起来吧。假设你的RabbitMQ在localhost上安装并且运行。

打开一个终端，执行python ./amqp_consumer.py让消费者运行，并且创建队列、交换机和绑定。

然后在另一个终端运行python ./amqp_publisher.py “AMQP rocks.” 。如果一切良好，你应该能够在第一个终端看到输出的消息。


(责任编辑：IT)
    最近关注
    热点内容