为了保证消息从队列可靠地到达消费者，RabbitMQ提供消息确认机制(message acknowledgment)。消费者在声明队列时，可以指定noAck参数，当noAck=false时，RabbitMQ会等待消费者显式发回ack信号后才从内存(和磁盘，如果是持久化消息的话)中移去消息。否则，RabbitMQ会在队列中消息被消费后立即删除它。
在Consumer中Confirm模式中分为手动确认和自动确认。
手动确认主要并使用以下方法：
basicAck(): 用于肯定确认，multiple参数用于多个消息确认。
basicRecover()：是路由不成功的消息可以使用recovery重新发送到队列中。
basicReject()：是接收端告诉服务器这个消息我拒绝接收,不处理,可以设置是否放回到队列中还是丢掉，而且只能一次拒绝一个消息,官网中有明确说明不能批量拒绝消息，为解决批量拒绝消息才有了basicNack。
basicNack()：可以一次拒绝N条消息，客户端可以设置basicNack方法的multiple参数为true。
消费者确认模式：自动确认
package com.it.rabbitmq.confirm;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receive {
public static void main(String[] args) {
//创建链接工厂对象
ConnectionFactory factory=new ConnectionFactory();
factory.setUsername("root");
factory.setPassword("123456");
factory.setHost("192.168.174.129");
factory.setPort(5672);
Connection connection=null;//定义链接对象
Channel channel=null;//定义通道对象
try {
connection=factory.newConnection();//实例化链接对象
channel=connection.createChannel();//实例化通道对象
//指定Exchange的类型
//参数1为 交换机名称
//参数2为交换机类型取值为 direct、queue、topic、headers
//参数3 为是否为持久化消息 true表示持久化消息 false表示非持久化
channel.queueDeclare("confirmQueue", true, false, false, null);
channel.exchangeDeclare("directConfirmExchange", "direct", true);
channel.queueBind("confirmQueue","directConfirmExchange","confirmRoutingKey");
/**
* 接收消息
* 参数2为消息的确认机制，true表示自动消息确认，确认以后消息会从队列中被移除
* 参数为false表示手动确认，必须显示的调用一些方法
* 注意:
* 1.如果我们只是接收了消息但是还没有来得及处理，当前应用或数据库方面出现问题
* 但当前消息已经自动确认并删除，这就会丢失消息
*/
channel.basicConsume("confirmQueue",true, "",new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body) throws IOException {
//获取消息数据
String bodyStr = new String(body);
System.out.println("消费者处理了消息："+bodyStr);
}
});
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}
进入控制台查看队列，消息已经被全部取出
消费者确认模式：手动确认
好处是不会因为程序有问题，导致自动确认消息后，造成部分数据丢失。
package com.it.rabbitmq.exchange.confirm.addConfirmListener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Send {
public static void main(String[] args) {
//创建连接工厂
ConnectionFactory factory=new ConnectionFactory();
//配置rabbitMQ的连接信息
factory.setHost("192.168.174.129");
factory.setPort(5672);
factory.setUsername("root");
factory.setPassword("123456");
//定义连接
Connection connection=null;
//定义通道
Channel channel=null;
try {
connection=factory.newConnection();
channel=connection.createChannel();
channel.queueDeclare("confirmQueue",true,false,false,null);
channel.exchangeDeclare("directConfirmExchange","direct",true);
channel.queueBind("confirmQueue","directConfirmExchange","confirmRoutingKey");
String message="普通发送者确认模式的测试消息！";
//启动发送者确认模式
channel.confirmSelect();
//异步消息确认监听器，需要在消息发送前启动
channel.addConfirmListener(new ConfirmListener() {
//消息确认以后的回调方法
//参数1：被确认的消息编号，从1开始自动递增，用于标记当前是第几个消息
//参数2：当前消息是否同时确定了多个
//参数2为true,表示小于等于当前消息编号的所有消息全部被确认，为false，表示只确定了当前消息
public void handleAck(long l, boolean b) throws IOException {
System.out.println("消息被确认了--消息编号"+l+" 是否确认了多条："+b);
}
//消息没有确认的回调方法
//如果这个方法被执行，表示当前消息没有被确认，需要进行消息的补发
//参数1：没有确认的消息编号，从1开始自动递增，用于标记当前是第几个消息
//参数2：当前消息是否同时没有确定了多个
//如果参数2位true,表示小于当前编号的所有消息都没有发送成功，需要进行消息的补发
//为false,表示当前编号消息没有发送成功，需要进行补发
public void handleNack(long l, boolean b) throws IOException {
System.out.println("消息没有被确认了--消息编号"+l+" 是否没有确认多条："+b);
}
});
for (int i=0;i<100;i++){
channel.basicPublish("directConfirmExchange","confirmRoutingKey",null,message.getBytes("utf-8"));
}
System.out.println("消息发送成功");
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} finally {
if (channel!=null){
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
if (connection!=null){
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
运行代码
成功的全部取出
如果不加basicAck方法会导致消息确认会全部转入上图的unacked队列，表示消息可以重复使用
参考文章
您阅读本篇文章共花了： 大家都在找：
Java：java官网
java-rabbitmq：java17
rabbitmq：rabbitmq是什么
发表评论