Error: Package: rabbitmq-server-3.6.1-1.noarch (/rabbitmq-server-3.6.1-1.noarch) 
Requires: erlang >= R16B-3 
You could try using --skip-broken to work around the problem 
You could try running: rpm -Va --nofiles --nodigest

这是由于erlang的版本问题，其实是没有影响的，你可以使用下面的命令进行安装：

#rpm -ivh --nodeps rabbitmq-server-3.6.1-1.noarch.rpm

启动

#service rabbitmq-server start --后台方式运行

#service rabbitmq-server stop  --停止运行

#service rabbitmq-server status --查看状态

#rabbitmq-server start 

可以看到使用的日志文件

日志目录

/var/log/rabbitmq

#cat /var/log/rabbitmq/rabbit@iZ94nxslz66Z.log 可以看到下面的日志记录

...................................................................................................................................................................................................................................................

卸载

管理

Rabbitmq服务器的主要通过rabbitmqctl和rabbimq-plugins两个工具来管理，以下是一些常用功能。

1）. 服务器启动与关闭

      启动: rabbitmq-server –detached

      关闭:rabbitmqctl stop

      若单机有多个实例，则在rabbitmqctlh后加–n 指定名称

2）. 插件管理

      开启某个插件：rabbitmq-pluginsenable xxx

      关闭某个插件：rabbitmq-pluginsdisablexxx

      注意：重启服务器后生效。

3）.virtual_host管理

      新建virtual_host: rabbitmqctladd_vhost  xxx

      撤销virtual_host:rabbitmqctl  delete_vhost xxx

4）. 用户管理

      新建用户：rabbitmqctl add_user xxxpwd

      删除用户:   rabbitmqctl delete_user xxx

      改密码: rabbimqctlchange_password {username} {newpassword}

      设置用户角色：rabbitmqctlset_user_tags {username} {tag ...}

              Tag可以为 administrator,monitoring, management

5）. 权限管理

      权限设置：set_permissions [-pvhostpath] {user} {conf} {write} {read}

               Vhostpath

               Vhost路径

               user

      用户名

              Conf

      一个正则表达式match哪些配置资源能够被该用户访问。

              Write

      一个正则表达式match哪些配置资源能够被该用户读。

               Read

      一个正则表达式match哪些配置资源能够被该用户访问。

6）. 获取服务器状态信息

       服务器状态：rabbitmqctl status

       队列信息：rabbitmqctl list_queues[-p vhostpath] [queueinfoitem ...]

                Queueinfoitem可以为：name，durable，auto_delete，arguments，messages_ready，

                messages_unacknowledged，messages，consumers，memory

       Exchange信息：rabbitmqctllist_exchanges[-p vhostpath] [exchangeinfoitem ...]

                 Exchangeinfoitem有：name，type，durable，auto_delete，internal，arguments.

       Binding信息：rabbitmqctllist_bindings[-p vhostpath] [bindinginfoitem ...]       

                 Bindinginfoitem有：source_name，source_kind，destination_name，destination_kind，routing_key，arguments

       Connection信息：rabbitmqctllist_connections [connectioninfoitem ...]

       Connectioninfoitem有：recv_oct，recv_cnt，send_oct，send_cnt，send_pend等。

       Channel信息：rabbitmqctl  list_channels[channelinfoitem ...]

      Channelinfoitem有consumer_count，messages_unacknowledged，messages_uncommitted，acks_uncommitted，messages_unconfirmed，prefetch_count，client_flow_blocked

常用命令：

查看所有队列信息

# rabbitmqctl list_queues

关闭应用

# rabbitmqctl stop_app

启动应用，和上述关闭命令配合使用，达到清空队列的目的

# rabbitmqctl start_app

清除所有队列

# rabbitmqctl reset

更多用法及参数，可以执行如下命令查看

# rabbitmqctl

rabbitmq常用命令

rabbitmq-server start  或者   service rabbitmq-server start     #启动rabbitmq

rabbitmqctl list_exchanges 

rabbitmqctl list_bindings

rabbitmqctl list_queues #分别查看当前系统种存在的Exchange和Exchange上绑定的Queue信息。

rabbitmqctl status  #查看运行信息

rabbitmqctl stop     #停止运行rabbitmq

rabbitmq-plugins enable rabbitmq_management  

