《RabbitMQ开发库的完整API文档》翻译
背景译文链接我的译文概述Connections and Channels连接到一个代理使用 Exchanges and Queues队列发布消息Publishing messages通道和并发性考虑事项线程安全通过订阅接收消息Push API恢复个人消息Pull API处理 unroutable无法发送的 消息关闭协议高级连接选项从网络故障中自动恢复未处理异常度...
背景
我的应用需要实时的接收RabbitMQ推送过来的消息，故找了对应的文档查看用法。这里特地翻译出来方便自己和诸位博友日后学习分享。
译文链接
我的译文
Java客户端API指南
该指南涵盖了RabbitMQ Java客户端API。然而，这并不是一个教程。这些都可以在不同的章节中找到。
5.x版本系列的库需要JDK 8，用于编译和运行时。在Android上，这意味着只支持Android 7.0或更高的版本。4.x发布系列支持JDK 6和Android7.0之前的版本。
该库是开源的，并且有以下三种授权：
这意味着用户可以考虑在上面的三种授权列表中的任何许可下使用该库。例如，用户可以选择Apache Public License 2.0，并将该客户端包含到一个商业产品中。在GPLv2下获得许可的代码库可以选择GPLv2，等等。
API引用(JavaDoc)是单独可用的。
还有一些命令行工具，这些工具曾经与Java客户端一起使用。
客户端API与AMQP 0-9-1协议规范进行了密切的建模，并提供了额外的抽象以方便使用。
概述
RabbitMQ Java客户端使用 com.RabbitMQ.client 作为它的top-level package。关键类和接口是:
Channel
Connection
ConnectionFactory
Consumer
协议操作通过 Channel 接口可用。Connection 用于打开channels，注册连接生命周期事件处理程序，以及关闭不再需要的连接。Connections 通过ConnectionFactory实例化，这是您配置各种连接设置的方式，例如vhost或用户名。
Connections and Channels
核心API类是Connection 和Channel，分别表示AMQP 0-9-1连接和通道。这些是使用前需要导入的：
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
连接到一个代理
下面的代码使用给定的参数(主机名、端口号等)连接到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服务器，所有这些参数都有合理的默认值
或者，也可以使用uri:
ConnectionFactory factory = new ConnectionFactory();
factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost");
Connection conn = factory.newConnection();
所有这些参数对于运行在本地的RabbitMQ服务器都有合理的默认值。
然后可以使用 Connection 接口打开通道：
Channel channel = conn.createChannel();
该通道现在可以用于发送和接收消息，后续部分会有讲解。
要断开连接，只需关闭通道和连接：
channel.close();
conn.close();
请注意，关闭通道可能被认为是良好的习惯，但在这里并不是绝对必要的，因为当底层连接关闭后，任何情况下通道都会自动的关闭
使用 Exchanges and Queues（队列）
客户端应用程序与交换机和队列一起工作，即AMQP的高级构建块。这些必须在使用之前被“声明”。声明任何类型的对象只会确保其中一个名称存在，并在必要时创建它。
继续之前的例子,下面的代码声明了一个交换机和一个队列，然后将它们绑定在一起。
channel.exchangeDeclare(exchangeName, "direct", true);
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, exchangeName, routingKey);
这将主动声明以下的对象，这两个对象都可以通过使用额外的参数进行定制。这里，对象有一些特殊的参数体现：
一个持久的、非自动删除的“直接”类型的exchange（交换机）
有一个已知名称的、非持久的、专有的、自动删除的队列
上面的函数调用然后用给定的路由键将队列绑定到交换机。
注意，当只有一个客户端希望使用它时，这将是声明队列的一种典型方式：它不需要一个已知的名称，其他客户端不能独占队列，并且队列会自动清理(自动删除)。如果有几个客户端想要共享一个已知名称的队列，那么这段代码将是所需要的:
channel.exchangeDeclare(exchangeName, "direct", true);
channel.queueDeclare(queueName, true, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
这些会主动声明：
一个持久的、非自动删除的“直接”类型的exchange（交换机）
有一个已知名称的、持久的、非专有的（这里理解为可共享的）、非自动删除的队列
请注意，所有这些 Channel API方法都是重载的。对于exchangeDeclare, queueDeclare 和 queueBind 这些方便的简写形式都是使用了合理的默认。还有更多参数的更长的形式，可以让您根据需要覆盖这些默认值，以便在需要的时候提供绝对控制权。
这种“简写形式、长形式（这里理解为带更多参数的一种形式）”模式在客户端的API中使用。
发布消息（Publishing messages）
要将消息发布到交换机中，请使用 Channel.basicPublish 如下：
byte[] messageBodyBytes = "Hello, world!".getBytes();
channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);
为了获得良好的控制，您可以使用重载的变式来指定 mandatory 标志，或者使用预先设置的消息属性来发送消息：
channel.basicPublish(exchangeName, routingKey, mandatory,
MessageProperties.PERSISTENT_TEXT_PLAIN,
messageBodyBytes);
发送一条消息，mode是2（即持久化的），priority 为1，content-type 为 “text/plain”。您可以构建自己的消息属性对象，使用一个Builder类，您可以使用您喜欢的属性，例如:
channel.basicPublish(exchangeName, routingKey,
new AMQP.BasicProperties.Builder().contentType("text/plain")
.deliveryMode(2)
.priority(1).userId("bob")
.build()),
messageBodyBytes);
举个例子， 发一个带自定义头部的消息:
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("latitude", 51.5252949);
headers.put("longitude", -0.0905493);
channel.basicPublish(exchangeName, routingKey,
new AMQP.BasicProperties.Builder()
.headers(headers)
.build()),
messageBodyBytes);
这个例子发布了一个expiration的消息:
channel.basicPublish(exchangeName, routingKey,
new AMQP.BasicProperties.Builder()
.expiration("60000")
.build()),
messageBodyBytes);
我们还没有说明所有的可能性。
注意，BasicProperties是自动生成的holder类AMQP的内部类。
如果资源驱动的警报生效，那么 Channel#basicPublish 的调用最终会被阻塞。
通道和并发性考虑事项(线程安全)
根据已有经验，在线程之间共享 Channel 实例是可以避免的。应用程序应该选择一个 Channel 对应一个线程，而不是在多个线程之间共享同一个 Channel 。
虽然通道上的一些操作是可以同时调用的，但有些操作是不安全的，并且会导致不正确的帧间交错问题，双确认问题等等。
共享通道上的并发发布可能会导致连接上不正确的帧，从而触发连接级别的协议异常和连接关闭。因此，它需要在应用程序代码中显式同步( Channel#basicPublish 必须在关键部分中调用)。在线程之间共享通道也会干扰发布者的确认。我们强烈建议在共享通道上避免并发发布。
在共享通道时，在一个线程中消费，在另一个线程中发布可以是安全的。
Server-pushed deliveries (见下面的部分)是同时发出的，保证每个通道的顺序被保留。
对每个连接，分派机制使用java.util.concurrent.ExecutorService。可以提供一个自定义执行器，它将由一个ConnectionFactory的ConnectionFactory#setSharedExecutor 调用所产生的所有连接共享。
当使用手动确认时，重要的是要考虑哪些线程做了确认。如果它与接收到的线程不同（例如，Consumer#handleDelivery委托给不同的线程）将 multiple 参数设置为 true 是不安全的，并将导致双重确认，因此会出现通道级协议异常，从而关闭通道。因此一次只确认一条信息可以是安全的。
通过订阅接收消息（”Push API”）
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
接收消息的最有效方式是使用 Consumer 接口设置订阅。消息将在到达时自动发送，而不用显式地请求。在调用与Consumers有关的API方法时,个人订阅总是与它们的消费者标签相关联。消费者标签是消费者的唯一标识符，它可以是客户端，也可以是服务器生成的。为了让RabbitMQ生成一个节点范围的惟一标记，使用 Channel#basicConsume 方法重写，它不需要携带一个消费者标记参数，或者传递一个空字符串给消费者标记，并使用 Channel#basicConsume 方法返回的值。消费者标签可被用来取消消费者。
不同的消费者实例必须具有不同的消费者标记。对连接上的重复的消费者标签是强烈反对的，当消费者被监控时，会导致自动连接恢复和令人困惑的监控数据的问题。
实现Consumer 的最简单方法是实例化 DefaultConsumer 类。这个子类的一个对象可以通过basicConsume的调用来设置订阅:
boolean autoAck = false;
channel.basicConsume(queueName, autoAck, "myConsumerTag",
new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException
{
String routingKey = envelope.getRoutingKey();
String contentType = properties.getContentType();
long deliveryTag = envelope.getDeliveryTag();
// (这里进行消息组件的处理 ...)
channel.basicAck(deliveryTag, false);
}
});
在这里，因为我们指定了autoAck=false，有必要确认消息传递给消费者，最方便的是在 handleDelivery 方法中完成，如上所示。
更复杂的Consumers需要覆盖更多的方法。特别地，当通道和连接关闭时，handleShutdownSignal被调用，而handleConsumeOk方法会在任何其他回调之前传递消费者标签给Consumer。
消费者还可以实现handleCancelOk 和handleCancel 方法，分别通知显式和隐式取消。
通过消费者标签，你可以用Channel.basicCancel明确地取消一个特定的Consumer :channel.basicCancel(consumerTag);
就像发布者一样，考虑消费者的并发性风险也是很重要的。
对使用者的回调被分派到线程池中，该线程池与实例化Channel的线程分开。这意味着消费者可以安全地调用Connection 或Channel上的阻塞方法，例如 Channel#queueDeclare 或者 Channel#basicCancel。
每个Channel 都有自己的分派线程，对于一个Channel 中一个Consumer 最常见的使用情况，这意味着Consumers不支持其他Consumers。如果一个通道都有多个消费者，那么一个长时间运行的消费者可能会持有该通道上的其他消费者的回调的引用。
关于并发性和并发性危害安全的其他主题，请参阅并发性考虑(线程安全)部分
恢复个人消息（“Pull API”）
要显式地检索消息，请使用Channel.basicGet。返回值是GetResponse的一个实例，从这个实例中可以提取头信息(属性)和消息体：
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); // 确认收到的消息
处理 unroutable（无法发送的） 消息
如果消息发布时带有“强制”标志，但是不能被路由，代理将把它返回给发送客户端(通过AMQP.Basic.Return 命令)。
收到返回通知，客户端可以实现ReturnListener 接口或者调用Channel.addReturnListener。如果客户端没有为特定的通道配置一个返回监听，那么相关的返回消息将被默认丢弃。
channel.addReturnListener(new ReturnListener() {
public void handleReturn(int replyCode,
String replyText,
String exchange,
String routingKey,
AMQP.BasicProperties properties,
byte[] body)
throws IOException {
...
}
});
例如，如果客户端将带有“mandatory”标志的消息发布到一个没有绑定到队列的“direct”类型的交换机上，则将调用一个返回监听。
关闭协议
AMQP客户端关闭的概述
AMQP 0-9-1连接和通道共享相同的通用方法来管理网络故障、内部故障和显式的本地关闭。
AMQP 0-9-1连接和通道具有以下生命周期状态:
open: 对象已经准备好使用了
closing:该对象已被显式地通知在本地关闭，并向任何支持底层对象的对象发出了关闭请求，并等待它们的关闭过程完成。
closed:该对象从任何较低级别的对象中接收了所有的关闭完成通知，结果已经关闭了它自己。
这些对象总是处于关闭状态，不管导致闭包的原因是什么，比如应用程序请求、内部客户机库故障、远程网络请求或网络故障。
AMQP连接和通道对象拥有以下与关闭相关的方法:
removeShutdownListener(ShutdownListener listener)
addShutdownListener(ShutdownListener listener)
管理任何监听器，当对象转换到关闭状态时将被触发。注意，向已经关闭的对象添加一个ShutdownListener将立即触发侦听器
getCloseReason(),允许调查对象关闭的原因是什么
isOpen()，用于测试对象是否处于开放状态
close(int closeCode, String closeMessage), 显式地通知对象关闭
监听器的简单用法如下：
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.ShutdownListener;
connection.addShutdownListener(new ShutdownListener() {
public void shutdownCompleted(ShutdownSignalException cause)
{
...
}
});
关于关闭情况的一些信息
可以检索ShutdownSignalException，其中包含关于关闭原因的所有可用信息，或者通过显式调用getCloseReason()方法，或者使用ShutdownListener类的服务(ShutdownSignalException起因)方法中的原因参数。
ShutdownSignalException类提供了分析关闭原因的方法。通过调用isHardError()方法，我们可以获得信息，无论它是连接还是通道错误，而getReason()将返回关于该原因的信息，这是一个AMQP方法，即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();
...
}
}
isOpen()方法的原子性和使用
对于生产代码，不建议使用通道和连接对象的isOpen()方法，因为方法返回的值依赖于关闭原因。下面的代码说明了竞态条件的可能性:
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);
}
}
相反，我们通常应该忽略这种检查，并简单地尝试所需的操作。如果在代码执行期间，连接的通道被关闭，则会抛出一个ShutdownSignalException，指示该对象处于无效状态。我们还应该捕获由SocketException引起的IOException，当代理关闭连接时，或者当代理启动清理关闭时，或者出现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
...
}
}
高级连接选项
消费者线程池
在默认情况下，消费者线程(see Receivingbelow)将自动在一个新的ExecutorService线程池中分配 。如果需要在newConnection()方法中提供一个ExecutorService用以更好的控制，那么就需要使用这个线程池下面是一个比通常分配的更大的线程池的例子:
ExecutorService es = Executors.newFixedThreadPool(20);
Connection conn = factory.newConnection(es);
Executors 和ExecutorService 类都在java.util.concurrent 包中。
当连接关闭时，默认的ExecutorService将会关闭，但是用户提供的ExecutorService(如上面的es)不会关闭。提供定制的ExecutorService的客户端必须确保它最终被关闭(通过调用它的shutdown() 方法)，否则池的线程可能会阻止JVM终止。
相同的执行器服务可以在多个连接之间共享，也可以在重新连接上进行串行重用，但在shutdown()之后不能使用。
只有当有证据表明在处理消费者回调过程中存在严重的瓶颈时，才应该考虑使用该特性。如果没有执行或很少的消费者回调，那么默认的分配就足够了。开销最初是最小的，并且分配的线程资源是有界的，即使偶尔会出现消费活动的爆发。
使用主机列表
可以将一个Address 数组传递给newConnection()。Address 只是com.rabbitmq.client 包中的一个方便类，带有主机和端口组件。举个列子：
Address[] addrArr = new Address[]{ new Address(hostname1, portnumber1)
, new Address(hostname2, portnumber2)};
Connection conn = factory.newConnection(addrArr);
如果它不能连接到hostname2:portnumber2，将尝试连接到hostname1:portnumber1。直到地址数组中的某个地址第一次成功连接之后返回。这完全等同于在工厂中反复地设置主机和端口，每次调用factory.newconnection()，直到其中一个成功。
如果还提供了一个ExecutorService(使用factory.newConnection(es, addrArr)形式）线程池将与(第一个)成功连接相关联。
如果您想要对主机进行更多的控制，请查看服务发现的支持。
使用AddressResolver接口的服务发现
在3.6.6版本中，可以让AddressResolver的实现在创建连接时选择在哪里进行连接:
Connection conn = factory.newConnection(addressResolver);
AddressResolver接口如下：
public interface AddressResolver {
List<Address> getAddresses() throws IOException;
}
就像一个主机列表一样，返回的第一个地址将首先被尝试，然后是第二个地址，如果客户端不能连接到第一个地址，等等。
如果还提供了一个ExecutorService(使用factory.newConnection(es, addressResolver)形式）线程池将与(第一个)成功连接相关联。
AddressResolver是实现自定义服务发现逻辑的完美场所，这在动态基础结构中尤其有用。与自动恢复相结合，客户端可以自动连接到第一次启动时甚至不启动的节点。关联和负载平衡是自定义AddressResolver可能有用的其他场景。
Java客户端附带了以下实现(请参阅javadoc以获得详细信息):
DnsRecordIpAddressResolver：给定主机名，返回其IP地址(针对平台DNS服务器的解析)。这对于简单的基于dns负载平衡或故障转移非常有用。
DnsSrvRecordAddressResolver:给定服务的名称，返回主机名/端口对。搜索是作为一个DNS SRV请求实现的。这在使用类似于HashiCorp Consul的服务注册时非常有用。
心跳超时
请参阅Heartbeats guide，了解关于heartbeats的更多信息，以及如何在Java客户机中配置它们。
自定义线程工厂
像Google App Engine(GAE)这样的环境可以限制直接的线程实例化。要在这样的环境中使用RabbitMQ Java客户端，有必要配置一个自定义的ThreadFactory，该工厂使用适当的方法来实例化线程，例如GAE的ThreadManager。下面是Google App Engine的一个例子：
import com.google.appengine.api.ThreadManager;
ConnectionFactory cf = new ConnectionFactory();
cf.setThreadFactory(ThreadManager.backgroundThreadFactory());
对Java非阻塞IO的支持
4.0版本的Java客户端的为Java非阻塞IO提供支持(a.k.a 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循环的服务执行器、内存中的写队列的参数(写请求在被发送到网络之前入队)。请阅读Javadoc以获得详细信息和默认值。
从网络故障中自动恢复
连接恢复
客户端和RabbitMQ节点之间的网络连接可能会失败。RabbitMQ Java客户端支持连接和拓扑(队列、交换、绑定和消费者)的自动恢复。许多应用程序的自动恢复过程遵循以下步骤:
- Reconnect（重新连接）
- Restore connection listeners（恢复连接监听器）
- Re-open channels（重新打开通道）
- Restore channel listeners（恢复通道侦听器）
- Restore channel basic.qos setting, publisher confirms and transaction settings（恢复基本频道。qos设置、发布者确认和事务设置）
拓扑恢复包括以下操作，为每个通道执行
- Re-declare exchanges (except for predefined ones)（重新声明交换(除了预定义的交换)）
- Re-declare queues（重新声明队列）
- Recover all bindings（恢复所有绑定）
- Recover all consumers（恢复所有消费者）
在4.0.0版本的Java客户端中，默认情况下自动恢复是启用的(因此也可以进行拓扑恢复)。
使用factory.setAutomaticRecoveryEnabled(boolean)方法可以禁用或启用自动连接恢复。下面的代码片段展示了如何显式地启用自动恢复(例如，在4.0.0之前的Java客户端):
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（每10秒尝试一次恢复）
factory.setNetworkRecoveryInterval(10000);
当提供一个地址列表时，列表会被打乱，所有的地址都会被尝试，一个接着一个：
ConnectionFactory factory = new ConnectionFactory();
Address[] addresses = {new Address("192.168.1.4"), new Address("192.168.1.5")};
factory.newConnection(addresses);
恢复监听器
可以在可恢复的连接和通道上注册一个或多个恢复侦听器。当启用连接恢复，连接将会返回通过实现com.rabbitmq.client.Recoverable包下的两个方法ConnectionFactory#newConnection 和Connection#createChannel 。提供两种具有相当描述性名称的方法:
addRecoveryListener
removeRecoveryListener
注意，为了使用这些方法，您现在需要将连接和通道转换为Recoverable
对发布的影响
当连接关闭时，使用Channel.basicPublish发布的消息将丢失。客户端不会在连接恢复后将它们储入队列。为了确保发布的消息能够到达RabbitMQ应用程序，需要使用发布者确认和对连接失败做出解释说明。
拓扑恢复
拓扑恢复涉及到交换、队列、绑定和消费者的恢复。当启用自动恢复功能时，默认启用它。因此，在Java客户端4.0.0中默认启用拓扑恢复。
如果需要，可以显式地禁用拓扑恢复:
ConnectionFactory factory = new ConnectionFactory();
Connection conn = factory.newConnection();
// enable automatic recovery (e.g. Java client prior 4.0.0)
// 启用自动恢复(例如，在4.0.0之前的Java客户端)
factory.setAutomaticRecoveryEnabled(true);
// disable topology recovery
// 禁用拓扑复苏
factory.setTopologyRecoveryEnabled(false);
故障检测和恢复限制
自动连接恢复有许多限制和专门的设计决策，应用程序开发人员需要知道这些决策。
当连接关闭或丢失时，它需要时间来检测。因此，有一个时间窗口，其中库和应用程序都不知道连接失败的有效性。在此期间发布的任何消息都将被序列化，并像往常一样被写入TCP套接字。他们向代理的交付只能通过发布者确认:AMQP 0-9-1的发布完全是异步的。
当一个连接启用了自动恢复的连接时，当一个套接字或输入/输出操作错误被检测到时，在默认情况下，恢复将在默认情况下启动，默认为5秒。该设计假设，即使许多网络故障是短暂的，并且通常是短暂的，但它们不会立即消失。连接恢复尝试将在相同的时间间隔内继续，直到成功打开新连接。
当连接处于恢复状态时，任何在其通道上尝试的发布都将被一个异常拒绝。针对外出的消息，客户端当下不执行任何内部缓冲。当恢复成功时，应用程序开发人员负责跟踪这些消息并重新发布它们。发布者确认是一个协议扩展，它应该被那些不能承受消息损失的发布者使用。
当通道由于通道级异常而关闭时，连接恢复将不会启动。此类异常通常表示应用程序级别的问题。该library无法对何时发生这种情况作出明智的决定。
即使在连接恢复启动后，关闭通道也无法恢复。这包括显式关闭通道和上面的通道级异常情况。
手动确认和自动恢复
当使用手动确认时，在消息传递和确认之间可能会出现与RabbitMQ节点的网络连接失败情况。在连接恢复之后，RabbitMQ将在所有通道上重新设置传输标记。这意味着basic.ack、basic.nack和basic.reject 使用旧的交付标记将导致通道异常。为了避免这种情况，RabbitMQ Java客户端保持跟踪和更新发送标记，使它们在恢复之间单调地增长。basicack()、channel.basicnack()和channel.basicreject()然后将调整后的交付标签翻译成RabbitMQ使用的标签。使用过期的交付标签将不会被发送，使用手动确认和自动恢复的应用必须能够处理重新交付的情况。
未处理异常
与连接、通道、恢复和消费者生命周期相关的未处理异常被委托给异常处理程序。异常处理程序是实现了ExceptionHandler 接口的对象。默认情况下，使用了DefaultExceptionHandler的实例。它将异常细节输出到标准输出。
可以使用ConnectionFactory#setExceptionHandler来覆盖处理程序。它将用于工厂创建的所有连接:
ConnectionFactory factory = new ConnectionFactory();
cf.setExceptionHandler(customHandler);
异常处理程序应该用于异常日志记录。
度量和监控
在版本4.0.0中，客户端收集运行时指标(如已发布的消息的数量)。度量指标集合是可选的，并在ConnectionFactory级别上设置，使用setmetricscollsetter(metricscollsetter)方法。这个方法需要一个度量标准的实例，这个实例在客户端代码的几个地方被调用。
客户端支持Micrometer(版本4.3)和下拉向导的标准。
以下是收集到的数据:
Number of open connections （打开的连接数量）
Number of open channels （打开的通道数量）
Number of published messages （已经发布了的消息）
Number of consumed messages （消费了的消息数量）
Number of acknowledged messages （确认了的消息数量）
Number of rejected messages （被拒绝的信息）
对于与消息相关的度量标准，不管是测微计和Dropwizard指标都提供了计数，但也包括平均速率、最后五分钟速率等。他们还支持用于监视和报告的常用工具(JMX, Graphite, Ganglia, Datadog, etc)。请参阅下面的专用部分，了解更多细节。
请注意以下有关度量标准的收集:
不要忘记在使用Micrometer 或Dropwizard 度量时，将适当的依赖项(in Maven, Gradle, 或者作为 JAR 文件)添加到JVM的classpath中。这些是可选的依赖项，并且不会被Java客户端自动引用。您还可能需要根据所使用的reporting backend(s)来添加其他依赖项。
指标集合是可扩展的。为特定的需求实现一个自定义的MetricsCollector是被鼓励的。
MetricsCollector是在ConnectionFactory级别设置的，但是可以在不同的实例之间共享。
度量收集不支持事务。例如，如果在事务中发送一个确认信息，然后事务被回滚，那么确认就会被计入客户端指标(但显然不是由代理进行的)。注意，确认实际上被发送到代理，然后被事务回滚取消，因此客户端指标在发送的确认中是正确的。作为总结，不要使用客户度量来作为关键的业务逻辑，它们不能保证是完全准确的。它们的目的是简化关于运行系统的推理，使操作更加高效。
Micrometer support
您可以以下方式启用Micrometer指标收集 ：
ConnectionFactory connectionFactory = new ConnectionFactory();
MicrometerMetricsCollector metrics = new MicrometerMetricsCollector();
connectionFactory.setMetricsCollector(metrics);
...
metrics.getPublishedMessages(); // 获取 Micrometer's Counter 对象
Micrometer 支持several reporting backends: Netflix Atlas, Prometheus, Datadog, Influx, JMX, 等待等。
你通常会把MeterRegistry 的实例传递给MicrometerMetricsCollector。
下面是一个使用JMX的示例:
JmxMeterRegistry registry = new JmxMeterRegistry();
MicrometerMetricsCollector metrics = new MicrometerMetricsCollector(registry);
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setMetricsCollector(metrics);
Dropwizard指标支持
你可以采用以下方式使用Dropwizard的数据收集：
ConnectionFactory connectionFactory = new ConnectionFactory();
StandardMetricsCollector metrics = new StandardMetricsCollector();
connectionFactory.setMetricsCollector(metrics);
...
metrics.getPublishedMessages(); // 得到指标的计量对象
Dropwizard度量支持多个several reporting backends:console, JMX, HTTP, Graphite, Ganglia, etc。
你通常会把MetricsRegistry的实例传递给StandardMetricsCollector。
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();
谷歌应用引擎上的RabbitMQ Java客户端
在Google App Engine(GAE)上使用RabbitMQ Java客户端需要使用一个自定义线程工厂，该工厂使用GAE的ThreadManager实例化线程(见上文)。此外，还需要设置一个低心跳间隔(4-5秒)，以避免在GAE上运行低InputStream读取超时:
ConnectionFactory factory = new ConnectionFactory();
cf.setRequestedHeartbeat(5);
警告和限制
为了使拓扑恢复成为可能，RabbitMQ Java客户端维护已声明的队列、交换和绑定的缓存。缓存是每个连接。某些RabbitMQ特性使得客户端不可能观察到某些拓扑更改，例如，由于TTL而删除了一个队列。RabbitMQ Java客户机尝试在下面的情况下使缓存项无效：
When queue is deleted.（当队列被删除）
When exchange is deleted.（当交换被删除）
When binding is deleted.（当绑定被删除）
When consumer is cancelled on an auto-deleted queue.（当消费者被自动删除的队列取消时。）
When queue or exchange is unbound from an auto-deleted exchange.（当队列或交换从自动删除的交换中释放时。）
但是，客户端无法跟踪这些拓扑更改，而不仅仅是单个连接。依赖于自动删除队列或交换的应用程序，以及队列TTL(注意:不是消息TTL！)，以及使用自动连接恢复，应该显式地删除那些不被使用或删除的实体，以清除客户端拓扑缓存。这是由Channel#queueDelete, Channel#exchangeDelete, Channel#queueUnbind和Channel#exchangeUnbind 在RabbitMQ 3.3.x中具有幂等性的。(删除不存在的内容并不会导致异常)。
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的一个简单的便利包装器，将消息体作为默认字符编码中的字符串实例。）
mapCall 的变体稍微复杂一点：它们将一个包含普通Java值的 java.util.Map 编码为AMQP 0-9-1二进制表表示，并以同样的方式解码响应。（注意，这里有一些限制，针对什么类型的值可以在这里使用——请参阅javadoc以获得详细信息）
所有 marshalling/unmarshalling 便利方法使用 primitiveCall 作为传输机制,并提供包装层之上。
TLS的支持
可以使用TLS加密客户端和代理之间的通信。还支持客户端和服务器身份验证(也称为对等验证)。下面是使用Java客户端的加密最简单的方法:
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5671);
factory.useSslProtocol();
注意，在上面的示例中，客户端默认不强制执行任何服务器身份验证(对等证书链验证)，“trust all certificates”TrustManager 被使用。这对于本地开发来说很方便，但容易发生中间人攻击，因此不建议生产。要在RabbitMQ中了解更多关于TLS支持的信息，请参阅“TLS指南”。如果您只想配置Java客户端（特别是对等验证和信任管理器部分），请阅读TLS指南的适当部分。
================================================================
转载声明：
文章转载自https://blog.csdn.net/csdnzouqi/article/details/78926603
================================================================
CSDN联合极客时间，共同打造面向开发者的精品内容学习社区，助力成长！
更多推荐
0
0- 0
已为社区贡献1条内容
所有评论(0)