本文章翻译自http://www.rabbitmq.com/api-guide.html，并没有及时更新。

术语对照

指南：Guide

教程：Tutorial

代理（即RabbitMQ服务端）：Broker、RabbitMQ server

客户端：Client

发布者：Publisher

消费者：Consumer

连接：Connection

连接工厂：Connection Factory

通道：Channel

交换机：Exchange

队列：Queue

声明：Declare

路由：Route、routable、unroutable

持久化：durable、non-durable

自动删除：autodelete、non-autodelete

排斥：exclusive

路由关键字：routing key

推：Push

拉：Pull

（消息）投递：Delivery

关闭：Shutdown

度量：Metrics

本指南涵盖RabbitMQ Java客户端API，但不是教程，教程在另外的文档。

Java客户端API使用三个License：

● Apache Public License 2.0

● Mozilla Public License

● GPL 2.0

如果需要更详细的信息，请参考相关的Javadoc文档。

与Java客户端一起的还有命令行工具。

客户端API严密地按照AMQP 0-9-1协议规范来进行建模，为简化使用进行了额外的抽象。

1 概述

RabbitMQ Java客户端使用“com.rabbitmq.client”作为顶级包。主要的类和接口：

● Channel

● Connection

● ConnectionFactory

● Consumer

通过“Channel”接口可以进行协议操作。

“Connection”用来打开通道，注册连接生命周期的事件处理器，和关闭不需要的连接。

“Connection”通过“ConnectionFactory”来实例化。你可以配置“ConnectionFactory”使用不同的连接参数，比如vhost或username。

2 连接和通道

核心的API类是“Connection”和“Channel”，“Connection”代表AMQP 0-9-1连接，“Channel”代表通道。使用之前需要导入：

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.Channel;

2.1 连接到代理

下面的代码使用给定参数（主机名、端口号、等等）连接到一个AMQP代理：

ConnectionFactory factory = new ConnectionFactory();

factory.setUsername(userName);

factory.setPassword(password);

factory.setVirtualHost(virtualHost);

factory.setHost(hostName);

factory.setPort(portNumber);

Connection conn = factory.newConnection();

如果RabbitMQ服务端运行在本地，这些参数都不需要配置，它们有合适的默认值。

还可以使用URIs 来进行配置：

ConnectionFactory factory = new ConnectionFactory();

factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost");

Connection conn = factory.newConnection();

然后，“Connection”接口可用用来打开一个通道：

Channel channel = conn.createChannel();

现在，这个通道就可以用来发送和接收消息了，下面章节将进行阐述。

要想与代理断连，简单的关闭通道和连接即可：

channel.close();

conn.close();

注意，关闭通道是一个很好的习惯，但并不是严格必须的。当底层的连接关闭的时候，通道都将会自动被关闭。

3 使用交换机和队列

客户端应用与交换机和队列一起工作，交换机和队列是AMQP的高级模块。在使用它们之前必须先“声明”。声明它们就能确保以该名字命名的交换机或队列是存在的，如果不存在，就创建。

继续上面的例子，下面的代码声明了一个交换机和一个队列，然后将它们绑定在一起。

channel.exchangeDeclare(exchangeName, "direct", true);

String queueName = channel.queueDeclare().getQueue();

channel.queueBind(queueName, exchangeName, routingKey);

这样就声明了交换机和队列，两者都可以用额外的参数来定制。下面是它们的参数：

● 交换机：持久化的（durable）、非自动删除的（non-autodelete）、类型是“direct”

● 队列：非持久化的（non-durable）、排斥的（exclusive）、自动删除的（autodelete）、名字是随机生成的（generated name）

然后，通过给定的路由关键字（routing key）将队列绑定到交换机。

注意：当只有一个客户端需要与该队列一起工作时，这是比较常见的声明方式。该队列不需要一个指定的名字，也不能被其他客户端所使用（即它是排斥的），而且将在该客户端消失时自动被清理（即它是自动删除的）。如果多个客户端要共用一个命名的队列，下面的代码会比较合适：

channel.exchangeDeclare(exchangeName, "direct", true);

