[大数据]RabbitMq 异步发布确认

public class ConfirmMessage {

    public static final Integer MESSAGE_COUNT=1000;

    public static void main(String[] args) throws Exception {
        publishMessageAsync();
    }

    /**
     * 异步发步确认
     * @throws Exception
     */
    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        // 生成消息队列
        channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null);
        // 发布确认
        channel.confirmSelect();
        /**
         * 线程安全有序的一个哈希表适用于高并发的情况
         * 1.轻松的将序号与消息进行关联 key value储存
         * 2.轻松批量删除条目 只要给到key
         * 3.支持高并发（多线程）
         */
        ConcurrentSkipListMap<Long,String> concurrentSkipListMap=new ConcurrentSkipListMap();

        // 消息确认成功回调
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            ConcurrentNavigableMap<Long, String> confirmed
                    = concurrentSkipListMap.headMap(deliveryTag);
            // 是否批量
            if(multiple){
                // 删除掉已近确认的消息，剩下的就是未确认的消息
                confirmed.clear();
            }else {
                // 删除掉已经确认的消息，剩下的就是未确认的消息
                confirmed.remove(deliveryTag);
            }
            System.out.println("确认的消息: " + deliveryTag);
        };
        // 消息确认失败回调
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("未确认的消息: " + deliveryTag);
        };
        // 消息的监听器，异步执行的
        channel.addConfirmListener(ackCallback,nackCallback);
        for (int i=0;i<MESSAGE_COUNT;i++){
            String message=i+"消息";
            // 记录所有要发送的消息
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(),message);
            // 发送消息
            channel.basicPublish("",Producer.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

        }
    }
}