代码编织梦想

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

六、原子性和 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);
    }
}

相反,我们通常应该忽略这种检查,并且 只需尝试所需的操作即可。如果在 执行代码的连接通道是 关闭,关闭信号异常将是 抛出指示对象处于无效状态 州。我们还应该捕获由套接字异常引起的 IOException,当 代理意外关闭连接,或关闭信号异常,当代理 启动干净关闭。

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
        ...
    }
}
版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/leesinbad/article/details/129613038