channel.queueDeclare(queueName, true, false, false, null);

channel.queueBind(queueName, exchangeName, routingKey);

这样就声明了：

● 交换机：持久化的（durable）、非自动删除的（non-autodelete）、类型是“direct”

● 队列：持久化的（durable）、非排斥的（non-exclusive）、非自动删除的（non-autodelete）、名字是指定的（well-known name）

注意，“Channel”的所有这些API都是重载的。参数较少的API比较方便，使用一些合适的默认值；参数较多的API在必要时让你能够覆盖那些默认值，以便有完全的控制。

这种“short form, long form”模式（即方法重载模式）在客户端API中经常使用。

4 发布消息

要发布一个消息到交换机，则使用“Channel.basicPublish”：

byte[] messageBodyBytes = "Hello, world!".getBytes();

channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);

为了更好的控制，你可以使用重载版本，指定“mandatory”标识符，或者使用预置的消息属性来发送消息：

channel.basicPublish(exchangeName, routingKey, mandatory,

    MessageProperties.PERSISTENT_TEXT_PLAIN,

    messageBodyBytes);

上面是以投递模式2（delivery mode 2）、优先级1、内容类型为“text/plain”来发送消息。你可以使用“Builder”类来构建你自己的消息属性对象，引用你喜欢的属性，比如：

channel.basicPublish(exchangeName, routingKey,

    new AMQP.BasicProperties.Builder()

        .contentType("text/plain")

        .deliveryMode(2)

        .priority(1)

        .userId("bob")

        .build()),

    messageBodyBytes);

下面的例子使用定制的headers来发布消息：

Map headers = new HashMap();

headers.put("latitude",51.5252949);

headers.put("longitude", -0.0905493);

channel.basicPublish(exchangeName, routingKey,

    new AMQP.BasicProperties.Builder()

        .headers(headers)

        .build()),

    messageBodyBytes);

下面的例子使用过期时间来发布消息：

channel.basicPublish(exchangeName, routingKey,

    new AMQP.BasicProperties.Builder()

        .expiration("60000")

        .build()),

    messageBodyBytes);

我们不在这里阐述所有的发布消息的属性配置。

注意，”BasicProperties”类是”AMQP”类的一个内部类，“AMQP”类是一个自动生成的持有者类（autogenerated holder class即专门用来定义静态内部类和静态属性）

如果资源不足告警 有效的话，那么调用”Channel#basicPublish”时可能最终会导致阻塞。

5 通道和并发考虑（线程安全）

一般来说，应该避免在线程间共享使用“Channel”实例。应用应该选择每个线程使用一个“Channel”，而不是在多个线程之间共享同一个“Channel”。

在通道上的一些操作是并发调用安全的，一些不是安全的，会导致线上的帧错误的交织在一起（即数据帧错乱）、重复的消息确认等等。

在一个共享通道上并发的发布消息可能会导致线上的帧错误的交织在一起（即数据帧错乱），同时会触发连接级别的协议异常和连接关闭。因此，在应用代码中必须显式的进行同步（”Channel#basicPublish”必须放在一个临界区进行调用）。线程间共享通道还会对发布者确认（PublisherConfirms） 造成干扰。我们强烈推荐应该避免在一个共享通道上并发的发布消息。

一个线程使用共享通道消费消息，另一个线程使用该通道发布消息，这样是安全的。

服务端推送（Server-pushed）的消息投递（见下节）在并发的执行时，会保证每个通道内的消息顺序是一致的。这种分发机制采用了每个连接分配一个“java.util.concurrent.ExecutorService”。通过使用“ConnectionFactory#setSharedExecutor”方法提供一个定制的在同一个“ConnectionFactory”建立的所有连接之间共享的executor是可能的。

当使用手动确认时，考虑由哪个线程来发送ack很重要。如果发送ack的线程不同于收到该次消息投递的的线程（比如，“Consumer#handleDelivery”方法把对消息投递的处理委托给一个不同的线程），那么，将“multiple”参数设置为true来发送ack是不安全的，这将会导致重复确认和通道级别的协议异常，该异常将关闭通道。每次对一个消息进行发送ack是安全的。

6 通过订阅接收消息（Push API）

import com.rabbitmq.client.Consumer;

import com.rabbitmq.client.DefaultConsumer;

接收消息的最有效的方法是使用“Consumer”接口发起订阅。当消息到达队列时，将自动被投递，而无需显式的请求消息。

当调用“Consumer”相关的API方法时，各自的订阅总是由它们的消费者标签（consumer tags）来引用。消费者tag就是消费者id，它由客户端或服务端生成。为了让RabbitMQ生成节点唯一的tag，使用不带消费者tag参数或者传空字符串给该参数的“Channel#basicConsume”方法，并使用该方法的返回值即可。消费者标签用于取消消费者的订阅。

不同的“Consumer”实例必须有不同的消费者标签。在一个连接上使用重复的消费者标签是绝对禁止的，它会引发连接自动恢复的问题，而且会混淆对消费者的监控。

实现一个“Consumer”的最简单的方法是继承“DefaultConsumer”。然后，将该子类的一个对象传入“basicConsume”方法即可发起订阅：

boolean autoAck = false;

channel.basicConsume(queueName, autoAck, "myConsumerTag",

    new DefaultConsumer(channel) {

        @Override

        public voidhandleDelivery(String consumerTag,

                                                    Envelope envelope,

                                                    AMQP.BasicProperties properties,

                                                    byte[] body)

            throws IOException

        {

            String routingKey =envelope.getRoutingKey();

            String contentType = properties.getContentType();

            long deliveryTag =envelope.getDeliveryTag();

            // (process the message components here ...)

            channel.basicAck(deliveryTag, false);

        }

    });

这里，我们设置了”autoAck = false”，所以必须对投递到该“Consumer”的消息进行确认，这个很方便就在“handleDelivery”方法（收到消息会被调用）内就完成了，如代码所示。

更加复杂的“Consumer”需要覆盖更多方法。特别是，“handleShutdownSignal”方法将在通道和连接关闭时被调用，”handleConsumeOk”方法在调用其他任何到该“Consumer”的回调之前进行调用，传入的参数是消费者标签。

消费者还可以实现“handleCancelOk”和“handleCancel”方法，这样就能够在显式和隐式取消消费者（订阅）时得到通知。

你可以显式的取消一个特定的“Consumer”，使用”Channel.basicCancel”方法，传入消费者标签：

channel.basicCancel(consumerTag);

跟发布者类似，考虑消费者的并发安全也很重要。

“Consumer”的回调是在一个线程池中进行分发的，该线程池独立于实例化该通道的线程。这意味着”ConsumerS”可以在该连接和通道上安全的调用阻塞方法，比如“Channel#queueDeclare”或者“Channel#basicCancel”。

每个通道都有自己的分发线程。最常见的使用场景是每个通道只有一个消费者，意味着该通道没有其他消费者。如果一个通道上有多个消费者，一个运行时间长的消费者可能会占用该通道其他消费者的回调的分发。

请参考“并发考虑（线程安全）”一节。

7 主动获取消息（Pull API）

要想显式的获取消息，使用“Channel.basicGet”方法。返回值是一个“GetResponse”的实例，可以从该实例中提取出header信息（属性）和消息体：

boolean autoAck = false;

GetResponse response = channel.basicGet(queueName, autoAck);

if (response == null) {

    // No message retrieved.

} else {

    AMQP.BasicProperties props =response.getProps();

    byte[] body =response.getBody();

    long deliveryTag =response.getEnvelope().getDeliveryTag();

    ...

由于上面设置了“autoAck = false”，你必须调用“Channel.basicAck”来却你已经成功收到消息：

...

    channel.basicAck(method.deliveryTag, false);// acknowledge receipt of the message

}

8 处理不可路由的消息

如果一个消息发布时设置了“mandatory”标识符，但又不能路由，代理将会把它返回给发送客户端（通过AMQP.Basic.Return命令）。

