在rabbitMq中 我们可以通持久化数据解决rabbitMQ服务器异常 数据丢失的问题。
问题: 生成者将消息发送出去，有没有到达rabbitMq默认是不知道的
两种方式:
-
AMQP实现了事务机制 -
Confirm模式
事务机制
RabbitMQ提供了txSelect()、txCommit()和txRollback()三个方法对消息发送进行事务管理。
-
txSelect: 用于将通道channel开启事务模式 -
txCommit: 用于提交事务 -
txRollback: 用户进行事务回滚操作。
生产者
package cn.lovingliu.transition.producer;
import cn.lovingliu.common.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Author：LovingLiu
* @Description: 生产者
* @Date：Created in 2020-01-16
*/
public class TxProducer {
/** 队列名称 */
private static final String QUEUE_NAME = "test_queue_tx";
public static void main(String[] args) throws IOException, TimeoutException {
/** 1.获取连接 */
Connection newConnection = ConnectionUtil.getConnection();
/** 2.创建通道 */
Channel channel = newConnection.createChannel();
/** 3.创建队列声明 */
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
try{
/** 4.开启事务 */
channel.txSelect();
/**4.发送消息 */
String msg = "Hello TX Message";
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
int error = 1 / 0; // 生产者异常 未发送到rabbitMQ服务器
System.out.println("Send Message :"+ msg);
/** 5.提交事务 */
channel.txCommit();
}catch (Exception e){
/** 6.若发生异常 回滚事务 */
channel.txRollback();
System.out.println("Send Message RollBack");
}
channel.close();
newConnection.close();
}
}
消费者
package cn.lovingliu.transition.consumer;
import cn.lovingliu.common.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Author：LovingLiu
* @Description: 消费者
* @Date：Created in 2020-01-16
*/
public class TxCustomer {
/**
* 队列名称
*/
private static final String QUEUE_NAME = "test_queue_tx";
public static void main(String[] args) throws IOException, TimeoutException {
/** 1.获取连接 */
Connection newConnection = ConnectionUtil.getConnection();
/** 2.获取通道 */
final Channel channel = newConnection.createChannel();
/** 3.声明队列 开启自动应答 */
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
/** 3.监听队列 */
DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String msgString = new String(body, "UTF-8");
System.out.println("消费者获取消息:" + msgString);
}
};
channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
}
}
运行截图
先运行消费者，再运行生产者
Confirm模式
事务机制（性能消耗大）和confirm模式（消耗小）
RabbitMQ提供了一种低消耗的事务管理方式，将channel设置成confirm模式。
confirm模式的channel，通过该channel发出的消息会生成一个唯一的有序ID（从1开始），一旦消息成功发送到相应的队列之后，RabbitMQ服务端会发送给生产者一个确认标志，包含消息的ID，这样生产者就知道该消息已经发送成功了。
如果消息和队列是持久化的，那么当RabbitMQ服务器将消息成功写入磁盘之后，生产者才会收到确认消息。此外服务端也可以设置basic.ack的mutiple域，表明是否是批量确认的消息，即该序号之前的所有消息都已经收到了。
confirm的机制是异步的，生产者可以在等待的同时继续发送下一条消息，并且异步等待回调处理，如果消息成功发送，会返回ack消息供异步处理，如果消息发送失败发生异常，也会返回nack消息。
confirm的时间没有明确说明，并且同一个消息只会被confirm一次。
我们在生产者使用如下代码开启channel的confirm模式，已经开启事务机制的channel是不能开启confirm模式的。
channel.confirmSelect(); // 开启confirm模式
处理ack或者nack的方式有三种：
串行confirm
每发送一条消息就调用waitForConfirms()方法等待服务端confirm
//开启confirm模式
channel.confirmSelect();
String message = "Hello World";
//发送消息
channel.basicPublish(EXCHANGE_NAME,"",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
//判断是否回复
if(channel.waitForConfirms()){
System.out.println("Message send success.");
}
其中waitForConfirms可以换成带有时间参数的方法waitForConfirms(Long mills)指定等待响应时间
批量confirm：
每发送一批次消息就调用waitForConfirms()方法等待服务端confirm
//开启confirm模式
channel.confirmSelect();
for(int i =0;i<1000;i++){
String message = "Hello World";
//发送消息
channel.basicPublish(EXCHANGE_NAME,
"",
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes());
if(i%100==0){
//每发送100条判断一次是否回复
if(channel.waitForConfirms()){
System.out.println("Message send success.");
}
}
}
批量的方法从数量级上降低了confirm的性能消耗，提高了效率，但是有个致命的缺陷，一旦回复确认失败，当前确认批次的消息会全部重新发送，导致消息重复发送。所以批量的confirm虽然性能提高了，但是消息的重复率也提高了。
异步confirm
Channel 对象提供的 ConfirmListener() 回调方法只包含deliveryTag(当前Channel发出的消息序列号)，我们需要自己为每一个Channel维护一个unconfirm的序列号的集合，每push一条数据，集合元素加1，没回调一次handleAck方法，unconfirm集合就删掉相应的一条(multiple=false) 或者多条(multiple=true)记录，从程序运行效率上来看，这个unconfirm集合最好采用有序集合SortedSet存储结构。
使用监听方法，当服务端confirm了一条或多条消息后，调用回调方法
//声明一个用来记录消息唯一ID的有序集合SortedSet
final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
//开启confirm模式
channel.confirmSelect();
//异步监听方法 处理ack与nack方法
channel.addConfirmListener(new ConfirmListener() {
//处理ack multiple 是否批量 如果是批量 则将比该条小的所有数据都移除 否则只移除该条
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
if (multiple) {
confirmSet.headSet(deliveryTag + 1).clear();
} else {
confirmSet.remove(deliveryTag);
}
}
//处理nack 与ack相同
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("There is Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
if (multiple) {
confirmSet.headSet(deliveryTag + 1).clear();
} else {
confirmSet.remove(deliveryTag);
}
}
});
while (true) {
//获取消息confirm的唯一ID
long nextSeqNo = channel.getNextPublishSeqNo();
String message = "Hello World.";
//发送消息
channel.basicPublish(EXCHANGE_NAME,"",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
//将ID加入到有序集合中
confirmSet.add(nextSeqNo);
}
每一个comfirm的通道维护一个集合，每发送一条数据，集合增加一个元素，每异步响应一条ack或者nack的数据，集合删除一条。SortedSet是一个有序的集合，它的有序是值大小的有序，不是插入时间的有序。JDK中waitForConfirms()方法也是使用了SortedSet集合
网友评论