为了保证消息从队列可靠地到达消费者,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队列,表示消息可以重复使用

参考文章

评论可见,请评论后查看内容,谢谢!!!
 您阅读本篇文章共花了: