RabbitMQ 消息确认机制之事务机制

在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集合