RabbitMQ 客户端API使用

￡神魔★判官ぃ 2022-05-13 05:24 243阅读 0赞

**目录**

一、连接RabbitMQ服务器

二、使用交换器和队列

1.声明查找删除交换器

2. 声明查找删除队列

3. 队列绑定交换器和解绑 queueBind

4.交换器和交换器绑定exchangeBind

三.发送消息

四.消费消息

1. 推模式push

2. 拉模式

3. 消费信息的确认

4. 消息的拒绝

5. 消息的重新获取

五.关闭连接

--------------------

RabbitMQ Java客户端使用com.rabbitmq.client作为顶级包名，关键的Class和Interface有Channel、Connection、ConnectionFactory、Consumer等。AMQP协议层面的操作通过Channel接口实现。Connection是用来开启Channel(信道)的，可以注册事件处理器，也可以在应用结束时关闭连接。与RabbitMQ相关的开发工作，基本上也是围绕Connection和Channel这两个类展开的。

这里按照一个完整的运转流程进行讲解，详细内容有:连接、交换器/队列的创建与绑定、发送消息、消费消息、消费消息的确认和关闭连接。

## 一、连接RabbitMQ服务器 ##

连接ActiveMQ服务器的方式一：

ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setVirtualHost(virtualHost); //虚拟消息服务器host
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection(); //创建连接

使用uri来连接ActiveMQ方式二:

ConnectionFactory factory = new ConnectionFactory();
            factory.setUri("amqp://username:password@ipAddress:portNumber/virtualHost");
            Connection connection = factory.newConnection(); //创建连接
            Channel channel = connection.createChannel();//创建信道

**注意点：**

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

Channel或者Connection中有个isOpen方法可以用来检测其是否己处于开启状态(），但是我们一般不使用这个方法，一般用CreateXXX,newXX方法后，我们默认的认为Connection，Channel已经成功的开启了，如果在使用Channel的时候其己经处于关闭状态，那么程序会抛出一个com.rabbitmq.client.ShutdownSignalException，我们只需捕获这个异常即可。当然同时也要试着捕获IOExceptio口或者SocketException，以防Connection意外关闭。

## 二、使用交换器和队列 ##

交换器和队列是AMQP中high level层面的构建模块，应用程序需确保在使用它们的时候就已经存在了，在使用之前需要先声明(declare)它们,然后再bind起来，这其中涉及到了bindingKey，代码：

Connection connection = factory.newConnection(); //创建连接
            Channel channel = connection.createChannel();//创建信道
            //创建一个type="direct" 、持久化，非自动删除的交换器
            channel.exchangeDeclare(EXCHANGE_NAME,"direct",true,false,null);
            //创建一个持久化，非排他的，非自动删除的队列
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            //将交换器与队列通过路由绑定
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);

### 1.声明查找删除交换器 ###

**1.1 声明交换器方法如下：   **

DeclareOk exchangeDeclare(String exchange, String type, boolean durable, 
        boolean autodelete,boolean internal, Map<String, Object> arguments) throws IOException;

> 参数详细说明如下所述：
> 
>  *  exchange:交换器的名称。
>  *  type:交换器的类型，常见的如fanout、direct、topic，header 详情看[上一章][Link 1]。
>  *  durable:设置是否持久化。durable设置为true表示持久化，反之是非持久化。持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息。
>  *  autoDelete:设置是否自动删除。autoDelete设置为true则表示自动删除。自动删除的前提是至少有一个队列或者交换器与这个交换器绑定，之后所有与这个交换器绑定的队列或者交换器都与此解绑。注意不能错误地把这个参数理解为:"当与此交换器连接的客户端都断开时，RabbitMQ会自动删除本交换器"。
>  *  internal:设置是否是内置的。如果设置为true，则表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式。
>  *  argument:其他一些结构化参数，比如alternate-exchange(有alternateexchange的详情后面介绍)。

相关的重载方法：

其中没有带的参数，布尔值默认是false,Map参数为null

DeclareOk exchangeDeclare(String exchange, String type) throws IOException;
        DeclareOk exchangeDeclare(tring exchange, String type, boolean durable) throws IOException;
        DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments) throws IOException;
        DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException;