#启动rabbitmq的图形管理界面，这个操作必须重启rabbitmq， 然后在web中 http://127.0.0.1:15672 用户名和密码都是guest guest。如果局域网无法访问设置防火墙过滤规则或关闭防火墙。

什么是复制?

RabbitMQ broker操作所需的所有数据/状态都可以在多个节点间复制. 例外是消息队列,默认情况下它驻留在一个节点， 尽管它们对所有节点来说，是可见的，可达的.要在集群中跨节点复制队列，可参考high availability 文档(注意，你仍然先需要一个工作集群).

主机名解析需求

RabbitMQ节点彼此之间使用域名，要么是简短的，要么是全限定的(FQDNs). 因此，集群中所有成员的主机名都必须是可解析的，也可用于机器上的命令行工具，如rabbitmqctl.

主机名解析可使用任何一种标准的操作系统提供方法:

• DNS 记录
• 本地主机文件(e.g. /etc/hosts)

要使用FQDNs, 参考RABBITMQ_USE_LONGNAME in the Configuration guide.

集群构成

集群可以通过多种方式来构建:

• 手动使用rabbitmqctl
• 通过在配置文件列举集群节点来声明
• 通过使用rabbitmq-autocluster (插件)来声明
• 使用rabbitmq-clusterer（插件）来声明

一个集群的构成可以动态修改. 所有RabbitMQ brokers开始都是以单个节点来运行的. 这些节点可以加入到集群中, 随后也可以脱离集群再次成为单一节点。

故障处理

RabbitMQ brokers 可以容忍个别节点故障. 节点可以随意地启动和关闭，只要在已知关闭的时间内能够联系到集群节点.

RabbitMQ 集群有多种模式来处理网络分化, 主要是一致性方向. 集群是在LAN中使用的，不推荐在WAN中运行集群. Shovel 或 Federation 插件对于跨WAN连接brokers ，有更好的解决方案. 注意 Shovel 和 Federation 不等同于集群.

磁盘和内存节点

节点可以是磁盘节点，也可以是内存节点。多数情况下，你希望所有的节点都是磁盘节点，但RAM节点是一种特殊情况，它可以提高集群中队列和，交换机，绑定的性能. 当有疑问时，最好只使用磁盘节点。

节点(以及CLI工具)之间如何来认证: Erlang Cookie

RabbitMQ 节点和CLI 工具(如rabbitmqctl) 使用cookie来确定每个节点之间是否可以通信. 两个节点之间要能通信，它们必须要有相同的共享密钥Erlang cookie. cookie只是具有字母数字特征的字符串。只要你喜欢，它可长可短. 每个集群节点必须有相同的cookie.

当RabbitMQ 服务器启动时，Erlang VM 会自动地创建一个随机的cookie文件. 最简单的处理方式是允许一个节点来创建文件，然后再将这个文件拷贝到集群的其它节点中。

在 Unix 系统中, cookie的通常位于/var/lib/rabbitmq/.erlang.cookie 或$HOME/.erlang.cookie.

在Windows中, 其位置在C:\Users\Current User\.erlang.cookie(%HOMEDRIVE% + %HOMEPATH%\.erlang.cookie) 或C:\Documents and Settings\Current User\.erlang.cookie, 对于RabbitMQ Windows service其位置在C:\Windows\.erlang.cookie。如果使用了Windows service ,  cookie可被放于这两个位置中.

作为替代方案，你可以在 rabbitmq-server 和 rabbitmqctl 脚本中调用erl时,插入"-setcookie cookie"选项.

当cookie未配置时 (例如，不相同), RabbitMQ 会记录这样的错误"Connection attempt from disallowed node" and "Could not auto-cluster".

启动独立节点

集群可通过重新配置，而将现有RabbitMQ 节点加入到集群配置中. 因此第一步是以正常的方式在所有节点上启动RabbitMQ:

rabbit1$ rabbitmq-server -detached 
rabbit2$ rabbitmq-server -detached 
rabbit3$ rabbitmq-server -detached

这会创建三个独立的RabbitMQ brokers, 每个节点一个，可通过cluster_status命令来验证:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1]}]},{running_nodes,[rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit2]}]},{running_nodes,[rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit3]}]},{running_nodes,[rabbit@rabbit3]}] ...done.

