RabbitMQ 常用 API

Connection 和 Channel 的创建、关闭

• 创建 Connection

  ConnectionFactory factory = new ConnectionFactory();
  // 方式1：通过设置参数创建
  factory.setHost(IP_ADDRESS);
  factory.setPort(PORT);
  factory.setUsername("guest");
  factory.setPassword("guest");
  Connection connection = factory.newConnection();
  
  // 方式2：通过uri创建
  factory.setUri("amqp:/userName:password@ipAddress:portNumber/virtualHost");
  Connection connection = factory.newConnection();
  

• 创建 Channel

  Channel channel = connection.createChannel();
  

  Connection 可以用来创建多个 Channel 实例，但是 Channel 实例不能在线程间共享，应用程序应该为每一个线程开辟一个Channel。某些情况下 Channel 的操作可以并发运行，但是在其他情况下会导致在网络,上出现错误的通信帧交错，同时也会影响发送方确认（publisherconfirm）机制的运行，所以多线程间共享 Channel 实例是非线程安全的。

  通常情况下，在调用 createXXX 或者 newXXX 方法之后，可以简单地认为 Connection 或者 Channel 已经成功地处于开启状态，而并不会在代码中使用 isOpen 这个检测方法。如果在使用 Channel 的时候其已经处于关闭状态，那么程序会抛出一个 com.rabbitmq.client.ShutdownSignalException，只需捕获这个异常即可。当然同时也要试着捕获 IOException 或者 SocketException，以防 Connection 意外关闭。

• 关闭连接

  // 关闭Connection
  connection.close();
  // 关闭Channel
  channel.close();
  

  AMQP 协议中的 Connection 和 Channel 采用同样的方式来管理网络失败、内部错误和显式地关闭连接。

  Connection 和Channel 所具备的生命周期如下：

  • Open：开启状态，代表当前对象可以使用
  • Closing：正在关闭状态。当前对象被显式地通知调用关闭方法（shutdown），这样就产生了一个关闭请求让其内部对象进行相应的操作，并等待这些关闭操作的完成
  • Closed：已经关闭状态。当前对象已经接收到所有的内部对象已完成关闭动作的通知，并且其也关闭了自身。

  Connection 和 Channel 最终都是会成为 Closed 的状态，不论是程序正常调用的关闭方法，或者是客户端的异常，再或者是发生了网络异常。

  拓展了解：

  在 Connection 和 Channel 中，与关闭相关的方法有：

  addShutdownListener(ShutdownListener listener) 
  removeShutdownListener(ShutdownListner listener)
  

  当 Connection 或者 Channel 的状态转变为 Closed 的时候会调用 ShutdownListener。而且如果将一个 ShutdownListener 注册到一个已经处于 Closed 状态的对象（这里特指 Connection 和 Channel 对象）时，会立刻调用 ShutdownListener

  • getCloseReason 方法：可以知道对象关闭的原因
  • isOpen 方法：检测对象当前是否处于开启状态
  • close (int closeCode, String closeMessage) 方法：显式地通知当前对象执行关闭操作

  connection.addShutdownListener(new ShutdownListener() {
      @Override
      public void shutdownCompleted(ShutdownSignalException cause) {
          //...
          System.out.println("shut down");
      }
  });
  

  触发 ShutdownListener 的时候，就可以获取到 ShutdownSignalException，这个 ShutdownSignalException 包含了关闭的原因，这里原因也可以通过调用前面所提及的 getCloseReason 方法获取。

  • ShutdownSignalException：提供了多个方法来分析关闭的原因
  • isHardError 方法：可以知道是 Connection 的还是 Channel 的错误
  • getReason 方法：可以获取cause相关的信息

  connection.addShutdownListener(new ShutdownListener() {
      @Override
      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();
                  //...
              }
          }
      }
  });
  

Channel 常用 API

交换器的创建、检测、删除

• 创建交换机

  Exchange.DeclareOk exchangeDeclare(String exchange, String type) throws IOException;  // 常用
  Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable) throws IOException;
  Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, 
                                     Map<String, Object> arguments) throws IOException;
  Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, 
                                     boolean autoDelete, boolean internal, 
                                     Map<String, Object> arguments) throws IOException;
  
  // 不需要服务器任何返回值的创建交换机
  void exchangeDeclareNoWait(String exchange, String type, boolean durable, 
                             boolean autoDelete, boolean internal,
                             Map<String, Object> arguments) throws IOException;
  

  • 返回值：Exchange . Declare0K，用来标识成功声明了一个交换器。

  • 入参：

    • exchange：交换器的名称

    • type：交换器的类型，常见的如 fanout、direct、 topic

    • durable：设置是否持久化。为 true 表示持久化，反之是非持久化。

      持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息。

    • autoDelete：设置是否自动删除。true 则表示自动删除。

      自动删除的前提是至少有一个队列或者交换器与这个交换器绑定，之后所有与这个交换器绑定的队列或者交换器都与此解绑。

      注意：不能错误地把这个参数理解为：“当与此交换器连接的客户端都断开时，RabbitMQ 会自动删除本交换器”。

    • internal：设置是否是内置的。

      为 true 则表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式。

    • argument：其他一些结构化参数，比如 alternate-exchange。

  • 不需要服务器任何返回值的创建交换机这个 nowait 指的是 AMQP 中 Exchange.Declare 命令的参数，意思是不需要服务器返回

    注意：

    • 这个方法的返回值是 void，而普通的 exchangeDeclare 方法的返回值是 Exchange.DeclareOk，意思是在客户端声明了一个交换器之后，需要等待服务器的返回（服务器会返回 Exchange. Declare-Ok 这个 AMQP 命令）
    • 在声明完一个交换器之后（实际服务器还并未完成交换器的创建），那么此时客户端紧接着使用这个交换器，必然会发生异常。如果没有特殊的缘由和应用场景，并不建议使用这个方法。

• 检测交换器是否存在

  Exchange.DeclareOk exchangeDeclarePassive(String name) throws IOException;
  

  这个方法在实际应用过程中还是非常有用的，它主要用来检测相应的交换器是否存在。

  如果存在则正常返回；如果不存在则抛出异常：404 channel exception，同时Channel也会被关闭。

• 删除交换器

  Exchange.DeleteOk exchangeDelete(String exchange) throws IOException;
  Exchange.DeleteOk exchangeDelete(String exchange, boolean ifUnused) throws IOException;
  
  void exchangeDeleteNoWait(String exchange, boolean ifUnused) throws IOException;
  

  • exchange：表示交换器的名称

  • ifUnused：用来设置是否在交换器没有被使用的情况下删除。

    为 true，则只有在此交换器没有被使用的情况下才会被删除；为 false，则无论如何这个交换器都要被删除。

队列的创建、检测、删除、清空

• 创建队列

  Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, 
                               Map<String, Object> arguments) throws IOException;
  // 默认创建一个由RabbitMQ命名的(类似这种amq.gen-LhQz1gv3GhDOv8PIDabOXA名称，这种队列也称之为匿名队列)、排他的、自动删除的、非持久化的队列。
  Queue.DeclareOk queueDeclare() throws IOException;
  
  void queueDeclareNoWait(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;
  

  • queue： 队列的名称。如果队列不存在，自动创建

  • durable： 设置是否持久化。为 true 则设置队列为持久化。

    持久化的队列会存盘，在服务器重启的时候可以保证不丢失相关信息。

  • exclusive： 设置是否排他。为 true 则设置队列为排他的。

    如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。

    注意：

    • 排他队列是基于连接（Connection）可见的，同一个连接的不同信道（Channel）是可以同时访问同一连接创建的排他队列
    • “首次”是指如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同;
    • 即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列适用于一个客户端同时发送和读取消息的应用场景。

  • autoDelete： 设置是否自动删除。为 true 则设置队列为自动删除。

    自动删除的前提是：至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。

    不能把这个参数错误地理解为：“当连接到此队列的所有客户端断开时，这个队列自动删除”，因为生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列。

  • arguments：设置队列的其他一些参数，如 x-message=ttl、x-expires、x-max-length、x-max-length-bytes、x-dead-letter-exchange、x-dead-letter-routing-key、x-max-priority等。

  **注意：**生产者和消费者都能够使用 queueDeclare 来声明一个队列，**但是如果消费者在同一个信道上订阅了另一个队列，**就无法再声明队列了。必须先取消订阅，然后将信道置为“传输”模式，之后才能声明队列。

• 检测队列

  Queue.DeclareOk queueDeclarePassive(String queue) throws IOException;
  

  • 存在，返回 Queue.DeclareOk
  • 不存在，抛出异常

• 删除队列

  Queue.DeleteOk queueDelete(String queue) throws IOException;
  Queue.DeleteOk queueDelete(String queue, boolean ifUnused, boolean ifEmpty) throws IOException;
  
  void queueDeleteNoWait(String queue, boolean ifUnused, boolean ifEmpty) throws IOException;
  

• 清空队列

  Queue.PurgeOk queuePurge(String queue) throws IOException;
  

  清空队列的内容，而不删除队列本身

队列和交换器的绑定、解除

• 绑定队列和交换器

  Queue.BindOk queueBind(String queue, String exchange, String routingKey) throws IOException;
  Queue.BindOk queueBind(String queue, String exchange, String routingKey, 
                         Map<String, Object> arguments) throws IOException;
  
  void queueBindNoWait(String queue, String exchange, String routingKey, 
                       Map<String, Object> arguments) throws IOException;
  

  • queue：队列名称
  • exchange：交换器的名称
  • routingKey：用来绑定队列和交换器的路由键
  • argument：定义绑定的一些参数

• 解除队列和交换器的绑定

  Queue.UnbindOk queueUnbind(String queue, String exchange, String routingKey) throws IOException;
  Queue.UnbindOk queueUnbind(String queue, String exchange, String routingKey, 
                             Map<String, Object> arguments) throws IOException;
  

绑定交换器和交换器

• API 方法

  Exchange.BindOk exchangeBind(String destination, String source, String routingKey) throws IOException;
  Exchange.BindOk exchangeBind(String destination, String source, String routingKey, 
                               Map<String, Object> arguments) throws IOException;
  
  void exchangeBindNoWait(String destination, String source, String routingKey, 
                          Map<String, Object> arguments) throws IOException;
  

• 示例：

  channel.exchangeDeclare("source", "direct", false, true, null);
  channel.exchangeDeclare("destination", "fanout", false, true, null);
  channel.exchangeBind("destination", "source", "exKey");
  channel.queueDeclare("queue", false, false, true, null);
  channel.queueBind("queue", "destination“, "");
  channel.basicPublish("source", "exKey", null, "exToExDemao".getBytes());
  

  生产者发送消息至交换器 source 中，交换器 source 根据路由键找到与其匹配的另一个交换器 destination，并把消息转发到destination中，进而存储在 destination 绑定的队列 queue 中

发送消息

• 方法：

  void basicPublish(String exchange, String routingKey, 
                    BasicProperties props, byte[] body) throws IOException;  // 常用
  void basicPublish(String exchange, String routingKey, boolean mandatory, 
                    BasicProperties props, byte[] body)throws IOException;
  void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, 
                    BasicProperties props, byte[] body)throws IOException;
  

  • exchange：交换器的名称，指明消息需要发送到哪个交换器中

    如果设置为空字符串，则消息会被发送到 RabbitMQ 默认的交换器中。

  • routingKey： 路由键，交换器根据路由键将消息存储到相应的队列之中

  • props ： 消息的基本属性集，其包含14个属性成员，分别有 contentType、contentEncoding、headers(Map<String, object>)、deliveryMode、priority、correlationId、replyTo、expiration、messageId、timestamp、type、userId、appId、clusterId。

  • byte[] body： 消息体（payload），真正需要发送的消息

  • mandatory：

    • 为 true 时，如果 exchange 根据自身类型和消息 routingKey 无法找到一个合适的 queue 存储消息，那么 broker 会调用 basic.return 方法将消息返还给生产者；
    • 为 false 时，出现上述情况 broker 会直接将消息丢弃；

  • immediate：

    • 为 true 时，如果该消息关联的 queue 上有消费者，则马上将消息投递给它，如果所有 queue 都没有消费者，直接把消息返还给生产者，不用将消息入队列等待消费者了。
    • 为 false 时，出现上述情况 broker 会直接将消息丢弃；

• 示例：

  byte[] messageBodyBytes = "Hello World!".getBytes();
  // 示例：发送一条内容为“Hello World!”的消息
  channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);
  // 示例：为了更好地控制发送，可以使用mandatory这个参数，或者可以发送一些特定属性的信息
  channel.basicPublish(exchangeName, routingKey, mandatory, 
                       MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
  // 示例：这条消息的投递模式(delivery mode)设置为2，即消息会被持久化(即存入磁盘)在服务器中。同时这条消息的优先级( priority)设置为1, content-type为“text/plain”。可以自己设定消息的属性。
  channel.basicPublish(exchangeName, routingKey, 
  		new AMQP.BasicProperties().builder().contentType("text/plain")
              .deliveryMode(2).priority(1).userId("hidden").build(), 
         messageBodyBytes);
  // 示例：发送一条带有headers的消息
  Map<String, Object> headers = new HashMap<>();
  headers.put("localtion", "here");
  headers.put("time", "today");
  channel.basicPublish(exchangeName, routingKey,
          new AMQP.BasicProperties().builder().headers(headers).build(),
          messageBodyBytes);
  // 示例：发送一条带有过期时间（expiration）的消息
  byte[] messageBodyBytes = "Hello World!".getBytes();
  channel.basicPublish(exchangeName, routingKey, 
  		new AMQP.BasicProperties.Builder().expiration("60000").build(), 
          messageBodyBytes); 
  

消费消息

RabbitMQ 的消费模式分两种:

• 推（Push）模式

  推模式采用 Basic.Consume 进行消费

• 拉（Pull）模式

  拉模式调用 Basic.Get 进行消费

推模式

在推模式中，可以通过持续订阅的方式来消费消息。

Channel 类中的常用 basicConsume 方法：

String basicConsume(String queue, Consumer callback) throws IOException;
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;  // 常用
String basicConsume(String queue, boolean autoAck, String consumerTag, 
                    Consumer callback) throws IOException;
String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments, 
                    Consumer callback) throws IOException;