为了得到这种消息被返回的通知，客户端可以实现“ReturnListener”接口，然后调用“Channel.setReturnListener”方法。如果客户端没有为通道配置一个返回监听器，那么相关的返回消息将被安静的丢弃。

channel.setReturnListener(new ReturnListener() {

    public voidhandleBasicReturn(int replyCode,

                                                    String replyText,

                                                    String exchange,

                                                    String routingKey,

                                                    AMQP.BasicProperties properties,

                                                    byte[] body)

    throws IOException {

        ...

    }

});

举个例子，如果客户端将设置“mandatory”标识符的消息发布到一个没有绑定任何队列的“direct”类型的交换机，就会调用返回监听器。

9 关闭协议

9.1 AMQP客户端关闭的概述

AMQP 0-9-1的连接和通道都使用相同的方法来管理网络故障、内部故障、和显式的本地关闭。

AMQP 0-9-1的连接和通道对象在其生命周期中有如下状态：

● 打开（open）：该对象可以使用。

● 正在关闭（closing）：该对象得到本地显式的通知要关闭，已经向底层的对象发送了一个关闭请求，正在等待它们完成关闭过程。

● 关闭（closed）：该对象从底层对象得到关闭完成的通知，并已经将自己关闭。

AMQP连接和通道对象总是以关闭状态终止，不管关闭的原因是什么，比如应用的请求、内部客户端库的故障、远程网络请求、或网络故障。

AMQP连接和通道对象拥有下列关闭相关的方法：

● “addShutdownListener(ShutdownListener listener)”

和“removeShutdownListener(ShutdownListener listener)”：用于管理关闭监听器。这些关闭监听器将在对象状态转换为关闭状态时触发。注意，添加一个关闭监听器到已经关闭的对象，将立即触发该监听器。

● “getCloseReason()”：用以允许检查导致该对象关闭的原因。

● “isOpen()”：用以检测该对象是否处于打开状态。

● “close(int closeCode, String closeMessage)”：显式通知该对象关闭。

监听器简单的用法如下：

import com.rabbitmq.client.ShutdownSignalException;

import com.rabbitmq.client.ShutdownListener;

connection.addShutdownListener(new ShutdownListener() {

    public voidshutdownCompleted(ShutdownSignalException cause)

    {

        ...

    }

});

9.2 关闭的环境

我们可以捕获“ShutdownSignalException”，该异常包含了关于关闭原因的所有可获取的信息，或是显式调用“getCloseReason()”方法能获取的，或是使用“ShutdownListener”类的“service(ShutdownSignalException cause)”方法的“cause”参数获取的。

“ShutdownSignalException”类提供了用于分析关闭原因的方法。调用“isHardError()”方法可以判别是连接出错还是通道出错。“getReason()”方法以一个AMQP method的形式返回该原因的相关信息-或是”AMQP.Channel.Close”或是“AMQP.Connection.Close”（或是为null，如果该原因是库内某个异常，比如网络通信失败，这种情况下可以用“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();

        ...

    }

}

9.3 isOpen()方法的原子性和使用

不推荐在生产代码中使用通道和连接对象的“isOpen()”方法。因为该方法的返回值取决于关闭原因是否存在。下面的代码阐述了竞争条件的可能性：

public void brokenMethod(Channel channel)

{

    if (channel.isOpen())

    {

        // The following codedepends on the channel being in open state.

        // However there is apossibility of the change in the channel state

        // between isOpen() andbasicQos(1) call

        ...

        channel.basicQos(1);

    }

}

相反，我们通常应该忽略这种检查，简单的尝试所需的操作。如果执行操作期间该连接的该通道被关闭，将会抛出“ShutdownSignalException”异常表示该对象处于无效的状态。当代理意外关闭连接时，我们还应该捕获”SocketException”引发的“IOException”；当代理发起正常关闭时，我们应该捕获“ShutdownSignalException”。

public void validMethod(Channel channel)

{

    try {

        ...

        channel.basicQos(1);

    } catch (ShutdownSignalExceptionsse) {

        // possibly check if channelwas closed

        // by the time we startedaction and reasons for

        // closing it

        ...

    } catch (IOException ioe) {

        // check why connection wasclosed

        ...

    }

}