1.2 与上面相对应的把第二参数变为**BuiltinExchangeType type**,这是封装的枚举type，也可以通过这个来生命交换器类型，源码如下：

public enum BuiltinExchangeType {
        DIRECT("direct"),
        FANOUT("fanout"),
        TOPIC("topic"),
        HEADERS("headers");
        private final String type;
        private BuiltinExchangeType(String type) {
            this.type = type;
        }
        public String getType() {
            return this.type;
        }
    }

** 1.3 还有声明为noWait的方法**，这个nowait参数指的是AMQP中Exchange.Declare命令的参数，意思是不需要服务器返回，注意这个方法的返回值是void，而普通的exchangeDeclare方法的返回值是Exchange.DeclareOk，意思是在客户端声明了一个交换器之后，需要等待服务器的返回(服务器会返回Exchange.Declare-Ok这个AMQP命令)。针对"exchangeDeclareNoWait不需要服务器任何返回值"这一点，考虑这样一种情况，在声明完一个交换器之后(实际服务器还并未完成交换器的创建)，那么此时客户端紧接着使用这个交换器，必然会发生异常。如果没有特殊的缘由和应用场景，并不建议使用这个方法。

public void exchangeDeclareNoWait(String exchange, String type, boolean durable, 
        boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException

** 1.4 检测交换器是否存在的方法，**这个方法在实际应用过程中还是非常有用的，它主要用来检测相应的交换器是否存在。如果存在则正常返回:如果不存在则抛出异常:404  channel exception，同时Channel也会被关闭。  **        **

public DeclareOk exchangeDeclarePassive(String name)

**  1.5 删除交换器的方法，**方法中exchange表示交换器的名称，而ifUnused用来设置是否在交换器没有被使用的情况下删除。如果isUnused设置为true，则只有在此交换器没有被使用的情况下才会被删除:如果设置false，则无论如何这个交换器都要被删除，默认是false的。

DeleteOk exchangeDelete(String exchange, boolean ifUnused) throws IOException;
        void exchangeDeleteNoWait(String exchange, boolean ifUnused) throws IOException;
        DeleteOk exchangeDelete(String exchange) throws IOException; //默认为false

### 2. 声明查找删除队列 ###

2.1 这里只有两个方法：

Queue.DeclareOk queueDeclare() throws IOException;
    
     Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, 
        boolean autoDelete, Map<String, Object> arguments) throws IOException;