以rabbitmq-server shell脚本来启动RabbitMQ broker的节点名称是rabbit@shorthostname,在这里，短节点名称是小写的(如上面的rabbit@rabbit1). 如果在windows上，你使用rabbitmq-server.bat批处理文件来启动，短节点名称是大写的（如：rabbit@RABBIT1). 当你输入节点名称时，不论是大写还是小写的，这些字符串都必须精确匹配。

创建集群

为了把这三个节点构建到一个集群中，我们可以告诉其中的两个节点, 假设为rabbit@rabbit2 和 rabbit@rabbit3, 将加入到第三个节点的集群中，这第三个节点假设为rabbit@rabbit1.

首先我们将rabbit@rabbit2加入到rabbit@rabbit1的集群中. 要做到这一点，我们必须在rabbit@rabbit2 上停止RabbitMQ应用程序，并将其加入到rabbit@rabbit1 集群中, 然后再重启RabbitMQ 应用程序. 

注意：加入集群会隐式地重置节点, 因此这会删除此节点上先前存在的所有资源和数据.（如何备份数据）

rabbit2$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit2 ...done. 
rabbit2$ rabbitmqctl join_cluster rabbit@rabbit1 
Clustering node rabbit@rabbit2 with [rabbit@rabbit1] ...done. 
rabbit2$ rabbitmqctl start_app 
Starting node rabbit@rabbit2 ...done.

在每个节点上通过运行cluster_status 命令，我们可以看到两个节点已经加入了集群:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2]}] ...done. 

现在我们将rabbit@rabbit3节点加入到同一个集群中. 操作步骤同上面的一致,除了这次我们选择rabbit2来加入集群，但这并不重要：

rabbit3$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit3 ...done. 
rabbit3$ rabbitmqctl join_cluster rabbit@rabbit2 
Clustering node rabbit@rabbit3 with rabbit@rabbit2 ...done. 
rabbit3$ rabbitmqctl start_app 
Starting node rabbit@rabbit3 ...done.

在任何一个节点上通过运行cluster_status命令，我们可以看到三个节点已经加入了集群:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit3,rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit3,rabbit@rabbit1,rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit3,rabbit@rabbit2,rabbit@rabbit1]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1,rabbit@rabbit3]}] ...done.

通过上面的步骤，当集群运行的时候，我们可以在任何时候将新的节点加入到集群中.

重启集群节点

注意，加入到集群中的节点可在任何时候停止， 对于崩溃来说也没有问题. 在这两种情况下，集群剩余的节点将不受影响地继续操作，当它们重启的时候，这些崩溃的节点会再次自动追赶上其它的集群节点。

我们关闭了节点rabbit@rabbit1和rabbit@rabbit3，并在每步观察集群的状态:

rabbit1$ rabbitmqctl stop 
Stopping and halting node rabbit@rabbit1 ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit3,rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit3]}] ...done. 
rabbit3$ rabbitmqctl stop 
Stopping and halting node rabbit@rabbit3 ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2]}] ...done.

译者注：关闭了rabbit1节点后，运行的节点已经没有rabbit1节点了

现在我们再次启动节点,并检查集群状态:

rabbit1$ rabbitmq-server -detached 
rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmq-server -detached 
rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1,rabbit@rabbit3]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1,rabbit@rabbit3]}] ...done.

这里有一些重要的警告:

• 当整个集群崩溃的时候， 最后一个崩溃的节点必须第一个上线.如果不是这样，节点将会等待最后一个磁盘节点30秒以确认其重新上线，否则就会失败. 如果最后一个下线的节点，不能再重新上线，那么它可能会使用forget_cluster_node命令来从集群中删除 - 查阅 rabbitmqctl页面来了解更多信息.
• 如果所有集群节点都在同一个时间内停止且不受控制（如断电）。在这种情况下，你可以在某个节点上使用force_boot命令使其再次成为可启动的－查阅 rabbitmqctl页面来了解更多信息.

脱离集群

当节点不再是集群的一部分时，可以明确地将其从集群中删除. 首先我们将节点rabbit@rabbit3从集群中删除, 以使其回归独立操作.要做到这一点，需要在rabbit@rabbit3节点上停止RabbitMQ 应用程序,重设节点，并重启RabbitMQ应用程序.

rabbit3$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit3 ...done. 
rabbit3$ rabbitmqctl reset 
Resetting node rabbit@rabbit3 ...done. 
rabbit3$ rabbitmqctl start_app 
Starting node rabbit@rabbit3 ...done.