10 高级连接选项

10.1 消费者线程池

消费者线程（见“Push API”一节）默认是从一个新的“ExecutorService”线程池中自动分配的。如果需要更多控制权，可以在调用“newConnection()”方法时传入一个“ExecutorService”，从而使用你传入的线程池。下面的例子传入一个比默认情况更大的线程池：

ExecutorService es = Executors.newFixedThreadPool(20);

Connection conn = factory.newConnection(es);

“Executors”和“ExecutorService”类都在“java.util.concurrent”包中。

当该连接关闭时，默认的“ExecutorService”将被“shutdown()”，但是用户提供的“ExecutorService”（像上面的es）将不会自动“shutdown()”。提供定制“ExecutorService”的客户端必须确保它最终是关闭的（通过调用它的“shutdown()方法”），否则线程池中的线程会阻止JVM的终止。

同一个executor service可能被多个连接所共享，或者当重新连接时连续复用，但当它“shutdown()”后就不能被使用了。

如果有证据表明在处理“Consumer”的回调时出现严重的瓶颈，才应该考虑使用此功能。如果没有“Consumer”回调可执行，或很少回调，默认的线程池分配就已经足够了。开销最初是最小的，而分配的总线程资源是有限的，即使消费者活动激增会偶尔发生。

10.2使用多个主机

我们也可以传入“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);

上面的代码尝试连接到“hostname1:portnumber1”，如果失败则尝试连接“hostname2:portnumber2”。返回的连接是数组中第一个成功连接上的（没有抛出“IOException”）。这跟每次重复设置连接工厂的主机名和端口并调用“factory.newConnection()”直到有一次连接成功，是完全等效的。

如果同时提供“ExecutoryService”（即调用“factory.newConnection(es, addrArr)”），线程池将和第一个成功的连接进行关联。

如果你想要对所连接的主机进行更多的控制，参考下一节关于服务发现的内容。

10.3 使用AddressResolver接口的服务发现

从版本3.6.6开始，就可以实现“AddressResolver”接口在创建连接时选择连接到哪个地址：

Connection conn = factory.newConnection(addressResolver);

“AddressResolver”接口如下：

public interface AddressResolver {

    List getAddresses()throws IOException;

}

就想上节使用多个主机一样，第一个返回的“Address”将首先尝试，如果失败，则尝试第二个，以此往下。

如果同时提供“ExecutoryService”（即调用“factory.newConnection(es, addressResolver)”），线程池将和第一个成功的连接进行关联。

“AddressResolver”接口可以用来很好的实现定制的服务发现的逻辑，这在一个动态的基础设施环境中会很有用。加上自动恢复功能（见11章），客户端可以自动连接上那些在客户端第一次启动时还未启动的节点。定制“AddressResolver”在亲和性和负载均衡场景下也很有用。

Java客户端API自带了如下实现（见Javadoc）：

● DnsRecordIpAddressResolver：给定主机名，返回其IP地址（针对DNS服务器平台的方案）。这对于简单的基于DNS的负载均衡和故障转移很有用。

● DnsSrvRecordAddressResolver：给定服务的名称，返回主机名和端口。这个搜索通过DNS SRV请求来实现。当使用诸如“HashiCorp Consul”的服务注册表时，这就很有用。

10.4 心跳超时

参见“心跳指南”来获取更多关于心跳和如何在Java客户端配置心跳的信息。

10.5 定制线程工厂

诸如Google App Engine（GAE）的环境可能会限制直接实例化线程。在这样的环境中使用RabbitMQ的Java客户端，必须配置一个定制的“ThreadFactory”，它使用合适的方法来实例化线程，比如GAE的“ThreadManager”。下面是一个针对GAE的例子：

import com.google.appengine.api.ThreadManager;

ConnectionFactory cf = new ConnectionFactory();

cf.setThreadFactory(ThreadManager.backgroundThreadFactory());

10.6 支持Java非阻塞IO

