消息确认机制，confirm异步

一：介绍

1.异步模式介绍

　　Channel对象提供ConfirmListener()回调方法只包含deliverTag（当前Channel发出的序列号），我们需要自己为每一个Channel维护一个unconfirm的消息序列集合，没publish一条数据，集合就加1，每回调一次handleAck方法，unconfirm集合删掉相应的一条（multiple=false）或者多条（multiple=true）记录。从程序运行效率上看，这个unconfirm集合最好采用有序集合SortedSet存储结构。

二：程序

1.生产者

package com.mq.AsynConfirm;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;

public class Send {
    private static final String QUEUE_NAME="test_queue_confirm_asyn";
    public static void main(String[] args)throws Exception{
        Connection connection= ConnectionUtil.getConnection();
        Channel channel=connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //生产者调用confirmSelect将channel设置为nconfirm模式
        channel.confirmSelect();
        final SortedSet<Long> confirmSet= Collections.synchronizedSortedSet(new TreeSet<Long>());
        channel.addConfirmListener(new ConfirmListener() {
            //没有问题
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if (multiple){
                    System.out.println("handleAck multiple");
                    confirmSet.headSet(deliveryTag+1).clear();
                }else{
                    System.out.println("handleAck false");
                    confirmSet.remove(deliveryTag);
                }
            }
            //有问题
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple){
                    System.out.println("handleNack multiple");
                    confirmSet.headSet(deliveryTag+1).clear();
                }else{
                    System.out.println("handleNack false");
                    confirmSet.remove(deliveryTag);
                }
            }
        });
        String msg="success";
        while (true){
            long seqNo=channel.getNextPublishSeqNo();
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
            confirmSet.add(seqNo);
        }

    }
}

2.消费者

package com.mq.AsynConfirm;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class Receive {
    private static final String QUEUE_NAME="test_queue_confirm_asyn";
    public static void main(String[] args)throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicConsume(QUEUE_NAME,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body,"utf-8"));
            }
        });
    }
}

3.现象

　　Send：