数据保存先入队列，然后批量处理入库

╰+哭是因爲堅強的太久メ 2022-01-11 06:13 283阅读 0赞

# **更多多线程知识访问 [www.itkc8.com][]** #

package com.example.demo.service;
    
    import com.example.demo.domain.MyStudent;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Service;
    import org.springframework.util.CollectionUtils;
    
    import javax.annotation.PostConstruct;
    import java.util.ArrayList;
    import java.util.Date;
    import java.util.List;
    import java.util.concurrent.*;
    
    /**
     * @author hux
     * @title: BatchExecutorDataJob
     * @projectName springboot-scala
     * @description: TODO
     * @date 2019/6/19 17:04
     * @Version 1.0
     */
    @Service
    @Slf4j
    public class BatchExecutorDataJob {
    
        /**
         * 定义一个容量为10000的阻塞队列，BlockingQueue线程安全可以多个生产者同时put
         */
        private BlockingQueue<MyStudent> dataQueue = new LinkedBlockingQueue<>(10000);
        ExecutorService singleThreadPool = new ThreadPoolExecutor(3, 5, 60, TimeUnit.MICROSECONDS, new LinkedBlockingQueue<>(1024));
    
        private List<MyStudent> list = new ArrayList<>();
    
        //put任务的方法，供生产者调用
        public void recordJob(MyStudent job) {
            System.out.println(job);
            try {
                dataQueue.put(job);
            } catch (InterruptedException e) {
                log.info("批量更新Job入队列异常");
                Thread.currentThread().interrupt();
            }
        }
    
        /**
         * 初始化即调用
         */
        @PostConstruct
        private void init() {
            log.info("启动批量更新线程，启动时间{}", new Date(System.currentTimeMillis()));
            singleThreadPool.execute(() -> {
                while (Boolean.TRUE) {
                    MyStudent stu = null;
                    boolean pollTimeOut = false;
                    try {
                        // poll时设置超时时间为2秒
                        stu = dataQueue.poll(6, TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
                        log.info("批量更新Job异常");
                        Thread.currentThread().interrupt();
                    }
    
                    if (null != stu) {
                        // poll到任务添加到List中
                        list.add(stu);
                    } else {
                        // poll超时，设置超时标志位
                        pollTimeOut = true;
                    }
    
                    // 如果任务List等于5000或poll超时且List中还有任务就批量更新
                    if (list.size() == 10 || (pollTimeOut && !CollectionUtils.isEmpty(list))){
                        //jobService.batchUpdateByPrimaryKeySelective(list);
                        log.info("Job任务批量更新{}条任务,耗时{}毫秒", list.size());
                        list.clear();
                    }
                }
            });
            singleThreadPool.shutdown();
            /*Thread thread = new Thread(() -> {
                log.info("启动批量更新守护线程，启动时间{}", new Date(System.currentTimeMillis()));
                while (Boolean.TRUE) {
                    MyStudent stu = null;
                    boolean pollTimeOut = false;
                    try {
                        // poll时设置超时时间为2秒
                        stu = dataQueue.poll(2, TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
                        log.info("批量更新Job异常");
                        Thread.currentThread().interrupt();
                    }
    
                    if (null != stu) {
                        // poll到任务添加到List中
                        list.add(stu);
                    } else {
                        // poll超时，设置超时标志位
                        pollTimeOut = true;
                    }
    
                    // 如果任务List等于5000或poll超时且List中还有任务就批量更新
                    if (list.size() == 300 || (pollTimeOut && !CollectionUtils.isEmpty(list))){
                        //jobService.batchUpdateByPrimaryKeySelective(list);
                        log.info("Job任务批量更新{}条任务,耗时{}毫秒", list.size());
                        list.clear();
                    }
                }
            });
            thread.setName("job-batchUpdate-deamon");
            // 设置启动的线程为守护线程
            thread.setDaemon(true);
            thread.start();*/
        }
    
    
    }

[www.itkc8.com]: http://www.itkc8.com