版本4.0的Java客户端API提供了对Java非阻塞IO（即Java NIO）的试验性支持。NIO并不一定比阻塞IO快，它只是允许你更容易的控制资源（本例中就是线程）。

在默认的阻塞IO模式下，每一个连接都使用一个线程来从网络套接口上读取数据。在NIO模式下，你可以控制读写网络套接口的线程数量。

如果你的Java进程使用很多连接（几十或上百个），就要使用NIO模式。这样应该会比默认的阻塞模式使用更少的线程。如果设置了适当的线程数，就不应该尝试降低性能，尤其是在连接不那么繁忙的情况下。

NIO必须显式开启：

ConnectionFactory connectionFactory = new ConnectionFactory();

connectionFactory.useNio();

NIO模式可以通过“NioParams”类进行配置：

connectionFactory.setNioParams(new NioParams().setNbIoThreads(4));

NIO模式使用了合理的参数默认值，但你也许需要根据你的工作负载来修改。其中一些设置是：使用的IO线程总数、缓冲区的大小、用以IO循环的service executor、内存中写队列的参数（写请求在发送到网络之前要先入队）。请阅读Javadoc获取更多细节和默认值。

11 从网络故障中自动恢复

11.1 连接恢复

客户端和RabbitMQ节点之间的网络连接可能会发生故障。RabbitMQ的Java客户端API支持连接和topology（队列、交换机、绑定和消费者）的自动恢复。很多应用的自动恢复过程都是如下步骤：

1.重连

2.恢复连接监听器

3.重新打开通道

4.恢复通道监听器

5.恢复通道的”basic.qos”设置，发布者确认和事务设置

topology恢复包含如下动作，针对每个通道而执行：

1.重新声明交换机（除预定义交换机以外）

2.重新声明队列

3.恢复所有绑定

4.恢复所有消费者

4.0.0版本的Java客户端API，自动恢复是默认开启的（topology恢复也是）。

要关闭或开启自动连接恢复，要使用“factory.setAutomaticRecoveryEnabled(boolean)”。下面的代码段展示了如何显示的开启启动恢复（版本4.0.0之后）：

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();

如果因为异常（比如RabbitMQ节点不可到达）导致恢复失败，将会以一个固定的时间间隔（默认5秒）继续尝试。该时间间隔可配置：

ConnectionFactory factory = new ConnectionFactory();

// attempt recovery every 10 seconds

factory.setNetworkRecoveryInterval(10000);

当使用多个地址时，将会随机排列地址列表并一个一个尝试所有的地址：

ConnectionFactory factory = new ConnectionFactory();

Address[] addresses = {new Address("192.168.1.4"), newAddress("192.168.1.5")};

factory.newConnection(addresses);

11.2 恢复监听器

可以注册一个或多个恢复监听器，来监听可恢复的连接和通道。当开启了连接恢复，由“ConnectionFactory#newConnection”和“Connection#createChannel”返回的连接实现了“com.rabbitmq.client.Recoverable”接口，提供了相当明显的两个方法：

● addRecoveryListener

● removeRecoveryListener

注意，目前需要将连接和通道强制转换成“Recoverable”类型，才能使用这两个方法。

11.3 对发布消息的影响

当连接断开时，使用“Channel.basicPublish”发布的消息将会丢失。客户端不会将它们入队缓存直到连接恢复后重新发布。要确保发布的消息到达了RabbitMQ，应用需要使用“发布者确认”，并对连接失效负责。

11.4 Topology恢复

topology恢复包括交换机、队列、绑定和消费者的恢复。当自动恢复开启时，默认也就开启了topology恢复。因此，Java客户端API的4.0.0版本默认是开启topology恢复的。

如有需要，可以显示关闭topology恢复：

ConnectionFactory factory = new ConnectionFactory();

Connection conn = factory.newConnection();

// enable automatic recovery (e.g. Java client prior 4.0.0)

factory.setAutomaticRecoveryEnabled(true);

// disable topology recovery

factory.setTopologyRecoveryEnabled(false);

11.5 手动确认和自动恢复