在节点上运行cluster_status 命令来确认rabbit@rabbit3节点现在已不再是集群的一部分，并且会独自操作:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit3]}]},{running_nodes,[rabbit@rabbit3]}] ...done. 

我们也可以远程地删除节点，这是相当有用的，举例来说，当处理无反应的节点时.举例来说，我们可以从 rabbit@rabbit2中删除rabbit@rabbi1.

rabbit1$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit1 ...done. 
rabbit2$ rabbitmqctl forget_cluster_node rabbit@rabbit1 
Removing node rabbit@rabbit1 from cluster ... ...done.

注意，rabbit1仍然认为它与rabbit2处在一个集群中，但尝试启动时会出现一个错误.这时，我们需要对其进行重置以使其能再次启动.

rabbit1$ rabbitmqctl start_app 
Starting node rabbit@rabbit1 ... Error: inconsistent_cluster: Node rabbit@rabbit1 thinks it's clustered with node rabbit@rabbit2, but rabbit@rabbit2 disagrees 
rabbit1$ rabbitmqctl reset 
Resetting node rabbit@rabbit1 ...done. 
rabbit1$ rabbitmqctl start_app Starting node rabbit@mcnulty ... ...done.

现在， cluster_status 命令会显示三个节点都是独立节点，并且操作是独立的:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1]}]},{running_nodes,[rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit2]}]},{running_nodes,[rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit3]}]},{running_nodes,[rabbit@rabbit3]}] ...done.

注意：rabbit@rabbit2节点仍然残留有集群的状态（译者注：怎么看出来的呢？）, 但是 rabbit@rabbit1 和rabbit@rabbit3 节点是新鲜的RabbitMQ brokers.如果我们想重新初始化rabbit@rabbit2节点，我们可以按其它节点的步骤来操作:

rabbit2$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit2 ...done. 
rabbit2$ rabbitmqctl reset 
Resetting node rabbit@rabbit2 ...done. 
rabbit2$ rabbitmqctl start_app 
Starting node rabbit@rabbit2 ...done.

创建内存节点

当节点加入集群时，我们可将其声明为内存节点. 我们可以通过使用像先前rabbitmqctl join_cluster命令再加--ram标志来达到目的:

rabbit2$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit2 ...done. 
rabbit2$ rabbitmqctl join_cluster --ram rabbit@rabbit1 
Clustering node rabbit@rabbit2 with [rabbit@rabbit1] ...done. 
rabbit2$ rabbitmqctl start_app Starting node rabbit@rabbit2 ...done.

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1]},{ram,[rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1]},{ram,[rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2]}] ...done. 

改变节点类型

我们可以将节点的类型从磁盘修改为内存，反之亦然. 假设我们想反转rabbit@rabbit2 和 rabbit@rabbit1的节点类型,即先将内存节点转换为磁盘节点，随后再将其从磁盘节点转换为内存节点.要做到这点，我们可以使用change_cluster_node_type命令. 首先节点必须先停止.

rabbit2$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit2 ...done. rabbit2$ 
rabbitmqctl change_cluster_node_type disc 
Turning rabbit@rabbit2 into a disc node ... ...done. Starting node rabbit@rabbit2 ...done. 
rabbit1$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit1 ...done. 
rabbit1$ rabbitmqctl change_cluster_node_type ram 
Turning rabbit@rabbit1 into a ram node ... 
rabbit1$ rabbitmqctl start_app 
Starting node rabbit@rabbit1 ...done.

1. Producer.java
2. Consumer.java
3. GetOne.java

1. 绑定到一个临时队列，即调用无参的channel.queueDeclare()方法
2. 绑定到一个非自动删除的命名队列
3. 绑定到一个非自动删除的命名队列，并且指定x-message-ttl ,如步骤2中展示的一样.

1. Producer.java
2. Consumer.java

1. 使用arguments.put("x-message-ttl", 10000)来设置消息TTL ,正如如何使指定队列上消息过期食谱中描述的一样.
2. 使用arguments.put("x-dead-letter-exchange", exchange_dead_letter)来设置死信交换器名称;

1. reason : 表示队列是否过期的或驳回的(requeue =false )
2. queue : 表示队列源，例如stat_queue_02/05
3. time : 表示消息变为死信的日期和时间
4. exchange : 表示交换器，如monitor_exchange_02/05
5. routing-keys : 表示发送消息时原先使用的路由键

