这篇文章主要和大家分享RabbitMQ Consumer端的知识点，主要包括Consumer的消费模式，消息是如何确认以及如何拒绝的，当消息拒绝之后，如何让消息重新进入队列。
推模式
RabbitMQ支持推和拉两种消费模式，推模式就是由Broker向Consumer端推送消息。
下面是示例代码，可以比较直观的看到使用方式。
String queueName = "";
boolean autoAck = false; // 关闭自动确认
String consumerTag = "myConsumerTag"; // 消费者标签，区分同一Channel中的不同消费者
channel.basicQos(64); // 消费端能保持的最大未确认消息数
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(Thread.currentThread().getName() + " -> consumer : " + consumerTag + " , receive message : " + new String(body));
// 手动确认消息，envelope.getDeliveryTag()表示消息的编号，确认哪条消息
// 第二参数：false确认单挑消息，true确认当前deliveryTag之前所有的消息
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
channel.basicConsume(queueName, autoAck, consumerTag, consumer);
上面的代码将autoAck设置为false，即手动确认，这样设置很重要，因为可以有效防止消息丢失的问题发生。
手动ack的逻辑是在DefaultConsumer的handleDelivery方法中，当然，在这个方法中还可以处理一些其它逻辑，比如拒绝消息，将消息重新入队等。 这个方法还有几个重载方法，使用方式一样，支持参数不同，但都很简单，小伙伴看一下接口就明白了。
推模式的优点就是消费消息的实时性比较高，但是需要评估消费能力，来控制未确认消息的缓存数量，防止客户端内存溢出。
拉模式
RabbitMQ Consumer客户端的拉模式则是主动从Borker上拉取消息，目前RabbitMQ提供的API只有get方法，而且一次只能拉取一条消息，使用方式也很简单，看下面的示例代码。
GetResponse response = channel.basicGet(QUEUE_NAME , false) ;
System.out.println(new String(response.getBody()));
channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
拉模式中也有autoAck的参数，最好也设置成false，采用手动ack的方式。
拉模式的优点就是自己可以根据当前的负载控制何时可以消费消息，缺点就是消费消息的实时性低，而且有可能造成消息在Broker上的堆积。
消息确认
为保证消息正确的到达Consumer，RabbitMQ提供了Consumer端的确认机制，如何使用在上面的代码示例中已有体现，就是设置autoAck参数控制。设置为false，Broker会等收到Consumer的ack之后，才会将消息打上删除标记，然后从磁盘或者内存中删除。设置为true，Broker发出消息之后，就会把消息移除，不管Consumer是否消费到了消息。
建议autoAck设为false的另外一点，RabbitMQ会一直保存未收到ack的消息，除非消费此消息的Consumer断开连接，RabbitMQ会把消息重新投入队列。RabbitMQ不会给未确认的消息设置TTL，判断消息是否需要重新投递给Consumer的唯一依据就是消费改消息的Consumer已经断开连接。
消息拒绝
Consumer有可能收到错误的消息或者无法处理的消息，而拒绝此消息，RabbitMQ也为此场景提供了相应的API。channel.basicReject 告诉Broker，拒绝消费此消息。来看代码示例。
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("consumer : " + consumerTag + " , receive message : " + new String(body));
// 拒绝消息，重新入队
channel.basicReject(envelope.getDeliveryTag(), true);
}
};
从使用上来看，和消息确认一样，只需将回调函数中的 channel.basicAck 换成 channel.basicReject ，第一个参数的含义和上面channel.basicAck一样，第二个参数表示是否重新入队，true：重新入队，false：直接从队列中移除，不再发送给Consumer。
注意，channel.basicReject 方法一次只能拒绝一条消息。
消息批量拒绝
如果Consumer端需要批量拒绝消息，可以使用channel.nack 方法，它的multiple参数可以控制拒绝单条消息还是批量拒绝消息。具体用法和参数含义参考下面示例代码。
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("consumer : " + consumerTag + " , receive message : " + new String(body));
// 第一个参数：当前消息的标识
// 第二个参数：拒绝之后，是否重新进入队列
// 第三个参数：是否批量拒绝，true：deliveryTag之前所有的消息都会被拒绝
channel.basicNack(envelope.getDeliveryTag(), false, false);
}
};
消息可重入队列
上面消息拒绝和消息批量拒绝的方法提供了参数，控制被拒绝的消息是否重新进入队列，RabbitMQ还提供了一个接口，用于直接将消息重新入队，channal.recover，它的参数默认就是true。使用的时候，需要注意，避免因为代码逻辑的问题造成消息循环入队。
下面的示例代码中重写了handleRecoverOk回调函数，当消息recover成功后，回调此函数。
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("consumer : " + consumerTag + " , receive message : " + new String(body));
// 让消息重新入队
channel.basicRecover(true);
}
// 监听消息recover是否成功
@Override
public void handleRecoverOk(String consumerTag) {
System.out.println(consumerTag);
}
};
好了，以上Consumer端关于消费模式，消息确认，消息拒绝，消息重新入队的内容了。
RabbitMQ系列文章会陆续更新，欢迎各位小伙伴关注后面的技术分享。