不带任何参数的queueDeclare方法默认创建一个由RabbitMQ命名的(类似这种amq.gen-LhQzlgv3DOv8PIDabOXA名称，这种队列也称之为匿名队列〉、排他的（true）、自动删除的(true)、非持久化（false）的队列,如下所示获取队列的名字：

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

> 方法的参数详细说明如下：
> 
>  *  queue:队列的名称
>  *  durable:设置是否持久化。为true则设置队列为持久化。持久化的队列会存盘，在服务器重启的时候可以保证不丢失相关信息
>  *  exclusive:设置是否排他。为t r u e则设置队列为排他的。如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。这里需要注意三点:排他队列是基于连接(Connection )可见的，同一个连接的不同信道( Channel )是可以同时访问同一连接创建的排他队列;"首次"是指如果一个连接己经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同:即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列适用于一个客户端同时发送和读取消息的应用场景
>  *  autoDelete:设置是否自动删除。为t r u e则设置队列为自动删除。自动删除的前提是:至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。不能把这个参数错误地理解为:"当连接到此队列的所有客户端断开时，这个队列自动删除"，因为生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列
>  *  argurnents:设置队列的其他一些参数，如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来声明一个队列，但是如果消费者在同一个信道上订阅了另一个队列，就无法再声明队列了。必须先取消订阅，然后将信道置为"传输"模式，才能声明队列

**2.2  这里也有一个nowait声明queue的方法**，返回值void和exchangeDeclareNoWait类似， 表示不需要服务端的任何返回。同样也需要注意，在调用完queueDeclareNoWait方法之后，紧接着使用声明的队列时有可能会发生异常情况。

public void queueDeclareNoWait(String queue, boolean durable, boolean exclusive, 
           boolean autoDelete, Map<String, Object> arguments)

**2.3  同样这里还有个queueDeclarePassive的方法**，也比较常用。这个方法用来检测相应的队列是否存在。如果存在则正常返回，如果不存在则抛出异常:404  channel  exception，同时Channel也会被关闭。方法定义如下:

public queueDeclarePassive(String queue)

**2.4 删除队列的方法**

>  *  其中queue表示队列的名称，
>  *  ifUnused以参考交换器（ifUnused用来设置是否在交换器没有被使用的情况下删除。如果isUnused设置为true，则只有在此交换器没有被使用的情况下才会被删除:如果设置false，则无论如何这个交换器都要被删除，默认是false的）
>  *   ifEmpty设置为t r u e表示在队列为空(队列里面没有任何消息堆积)的情况下才能够删除。

public com.rabbitmq.client.AMQP.Queue.DeleteOk queueDelete(String queue) throws IOException {
            return this.queueDelete(queue, false, false);
        }
    
        public com.rabbitmq.client.AMQP.Queue.DeleteOk queueDelete(String queue, boolean ifUnused, boolean ifEmpty) throws IOException {
            this.deleteRecordedQueue(queue);
            return this.delegate.queueDelete(queue, ifUnused, ifEmpty);
        }
    
        public void queueDeleteNoWait(String queue, boolean ifUnused, boolean ifEmpty) throws IOException {
            this.deleteRecordedQueue(queue);
            this.delegate.queueDeleteNoWait(queue, ifUnused, ifEmpty);
        }

### 3. 队列绑定交换器和解绑 queueBind ###

**     3.1 绑定队列和交换器    **

>        方法中涉及的参数详解：
> 
>  *  queue:队列名称
>  *  exchange:交换器的名称
>  *  routingKey:用来绑定队列和交换器的路由键
>  *  argument:定义绑定的一些参数

public Queue.BindOk queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments)
    
    public AMQP.Queue.BindOk queueBind(String queue, String exchange, String routingKey) 
    
    public void queueBindNoWait(String queue, String exchange, String routingKey, Map<String, Object> arguments)

**     3.2 将绑定的队列和交换器解绑**

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

### 4.交换器和交换器绑定exchangeBind ###

这里的destination表示和队列相连的exchange交换器的名字，source表示和消息发布者相连的交换器名字。

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

例子:

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

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 ,"exToExDemo".getBytes( )) ;

![70][]

## 三.发送消息 ##

方法用basicPublish如下：

void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body);
    
    void basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body);
    
    void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body);

上面的参数解析如下：

>  *  exchange:交换器的名称，指明消息需要发送到哪个交换器中。如果设置为空字符串，则消息会被发送到RabbitMQ默认的交换器中。
>  *  routingKey:路由键，交换器根据路由键将消息存储到相应的队列之中。
>  *  props:消息的基本属性集，其包含14个属性成员，分别有contentType、contentEncoding、headers(Map<String，Object>)、deliveryMode、priority、correlationld、replyTo、expiration、messageld、timestamp、type、userld、appld、clusterld。其中常用的几种都在上面的示例中进行了演示。
>  *  byte\[\] body:消息体(payload)，真正需要发送的消息。
>  *  mandatory和immediate的详细内容后面讲解

常用的消息发送例子：

//发送持久化信息
    String message = "hello world";
    channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
    
    这里的PERSISTENT_TEXT_PLAIN如下所示：
     public static final BasicProperties PERSISTENT_TEXT_PLAIN = new BasicProperties("text/plain",
    (String)null, (Map)null, 2, 0, (String)null, (String)null, (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);

5.1 上面这行代码发送了一条消息，这条消息的投递模式(deliveryMode)设置为2，即消息会被持久化(即存入磁盘)在服务器中。同时这条消息的优先级(priority)设置为1，content-type为"text/plain ",可以自己设消息的属性：

channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,
                    new AMQP.BasicProperties().builder()
                            .contentType("text/plain")
                            .deliveryMode(2)
                            .priority(1)
                            .userId("hidden")
                            .build(),
                    message.getBytes());

5.2 也可以发送带有headers的消息：

Map<String,Object> headers = new HashMap<>();
            headers.put("location","here");
            headers.put("time","today");
            channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,
                    new AMQP.BasicProperties().builder().headers(headers).build(),message.getBytes());