当使用手动确认时，到RabbitMQ节点的网络连接可能在消息投递和确认中间发生故障。在连接恢复之后，RabbitMQ将会重置所有通道的投递标签。这意味着带有旧投递标签的“basic.ack”、“basic.nack”和“basic.reject”将会引发通道异常。为了避免这种情况，RabbitMQ的Java客户端API跟踪并更新投递标签，并使投递标签在每次连接恢复时单调增长。然后，“Channel.basicAck”、“Channel.basicNack”和“Channel.basicReject”将调整过后的投递标签转换成RabbitMQ使用的标签。这样，带有旧投递标签的确认将不会被发送出去。使用手动确认和自动恢复的应用必须能够处理重新投递的情况。

12 未处理的异常

有关连接、通道、恢复、和消费者生命周期的未处理异常，都被委托给异常处理器。异常处理器可以是实现了“ExceptionHandler”接口的任何对象。默认情况下，使用的是“DefaultExceptionHandler”的一个实例。它将异常的详细信息打印到标准输出。

使用“ConnectionFactory#setExceptionHandler”可以覆盖默认的处理器，它将用于由该工厂创建的所有连接：

ConnectionFactory factory = new ConnectionFactory();

cf.setExceptionHandler(customHandler);

异常处理器应该用于异常的记录。

13 度量和监控

从版本4.0.0开始，客户端收集运行时的度量（比如已经发布的消息数量）。度量收集是可选的并在“ConnectionFactory”级别使用“setMetricsCollector”方法进行设置。该方法需要一个“MetricsCollector”实例，该实例将在客户端代码的多个地方被调用。

客户端自带有一个使用“Dropwizard Metrics”库的“MetricsCollector”实现。可以按如下方式来开启度量收集：

ConnectionFactory connectionFactory = new ConnectionFactory();

StandardMetricsCollector metrics = new StandardMetricsCollector();

connectionFactory.setMetricsCollector(metrics);

...

metrics.getPublishedMessages(); // get Metrics' Meter object

下面是收集的各种度量：

● 打开的连接数量（默认实现的一个“Counter”）

● 打开的通道数量（默认实现的一个“Counter”）

● 已经发布的消息数量（默认实现的一个“Meter”）

● 已经消费的消息数量（默认实现的一个“Meter”）

● 已经确认的消息数量（默认实现的一个“Meter”）

● 已经拒绝的消息数量（默认实现的一个“Meter”）

通过使用Dropwizard Metrics，不仅可以获得计数，还可以获得过去五分钟的速率等，还有各种开箱即用的报告工具（JMX、Graphite、Ganglia、HTTP）。

请注意下面关于度量收集的信息：

● 如果你使用基于Dropwizard Metrics的默认实现，不要忘记在你的classpath中添加合适的JAR文件。（Java客户端API没有自带Dropwizard Metrics，这是一个可选依赖）。

● 度量收集是可扩展的，你可以实现自己的”MetricsCollector”满足特殊需求。

● “MetricsCollector”是在“ConnectionFactory”级别设置的，但可在多个实例间共享。

● 度量收集不支持事务。比如，如果在一个事务中发送了一个ack，而该事务后来又回滚了，该ack在客户端度量中被计数了（但显然在代理上没有计数）。注意，该ack实际上被发送到代理，而然后又被事务回滚所取消，所以客户端的发送ack数量这个度量是正确的。总之，不要为敏感业务监控使用客户度量，因为它们不保证完全正确。

13.1 度量报告

如果你使用基于Dropwizard Metrics的“StandardMetricsCollector”，你可以发送这些度量值到多个报告后端：控制台、JMX、HTTP、Graphite、Ganglia等等。

你通常要传入一个“MetricsRegistry”的实例到“StandardMetricsCollector”。下面是JMX的一个例子：

MetricRegistry registry = new MetricRegistry();

StandardMetricsCollector metrics = new StandardMetricsCollector(registry);

ConnectionFactory connectionFactory = new ConnectionFactory();

connectionFactory.setMetricsCollector(metrics);

