RabbitMQ Java开发教程（二）—官方原版
一、通道和并发注意事项（线程安全）
应避免在线程之间共享通道实例。应用程序应该每个线程使用一个通道，而不是在多个线程之间共享同一个通道。
虽然通道上的一些操作可以安全地同时调用，但有些操作则不然，并且会导致线路上不正确的帧交织、双重确认等。
共享通道上的并发发布可能会导致连线上不正确的帧交错，从而触发连接级协议异常，并由代理立即关闭连接。因此，它需要在应用程序代码中进行显式同步（必须在关键部分调用Channel#basicPublish）。线程之间共享通道也会干扰Publisher Confirms。最好完全避免在共享通道上并发发布，例如每个线程使用一个通道。
可以使用通道池来避免在共享通道上并发发布：一旦一个线程处理完一个通道，它就会将其返回到池中，使该通道可供另一个线程使用。信道池可以被认为是一种特定的同步解决方案。建议使用现有的池库，而不是自行开发的解决方案。例如，Spring AMQP，它附带了一个可随时使用的通道池功能。
通道消耗资源，在大多数情况下，应用程序很少需要在同一JVM进程中打开数百个以上的通道。如果我们假设应用程序为每个通道都有一个线程（因为通道不应该同时使用），那么单个JVM的数千个线程已经是可以避免的相当大的开销。此外，一些快速发布者可以很容易地使网络接口和代理节点饱和：发布所需的工作比路由、存储和传递消息所需的工作量少。
要避免的一个经典反模式是为每个发布的消息打开一个通道。信道应该是相当长寿的，而打开一个新的信道是一个网络往返，这使得这种模式效率极低。
在一个线程中消费并在共享通道上的另一个线程上发布是安全的。
服务器推送的交付（请参阅下面的部分）是在保证保留每个通道的订购的同时进行调度的。调度机制使用java.util.concurrent.ExecutorService，每个连接一个。可以提供一个自定义执行器，该执行器将由单个ConnectionFactory使用ConnectionFactory#setSharedExecutor setter生成的所有连接共享。
当使用手动确认时，重要的是要考虑是哪个线程进行确认。如果它与接收传递的线程不同（例如，Consumer#handleDelivery将传递处理委托给不同的线程），将多个参数设置为true进行确认是不安全的，并且会导致双重确认，从而导致关闭通道的通道级协议异常。一次确认一条信息是安全的。
二、通过订阅接收消息（“PUSH API”）
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;接收消息的最有效方法是设置 使用使用者界面进行订阅。然后消息将被传递 在他们到达时自动，而不必 明确要求。
调用与使用者相关的 API 方法时，单个订阅是 总是由他们的消费者标签引用。消费者标签是消费者 标识符，可以是客户端生成的，也可以是服务器生成的。要让 RabbitMQ 生成节点范围的唯一标签，使用 Channel#basicConsumption 覆盖 不接受消费者标签参数或传递空字符串 ，并使用 Channel#basicConsumption 返回的值。 消费者标签用于取消消费者。
不同的使用者实例必须具有不同的 消费者标记。连接上的重复使用者标记是 强烈建议不要使用，并可能导致自动问题 连接恢复和混淆监控数据时 消费者受到监控。
实现消费者的最简单方法是 子类 便利类 DefaultConsumer。 可以通过 basicConsumption 调用传递此子类的对象以设置订阅：
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();
// (process the message components here ...)
channel.basicAck(deliveryTag, false);
}
});这里，由于我们指定了autoAck=false，因此有必要确认传递给Consumer的消息，这在handleDelivery方法中最方便，如图所示。
更复杂的消费者将需要推翻进一步的方法。特别是，当通道和连接关闭时，将调用handleShutdownSignal，并且在调用对该consumer的任何其他回调之前，将向handleConsumeOk传递consumer标记。
消费者还可以实现handleCancelOk和handleCancel方法，分别获得显式和隐式取消的通知。
您可以使用Channel.basicCancel明确取消特定消费者：
channel.basicCancel(consumerTag);
传递消费者标签。
就像发行商一样，考虑消费者的并发危害安全也是很重要的。
对消费者的回调是在一个线程池中调度的，该线程池与实例化其通道的线程分开。这意味着消费者可以安全地在Connection或Channel上调用阻塞方法，如Channel#queueDeclare或Channel#basicCancel。
每个通道将按照RabbitMQ发送的顺序，将所有交付发送到其上的Consumer处理程序方法。无法保证渠道之间的交货顺序：这些交货可以并行发送。
对于每个频道一个消费者的最常见用例，这意味着消费者不会阻碍其他消费者。由于每个频道有多个消费者，请注意，长时间运行的消费者可能会阻碍向该频道上的其他消费者发送回调。
三、检索单个消息（“pull API”）
也可以按需检索单个消息（“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); // acknowledge receipt of the message
四、处理不可路由的消息
如果发布消息时设置了“强制”标志， 但无法路由，经纪人会将其返回给 发送客户端（通过 AMQP。基本返回命令）。
要收到此类返回的通知，客户端可以实现 ReturnListener 接口并调用 Channel.addReturnListener。 如果客户端尚未为特定通道配置返回侦听器， 然后，关联的返回消息将被静默丢弃。
channel.addReturnListener(new ReturnListener() {
publicvoidhandleReturn(int replyCode,
String replyText,
String exchange,
String routingKey,
AMQP.BasicProperties properties,
byte[] body)throws IOException {
...
}
});将调用返回侦听器，例如，如果客户端发布消息 “强制”标志设置为未绑定到队列的“直接”类型的交换。
五、关断协议
1、客户端关闭过程概述
AMQP 0-9-1 连接和通道共享相同的常规 管理网络故障、内部故障的方法、 和显式本地关闭。
AMQP 0-9-1 连接和通道具有以下生命周期状态：
打开：对象已准备就绪，可供使用
关闭：对象已显式 通知本地关闭，已发出关闭 请求任何支持的下层对象，并且 等待其关闭程序完成
已关闭：对象已收到全部 来自任何较低层的关机完成通知 对象，并因此自行关闭
这些对象总是以关闭状态结束， 无论导致关闭的原因是什么，例如 应用程序请求，内部客户端库 故障、远程网络请求或网络故障。
连接和通道对象具有 以下与关机相关的方法：
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 void shutdownCompleted(ShutdownSignalException cause)
{
...
}
});2、有关关闭情况的信息
可以检索 ShutdownSignalException，其中包含所有 有关关闭原因的可用信息，或者 通过显式调用 getCloseReason（） 方法或使用 ShutdownListener 类的服务（ShutdownSignalException cause） 方法。
类提供 分析关机原因的方法。由 调用 isHardError（） 方法我们得到 信息是连接还是通道 错误，getReason（） 返回信息 关于原因，以AMQP方法的形式 - 要么AMQP。Channel.Close 或 AMQP。Connection.Close （如果原因为 null，则为 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();
...
}
}