String basicConsume(String queue, boolean autoAck, String consumerTag, 
                    boolean noLocal, boolean exclusive, 
                    Map<String, Object> arguments, Consumer callback) throws IOException;
String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments, 
                    DeliverCallback deliverCallback, CancelCallback cancelCallback, 
                    ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;

• queue：消息队列的名称

• autoAck：是否自动确认（为了确保消息不会丢失，RabbitMQ 支持消息应答）

  建议设成 false，即不自动确认。消费者在已经接收并且处理完毕消息后调用 channel.basicAck 来确认消息已被成功接收（发送一个消息应答，告诉 RabbitMQ 这个消息已经接收并且处理完毕了，RabbitMQ 可以删除它了）

• consumerTag：消费者标签，用来区分多个消费者

  当调用与 Consumer 相关的 API 方法时，不同的订阅采用不同的消费者标签（consumerTag）来区分彼此，在同一个 Channel 中的消费者也需要通过唯一的消费者标签以作区分。

• noLocal：为 true 则表示不能将同一个 Connection 中生产者发送的消息传送给这个 Connection 中的消费者

• exclusive：设置是否排他

• arguments：设置消费者的其他参数

• deliverCallback：（消息传递时回调）设置消费者的回调参数。

  用来处理 RabbitMQ 推送过来的消息，比如 DefaultConsumer，使用时需要客户端重写其中的方法

• cancelCallback：消费者被取消时回调

• shutdownSignalCallback：当通道/连接关闭时回调