1. ref_exchange_c1_6 (广播消息) 与exchange绑定.
2. ref_exchange_c1_8 (使用topic来处理消息路由)与exchange绑定 .

1. Producer.java
2. Consumer.java
3. StatsConsumer.java
4. CCStatsConsumer.java
5. BCCStatsConsumer.java

AMQP client 关闭概述

AMQP 0-9-1 connection和channel 使用相同的方法来管理网络故障,内部故障,以及显示本地关闭.

AMQP 0-9-1 connection  和 channel 有如下的生命周期状态:

• open: 准备要使用的对象
• closing: 对象已显示收到收到本地关闭通知, 并向任何支持的底层对象发出关闭请求,并等待其关闭程序完成
• closed: 对象已收到所有底层对象的完成关闭通知,最终将执行关闭操作

这些对象总是以closed状态结束的,不管基于什么原因引发的关闭,比如:应用程序请求,内部client library故障, 远程网络请求或网络故障.

AMQP connection 和channel 对象会持有下面与关闭相关的方法:

• addShutdownListener(ShutdownListener 监听器)和removeShutdownListener(ShutdownListener 监听器),用来管理监听器,当对象转为closed状态时,将会触发这些监听器.注意,在已经关闭的对象上添加一个ShutdownListener将会立即触发监听器
• getCloseReason(), 允许同其交互以了解对象关闭的理由
• isOpen(), 用于测试对象是否处于open状态
• close(int closeCode, String closeMessage), 用于显示通知对象关闭

可以像这样来简单使用监听器:

import com.rabbitmq.client.ShutdownSignalException; 
import com.rabbitmq.client.ShutdownListener;  
connection.addShutdownListener(new ShutdownListener() {     
public void shutdownCompleted(ShutdownSignalException cause)     {         ...     } }
);

关闭环境信息

可通过显示调用getCloseReason()方法或通过使用ShutdownListener类中的业务方法的cause参数来从ShutdownSignalException中获取关闭原因的有用信息.

ShutdownSignalException 类提供方法来分析关闭的原因.通过调用isHardError()方法,我们可以知道是connection错误还是channel错误.getReason()会返回相关cause的相关信息,这些引起cause的方法形式-要么是AMQP.Channel.Close方法,要么是AMQP.Connection.Close (或者是null,如果是library中引发的异常,如网络通信故障,在这种情况下,可通过getCause()方法来获取信息).

public void shutdownCompleted(ShutdownSignalException cause) {   if (cause.isHardError())   {     
Connection conn = (Connection)cause.getReference();     
if (!cause.isInitiatedByApplication())     {       
Method reason = cause.getReason();       ...     }     ...   } 
else {     Channel ch = (Channel)cause.getReference();     ...   } }

原子使用isOpen()方法

channel和connection对象的isOpen()方法不建议在在生产代码中使用,因为此方法的返回值依赖于shutdown cause的存在性. 下面的代码演示了竟争条件的可能性:

