• rabbitMQ生产者的异步的异步批量发布确认demo


    
    
        //批量发布异步确认
        public static void publishSync() throws Exception {
             //创建链接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
    
            //设置链接
            connectionFactory.setHost("192.168.43.37");
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
    
            //链接工厂创建链接
            Connection connection = connectionFactory.newConnection();
    
            //获取信道
            Channel channel = connection.createChannel();
    
            //信道声明
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, true, false, false, null);
    
            //开启发布确认
            channel.confirmSelect();
    
    
            /**
             * 线程安全有序的一个哈希表 适用于高并发的情况下
             *
             * 1.轻松的将需要与消息进行关联
             * 2.轻松的批量删除条数
             * 3.支持高并发
             */
            ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>();
    
    
    
            //确认监听器
            ConfirmCallback confirmCallback = ( deliveryTag,multiple)->{
                System.out.println("消息确认监听器成功确认:" +  deliveryTag);
                //删除发送过的消息
                if(multiple){
                    //批量的删除确认的消息 剩下的是未确认的
                    ConcurrentNavigableMap<Long,String> confirmed =
                            outstandingConfirms.headMap(deliveryTag);
                    confirmed.clear();
    
                }else {
                    outstandingConfirms.remove(deliveryTag);
                }
    
            };
    
            ConfirmCallback nconfirmCallback= ( deliveryTag,multiple)->{
    
                //未确认的消息
                String message = outstandingConfirms.get(deliveryTag);
                System.out.println("未确认的消息:"+deliveryTag +"-"+message);
    
            };
    
            channel.addConfirmListener(confirmCallback,nconfirmCallback);
    
    
            //开始时间
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String mes = i + "";
                channel.basicPublish("",queueName,null,mes.getBytes("UTF-8"));
                outstandingConfirms.put(channel.getNextPublishSeqNo(),mes);
    
                //记录所有发布的消息
            }
    
            long end = System.currentTimeMillis();
            System.out.println("发布"+MESSAGE_COUNT+"条用时共计:" + (end-begin) );
    
        }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
  • 相关阅读:
    【NLP的python库(02/4) 】:Spacy
    C++:容量适配器(栈、队列、优先级队列)
    详解Python安装requests库的实例代码
    梳理RWKV 4,5(Eagle),6(Finch)架构的区别以及个人理解和建议
    Codeforces Round 895 (Div. 3) A-F
    02 【axios fetch 跨域】
    Linux输出文件夹下所有文件的完整路径shell脚本
    表单校验,日期比较
    C++类和对象-多态->案例1计算器类、案例2制作饮品、案例3电脑组装需求分析和电脑组装具体实现
    ActiveMQ安装
  • 原文地址:https://blog.csdn.net/qq_17040587/article/details/134420833