JAVA如何处理各种批量数据入库(BlockingQueue)

灰太狼 2024-02-18 11:44 68阅读 0赞

#### JAVA如何处理各种批量数据入库(BlockingQueue) ####

这里我推荐大家使用`BlockingQueue`，该队列为阻塞队列

将数据先写入队列中，然后开启多个消费线程慢慢消费入库。从队列中消费数据有两种方式：

单条消费
    
     批量消费

##### 一、案例：人脸识别设备—开门记录（批量消费） #####

###### 1.创建队列 ######

@Component
    public class RequestQueue<T> {
        
       
    	/**
    	* 获取开门记录队列
    	*/
        BlockingQueue<JSONObject> blockingQueue = new ArrayBlockingQueue<>(1000000,true);
        
         
        /**
         * 获取请求队列数据
         */
        public BlockingQueue<JSONObject> getBlockingQueue(){
        
            return blockingQueue;
        }
       
        
        
        /**
         * 保存数据进入队列
         */
        public void putJsonDataQueue(JSONObject jsonObject){
        
            try {
        
                getBlockingQueue().put(jsonObject);
    
            }catch (Exception e){
        
                e.printStackTrace();
            }
    
        }
        
    }

###### 2.创建消费队列任务 ######

@Component
    public class DoorOpeningRecordTask extends Thread{
        
        @Autowired
        private RequestQueue requestQueue;
    
        @Autowired
        private DoorOpeningRecordService doorOpeningRecordService;
    
        @Override
        public void run(){
        
            while(true){
        
                try {
        
                    //取出信息列表
                    List<JSONObject> list = new ArrayList<>();
                    //参数列表： 获取队列  数据  批量处理一百条  500毫秒
                    Queues.drain(requestQueue.getBlockingQueue(), list, 100, 500, TimeUnit.MILLISECONDS);
                    //对数据进行保存
                    doorOpeningRecordService.save(list);
    
                }catch (Exception e){
        
                    e.printStackTrace();
                }
    
            }
    
        }
    
    
    }

###### 3.监听队列信息 ######

@Component
    public class QueueListener {
        
        
        @Autowired
        private DoorOpeningRecordTask doorOpeningRecordTask;
    
        /**
         * 初始化时启动监听请求队列
         */
        @PostConstruct
        public void init() {
        
            doorOpeningRecordTask.start();
        }
    
        /**
         * 销毁容器时停止监听任务
         */
        @PreDestroy
        public void destory() {
        
           
            doorOpeningRecordTask.interrupt();
        }
    
    }