• consumer：设置消费者的回调函数

  用来处理RabbitMQ 推送过来的消息，接收消息一般通过实现 com.rabbitmq.client.Consumer 接口或者继承 com.rabbitmq.client.DefaultConsumer 类来实现。

  对于消费者客户端来说重写 DefaultConsumer 类的 handleDelivery 方法是十分方便的。更复杂的消费者客户端会重写更多的方法，具体如下：

  // 在其他方法之前调用
  public void handleConsumeOk(String consumerTag)
  // 在显示地或者隐式的取消订阅时调用
  public void handleCancelOk(String consumerTag)
  public void handleCancel(String consumerTag) throws IOException
  // 当Channel或者Connection关闭的时候会调用
  public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig)
      
  public void handleRecoverOk(String consumerTag)
  

  注：

  • 可以通过 channel.casicCancel 方法来显示的取消一个消费者的订阅：

    channel.basicCancel(consumerTag);
    

    该行代码会首先触发 handleConsumerOk 方法，之后触发 handleDelivery 方法，最后才触发 handleCancelOk 方法

和生产者一样，消费者客户端同样需要考虑线程安全的问题。消费者客户端的这些 callback 会被分配到与 Channel 不同的线程池上，这意味着消费者客户端可以安全地调用这些阻塞方法，比如 channel.queueDeclare、channel.basicCancel 等。

每个 Channel 都拥有自己独立的线程。最常用的做法是一个 Channel 对应一个消费者，也就是意味着消费者彼此之间没有任何关联。当然也可以在一个 Channel 中维持多个消费者，但是要注意一个问题，如果 Channel 中的一个消费者一直在运行，那么其他消费者的 callback 会被“耽搁”。

示例：

boolean autoAck = false;
channel.basicQos(64);
String consumerTag = "myConsumerTag";
channel.basicConsume(queueName, autoAck, consumerTag,
        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();
                
                //process the message components here ...
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });

注意：上面代码中显式地设置 autoAck=false，然后在接收到消息之后进行显式 ack 操作(channel.basicAck )，对于消费者来说这个设置是非常必要的，可以防止消息不必要地丢失。

拉模式

拉模式的消费方式：通过 channel.basicGet 方法可以单条地获取消息，其返回值是 GetRespone

Channel 类的 basicGet 方法：

GetResponse basicGet (String queue， boolean autoAck).throws IOException;

• queue ：队列的名称

• autoAck：是否自动确认（为了确保消息不会丢失，RabbitMQ 支持消息应答）

  建议设成 false，即不自动确认。消费者在已经接收并且处理完毕消息后调用 channel.basicAck 来确认消息已被成功接收（发送一个消息应答，告诉 RabbitMQ 这个消息已经接收并且处理完毕了，RabbitMQ 可以删除它了）

示例：

GetResponse getResponse = channel.basicGet(QUEUE_NAME, false);
System.out.println(new String(getResponse.getBody()));
channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false);

图示：

推 | 拉模式 总结

• Basic. Consume 将信道（Channel）置为接收模式，直到取消队列的订阅为止。

  在接收模式期间，RabbitMQ 会不断地推送消息给消费者，当然推送消息的个数还是会受到 Basic.Qos 的限制。

• 如果只想从队列获得单条消息而不是持续订阅，建议使用 Basic.Get 进行消费。

  但是不能将 Basic.Get 放在一个循环里来代替 Basic.Consume，这样做会严重影响 RabbitMQ 的性能。如果要实现高吞吐量，消费者理应使用 Basic. Consume 方法。

消费端的确认与拒绝

为了保证消息从队列可靠地达到消费者，RabbitMQ 提供了消息确认机制（messageacknowledgement）。消费者在订阅队列时，可以指定 autoAck 参数，当 autoAck 等于 false 时，RabbitMQ 会等待消费者显式地回复确认信号后才从内存（或者磁盘）中移去消息（实质上是先打上删除标记，之后再删除）。当 autoAck 等于 true 时，RabbitMQ 会自动把发送出去的消息置为确认，然后从内存（或者磁盘）中删除，而不管消费者是否真正地消费到了这些消息。