public void brokenMethod(Channel channel) {     if (channel.isOpen())     {         // The following code depends on the channel being in open state.         // However there is a possibility of the change in the channel state         // between isOpen() and basicQos(1) call         ...         channel.basicQos(1);     } }

相反,我们应该忽略这种检查,并简单地尝试这种操作.如果代码执行期间,connection的channel关闭了,那么将抛出ShutdownSignalException,这就表明对象处于一种无效状态了.当broker意外关闭连接时,我们也应该捕获由SocketException引发的IOException,或者当broker清理关闭时,捕获ShutdownSignalException.

public void validMethod(Channel channel) {     try {         ...         channel.basicQos(1);     } catch (ShutdownSignalException sse) {         // possibly check if channel was closed         // by the time we started action and reasons for         // closing it         ...     } catch (IOException ioe) {         // check why connection was closed         ...     } }

Consumer线程池

Consumer 线程默认是通过一个新的ExecutorService线程池来自动分配的(参考下面的Receiving).如果需要在newConnection() 方法中更好地控制ExecutorService,可以使用定制的线程池.下面的示例展示了一个比正常分配稍大的线程池:

ExecutorService es = Executors.newFixedThreadPool(20); Connection conn = factory.newConnection(es); 

当连接关闭时,默认的ExecutorService将会被shutdown(), 但用户自定义的ExecutorService (如上面所示)将不会被shutdown(). 提供自定义ExecutorService的Clients必须确保最终它能被关闭(通过调用它的shutdown() 方法), 否则池中的线程可能会阻止JVM终止.

同一个executor service,可在多个连接之间共享,或者连续地在重新连接上重用,但在shutdown()后,则不能再使用.

使用这种特性时,唯一需要考虑的是:在消费者回调的处理过程中,是否有证据证明有严重的瓶颈. 如果没有消费者执行回调,或很少,默认的配置是绰绰有余. 开销最初是很小的,分配的全部线程资源也是有界限的,即使偶尔可能出现一阵消费活动.

使用Host列表

可以传递一个Address数组给newConnection(). Address只是 com.rabbitmq.client 包中包含host和port组件的简单便利类. 例如:

Address[] addrArr = new Address[]{ new Address(hostname1, portnumber1)                                  , new Address(hostname2, portnumber2)}; Connection conn = factory.newConnection(addrArr); 

如果提供了ExecutorService(在factory.newConnection(es, addrArr)中使用),那么线程池将与第一个成功连接相关联.

心跳超时

参考Heartbeats guide 来了解更多关于心跳及其在Java client中如何配置的更多信息.

自定义线程工厂

像Google App Engine (GAE)这样的环境会限制线程直接实例化. 在这样的环境中使用RabbitMQ Java client, 需要配置一个定制的ThreadFactory,即使用合适的方法来实例化线程,如: GAE's ThreadManager. 下面是Google App Engine的相关代码.

import com.google.appengine.api.ThreadManager;  ConnectionFactory cf = new ConnectionFactory(); cf.setThreadFactory(ThreadManager.backgroundThreadFactory()); 

Connection恢复

clients和RabbitMQ节点之间的连接可发生故障. RabbitMQ Java client 支持连接和拓扑(queues, exchanges, bindings, 和consumers)的自动恢复. 大多数应用程序的连接自动恢复过程会遵循下面的步骤:

1. 重新连接
2. 恢复连接监听器
3. 重新打开通道
4. 恢复通道监听器
5. 恢复通道basic.qos 设置,发布者确认和事务设置

1. 重新声明交换器(except for predefined ones)
2. 重新声明队列
3. 恢复所有绑定
4. 恢复所有消费者

ConnectionFactory factory = new ConnectionFactory(); factory.setUsername(userName); factory.setPassword(password); factory.setVirtualHost(virtualHost); factory.setHost(hostName); factory.setPort(portNumber); factory.setAutomaticRecoveryEnabled(true); // connection that will recover automatically Connection conn = factory.newConnection();

ConnectionFactory factory = new ConnectionFactory(); // attempt recovery every 10 seconds factory.setNetworkRecoveryInterval(10000);

ConnectionFactory factory = new ConnectionFactory();  Address[] addresses = {new Address("192.168.1.4"), new Address("192.168.1.5")}; factory.newConnection(addresses);

恢复监听器

可在可恢复连接和通道上注册一个或多个恢复监听器. 当启用了连接恢复时,ConnectionFactory#newConnection 和 Connection#createChannel 的连接已实现了com.rabbitmq.client.Recoverable,并提供了两个方法:

• addRecoveryListener
• removeRecoveryListener

发布影响

当连接失败时,使用Channel.basicPublish方法发送的消息将会丢失. client不会保证在连接恢复后,消息会得到分发.要确保发布的消息到达了RabbitMQ,应用程序必须使用Publisher Confirms 

拓扑恢复

拓扑恢复涉及交换器,队列,绑定以及消费者恢复.默认是启用的,但也可以禁用:

ConnectionFactory factory = new ConnectionFactory();  Connection conn = factory.newConnection(); factory.setAutomaticRecoveryEnabled(true); factory.setTopologyRecoveryEnabled(false);

手动应答和自动恢复

当使用手动应答时,在消息分发与应答之间可能存在网络连接中断. 在连接恢复后,RabbitMQ会在所有通道上重设delivery标记. 也就是说,使用旧delivery标记的basic.ack, basic.nack, 以及basic.reject将会引发channel exception. 为了避免发生这种情况, RabbitMQ Java client可以跟踪,更新,以使它们在恢复期间单调地增长. Channel.basicAck, Channel.basicNack, 以及Channel.basicReject 然后可以转换这些 delivery tags,并且不再发送过期的delivery tags. 使用手动应答和自动恢复的应用程序必须负责处理重新分发.