JmxReporter reporter = JmxReporter

    .forRegistry(registry)

    .inDomain("com.rabbitmq.client.jmx")

    .build();

reporter.start();

14 Google App Engine上的RabbitMQJava Client

在Google App Engine（GAE）上使用RabbitMQ Java客户端需要使用一个定制的线程工厂，该工厂使用GAE的“ThreadManager”（见之前所述）来实例化线程。此外，必须设置比较短的心跳间隔（4-5秒）来避免遇到GAE上的“InputStream”读超时。

ConnectionFactory factory = new ConnectionFactory();

cf.setRequestedHeartbeat(5);

15 警告和限制

为了topology恢复成为可能，RabbitMQ Java客户端维护了一个已声明的队列、交换机、和绑定的缓存。每个连接都有自己的缓存。一些RabbitMQ功能让客户端有可能观察topology的变化，比如TTL导致队列被删。RabbitMQ Java客户端在下列情况下尝试让缓存项无效：

● 队列被删

● 交换机被删

● 绑定被删

● 在自动删除的队列上消费者被取消

● 当队列或交换机从一个自动删除的交换机上解除绑定。

但是，客户端不能再单个连接之外跟踪这些topology的变化。那些依赖于自动删除的队列或交换机，还有队列TTL（注意：不是消息TTL），和使用自动连接恢复的应用，应该显式的删除那些未使用或已被删除的实体（即队列、交换机、绑定等），来清理客户端的topology缓存。这可以使用“Channel#queueDelete”、“Channel#exchangeDelete”、“Channel#queueUnbind”、和“Channel#exchangeUnbind”来实现，它们在RabbitMQ

3.3.x中是幂等的操作（即多次操作的结果是一样的，推导出：删除不存在的不会导致异常）。

16 RPC（请求/响应）模式

为方便编程，Java客户端API提供了一个“RpcClient”类，它使用一个临时的应答队列来提供简单的符合AMQP 0-9-1的RPC风格的通信设施。

这个类没有对RPC参数和返回值强制为特殊格式。它简单的提供了一种机制：发送带有特定路由关键字的消息到一个给定交换机，然后在一个应答队列上等待响应。

import com.rabbitmq.client.RpcClient;

RpcClient rpc = new RpcClient(channel, exchangeName, routingKey);

（关于这个类如何使用AMQP 0-9-1的实现细节如下：请求消息将“basic.correlation_id”字段设置为在该“RpcClient”实例中是唯一的值，将“basic.reply_to”字段设置为应答队列的名称。）

一旦你创建了该类的实例，你可以以如下任意一种方法来发送RPC请求：

● byte[] primitiveCall(byte[] message);

● String stringCall(String message)

● Map mapCall(Map message)

● Map mapCall(Object[] keyValuePairs)

“primitiveCall”方法传输原始的字节数组来作为请求体和响应体。“stringCall”是对“primitiveCall”的方便的包裹（wrapper），把消息体作为默认字符编码的“String”的实例。

“mapCall”有一点复杂：它把包含普通Java值的“java.util.Map”编码成AMQP 0-9-1的二进制表格的形式，对响应也是以同样的方式解码。（注意，这里所使用的值的类型存在一些限制，参见javadoc）

所有编码/解码的比较方便的方式都是使用“primitiveCall”作为传输机制的，仅仅是在此之上提供了一个封装层而已。

17 TLS支持

在客户端和代理之间的通信可以使用TLS来进行加密。也支持客户端和服务器之间的身份验证（即节点认证，peer verification）。下面是Java客户端使用加密的最简单的方式：

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("localhost");

factory.setPort(5671);

factory.useSslProtocol();

注意，在上面的例子中，客户端默认是不强制服务端认证的（同行证书链验证：peer certificate chain verification），例子中使用的是“信任所有证书”的“TrustManager”。这在本地开发的时候很方便，但很容易受到中间人攻击，因此不推荐在生产中使用。要学习更多关于RabbitMQ的TLS支持的信息，参见“TLS指南”。如果你仅仅是想配置Java客户端（尤其是节点认证和trust manager部分），请阅读TLS指南的相关章节。