采用消息确认机制后，只要设置 autoAck 参数为 false，消费者就有足够的时间处理消息（任务），不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，因为 RabbitMQ 会一直等待持有消息直到消费者显式调用 Basic.Ack 命令为止。

当 autoAck 参数置为 false，对于 RabbitMQ 服务端而言，队列中的消息分成了两个部分:

• 一部分是等待投递给消费者的消息

• 一部分是已经投递给消费者，但是还没有收到消费者确认信号的消息

  如果 RabbitMQ 一直没有收到消费者的确认信号，并且消费此消息的消费者已经断开连接，则 RabbitMQ 会安排该消息重新进入队列，等待投递给下一个消费者，当然也有可能还是原来的那个消费者。

  RabbitMQ 不会为未确认的消息设置过期时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开，这么设计的原因是 RabbitMQ 允许消费者消费一条消息的时间可以很久很久。

RabbtiMQ的Web 管理平台上可以看到当前队列中的“Ready”状态和“Unacknowledged”状态的消息数，分别对应上文中的等待投递给消费者的消息数和已经投递给消费者但是未收到确认信号的消息数：

限制未确认消息数量

void basicQos(int prefetchCount) 
void basicQos(int prefetchCount, boo1ean global) 
void basicQos(int prefetchSize, int prefetchCount, boo1ean global) 

• prefetchSize：消费者所能接收未确认消息的总体大小的上限，单位为 B

• prefetchCount：限制信道上的消费者所能保持的最大未确认消息的数量

  即一旦超过指定个数的消息还没有 ack（确认），则该 consumer 将 block 掉，直到有消息 ack

• global：是否将上面设置应用于 channel，简单点说，就是上面限制是 channel 级别的还是 consumer 级别

拒绝消息

在消费者接收到消息后，如果想明确拒绝当前的消息而不是确认，RabbitMQ 在 2.0.0 版本开始引入了Basic.Relject 这个命令，消费者客户端可以调用与其对应的 channel. basicReject 方法来告诉 RabbitMQ 拒绝这个消息。

Channel 类中的 API 方法定义如下：

// 拒绝一条消息
void basicReject(long deliveryTag, boolean requeue) throws IOException;
// 批量拒绝消息
void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;

• deliveryTag 参数：消息的编号，它是一个 64 位的长整型值，最大值是 9223372036854775807
• requeue 参数：是否重新将这条消息存入队列
  • 设置为 true，则 RabbitMQ 会重新将这条消息存入队列，以便可以发送给下一个订阅的消费者
  • 设置为 false，则 RabbitMQ 立即会把消息从队列中移除，而不会把它发送给新的消费者
• multiple 参数：是否批量拒绝消息
  • 设置为 false，则表示拒绝编号为 deliveryTag 的这一条消息，这时候 basicNack 和 basicReject 方法一样
  • 设置为 true，则表示拒绝 deliveryTag 编号之前所有未被当前消费者确认的消息
• 注：将 channel.basicReject 或者 channel.basicNack 中的 requeue 设置为 false，可以启用“死信队列”的功能。死信队列可以通过检测被拒绝或者未送达的消息来追踪问题。

重新发送未确认的消息

对于 requeue，AMQP 中的 Basic. Recover 命令具备可重入队列的特性，可以用来请求 RabbitMQ 重新发送还未被确认的消息。

其对应的客户端方法为:

Basic.RecoverOk basicRecover() throws IOException;
Basic.RecoverOk basicRecover(boolean requeue) throws IOException;

• requeue 参数：是否将未被确认的消息分配给与之前相同的消费者
  • 设置为 true，则未被确认的消息会被重新加入到队列中，对于同一条消息来说，可能会被分配给与之前不同的消费者
  • 设置为 false，那么同一条消息会被分配给与之前相同的消费者
  • 默认情况下，如果不设置 requeue 参数，requeue 默认为 true

参考

• 官网
• 官方教程（英文）
• 官网下载地址
• [RabbitMQ消息模型详解](