5.3 带有过期（expiration）信息的消息：

channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,
                    new AMQP.BasicProperties().builder().expiration("60000").build(),message.getBytes());

## 四.消费消息 ##

RabbitMQ的消费模式分两种:推(Push)模式和拉(Pull)模式。推模式采用Basic.Consume进行消费，而拉模式则是调用Basic.Get进行消费。

### **1. 推模式push** ###

**           **  可以通过持续订阅的方式来消费，使用到的类Consumer,DefaultConsumer，接受消息一般通过实现Consumer接口或者继承DefaultConsumer类，当调用Consumer相关的API方法时，不同的订阅需要制定不同的消费者标签（consumerTag）来区分彼此，在同一个channel中的消费者也需要通过唯一的消费标签用来做区分，关键消费代码如下：

boolean autoAck = false;
            channel.basicQos(64);//设置客户端最多接受违背ack的信息的个数
            Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("recieve message :"+new String(body));
                    try {
                        TimeUnit.SECONDS.sleep(1);  //休眠一秒在返回确定信息
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
           //这里的自动应答被关闭，通过手动确认收到消息后在做应答，这个可以有效的防止消息不必要的丢失。
            channel.basicConsume(QUEUE_NAME,autoAck,"myConsumerTag",consumer);

相关的消费方法：

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;

> 对应的参数说明如下所述：
> 
>  *  queue:队列的名称；
>  *  autoAck:设置是否自动确认。建议设成false，即不自动确认,默认不设置也是false；
>  *  consumerTag:消者标签，用来区分多个消费者，不设置默认“”；
>  *  noLocal:设置为true则表示不能将同一个Connection中生产者发送的消息传送给这个Connection中的消费者，默认是false；
>  *  exclusive:设置是否排他,默认是false；
>  *  arguments:设置消费者的其他参数,默认是null；
>  *  callback:设置消费者的回调函数。用来处理RabbitMQ推送过来的消息，比如DefaultConsumer，使用时需要客户端重写(override )其中的方法。

对于消费者需要消费消息一般都是重写handleDelivery方法

public interface Consumer {
        //这个方法会在其他方法调用前，返回消费者标签
        void handleConsumeOk(String consumerTag);
        //取消订阅的时候调用
        void handleCancelOk(String consumerTag);
    
        void handleCancel(String consumerTag) throws IOException;
        //重写这个方法会在channel和Connection关闭时候调用
        void handleShutdownSignal(String consumerTag, ShutdownSignalException sig);
    
        void handleRecoverOk(String consumerTag);
    
        void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException;
    }

我们可以通过channel.basicCancel方法来显式地取消一个消费者的订阅,channel.basicCancel(consumerTag);这行代码执行会首先触发handleConsumerOk方法，然后调用handleDelivery方法，最后才出发handleCancelOk方法。

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

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

### **2. 拉模式** ###

**     **      拉模式的消费方式，通过channel.basicGet方法可以单条地获取消息，其返回值是GetResponse，Channel类的basicGet方法没有其他重载方法，只有:

GetResponse basicGet(String queueName, boolean autoAck) throws IOException;

其中中queueName代表队列的名称，如果设置autoAck为false，那么同样需要调用channel.basicAck来确认消息己被成功接收。

例子：

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

> **注意点：**推模式和拉模式的对比。Basic.Consume将信道(Channel)直为接收模式，直到取消队列的订阅为止。在接收模式期间，RabbitMQ会不断地推送消息给消费者，当然推送消息的个数还是会受到Basic.Qos的限制.如果只想从队列获得单条消息而不是持续订阅，建议还是使用Basic . Get进行消费.但是不能将Basic.Get放在一个循环里来代替Basic.Consume，这样做会严重影响RabbitMQ的性能.如果要实现高吞吐量，消费者理应使用Basic.Consume方法。

![70 1][]

### 3.** 消费信息的确认** ###

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

**             **采用消息确认机制后，只要设置autoAck参数为false，消费者就有足够的时间处理消息(任务)，不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，因为RabbitMQ会一直等待持有消息直到消费者显式调用Basic.Ack命令为止。当autoAck参数置为false，对于RabbitMQ服务端而言，队列中的消息分成了两个部分:一部分是等待投递给消费者的消息:一部分是己经投递给消费者，但是还没有收到消费者确认信号的消息。如果RabbitMQ一直没有收到消费者的确认信号，并且消费此消息的消费者己经断开连接，则RabbitMQ会安排该消息重新进入队列，等待投递给下一个消费者，当然也有可能还是原来的那个消费者**。          **

boolean autoAck = false;  
            channel.basicQos(64);//设置客户端最多接受违背ack的信息的个数
            Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("recieve message :"+new String(body));
                    try {
                        TimeUnit.SECONDS.sleep(1);  //休眠一秒在返回确定信息
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                    //这里在消费信息后手动确认已经获取到信息了，然后Broker好删除消息
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            //这里设置的自动应答会false
            channel.basicConsume(QUEUE_NAME,autoAck,"myConsumerTag",consumer);

### ** 4. 消息的拒绝** ###

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

其中deliveryTag是消息的编号，是一个64位的长整型数值。

如果requeue参数设置为true，RabbitMQ会重新将这条消息存入队列，以便可以发送给下一个订阅的消费者;如果requeue参数设置为false，则RabbitMQ立即会把消息从队列中移除，而不会把它发送给新的消费者。

multiple参数设置为false则表示拒绝编号为deliveryTag的这一条消息，这时候basicNack和basicReject方法一样;multiple参数设置为true则表示拒绝deliveryTag编号之前所有未被当前消费者确认的消息。

注意要点:将channel.basicReject或者channel.basicNack中的requeue设直为f a l s e，可以启用"死信队列"的功能。死信队列可以通过检测被拒绝或者未送达的消息来追踪问题,后面讲。

### **5. 消息的重新获取** ###

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

这个channel.basicRecover方法用来请求RabbitMQ重新发送还未被确认的消息。如果requeue参数设置为true，则未被确认的消息会被重新加入到队列中，这样对于同一条消息来说，可能会被分配给与前不同的消费者。如果requeue参数设置为false，那么同一条消息会被分配给与之前相同的消费者。默认情况下，如果不设置requeue这个参数，相当于channel.basicRecover(true)，即requeue默认为true。

## 五.关闭连接 ##

channel.close();
    conn.close()

显式地关闭Channel个好习惯，但这不是必须的，在Connection关闭的时候，Channel也会自动关闭。

AMQP协议中的Connection和Channel采用同样的方式来管理网络失败、内部错误和显式地关闭连接。Connection和Channel所具备的生命周期如下所述:

>  *   Open:开启状态，代表当前对象可以使用。
>  *  Closing:正在关闭状态。当前对象被显式地通知调用关闭方法(shutdown)，这样就产生了一个关闭请求让其内部对象进行相应的操作，并等待这些关闭操作的完成。
>  *   Closed:已经关闭状态。当前对象已经接收到所有的内部对象已经完成关闭动作的通知，并且其也关闭了自身。Connection和Channel最终都是会成为losed的状态，不论是程序正常调用的关闭方法，或者是客户端的异常，再或者是发生了网络异常。

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

> addShutdownListener(ShutdownListenerlistener) ；
> 
> removeShutdownListener(ShutdownListner listener )

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

> getCloseReason方法可以让你知道对象关闭的原因;
> 
> isOpen方法检测对象当前是否处于开启状态;
> 
> close(int  closeCode ,  StringcloseMessage方法显式地通知当前对象执行关闭操作。

可以根据这个添加关闭监听器，检测失败的原因，异常情况，例如：

connection.addShutdownListener(new ShutdownListener() {
                @Override
                public void shutdownCompleted(ShutdownSignalException e) {
                    if (e.isHardError()){  //这个可以知道是connection异常还是channel异常
                        Connection conn = (Connection) e.getReference();
                        if (!e.isInitiatedByApplication()){
                            Method reason = e.getReason();
                            ...
                        }
                    }else {
                        Connection conn = (Connection) e.getReference();
                    }
    
                }
            });

[Link 1]: https://blog.csdn.net/weixin_40792878/article/details/82560158
[70]: /images/20220513/d0e3f39d55c14714b04f26d8ac0ec547.png
[70 1]: /images/20220513/0dfc68f80e4d4d2a91f0b0887db057fd.png