java实现分布式系统mysql的批量更新，数据先放队列中然后批量入库

偏执的太偏执、 2022-01-11 05:03 1154阅读 0赞

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

背景  
项目中开发一个批量新增，更新的需求，由于之前的数据是一条一条入库的，每天入库量在十万左右，后来增加需求每天的入库量上升到百万时，入库遭遇瓶颈，mq积压严重。后来发现这样需要频繁的与Mysql交互，且需要等待写库结果返回，效率堪忧拖慢其他模块，就有了批量新增和批量更新的需求。

优化  
第一步：用线程池来更新，将更新代码提交到线程池中，由线程池调度入库   
缺点：没有解决与数据库频繁交互的问题。   
第二步：执行模块不管更新结果，只需将更新任务放入一个队列中然后直接返回；用Spring的定时任务注解@Scheduled，指定一个方法，隔一段时间调用一次入库方法；入库的逻辑是，获取队列当前任务个数cnt，循环poll任务然后添加到一个List中，poll够cnt个之后，通过批量更新方法将List更新到数据库。   
缺点：定时执行无法控制队列大小，可能一次会取出很多条任务，也可能会把队列撑得过大。   
第三步：使用阻塞队列放更新任务，用守护线程poll的队列中的任务，当条数等于300条(此值根据实际情况，我们大佬建议300~500)，则批量更新一次，在poll时设置超时时间为2秒，当超过2秒还是没有取到任务，则也批量把已经渠道的任务更新一次。

@Service  
public class BatchExecutorDataJob \{  
   
    final Logger logger = LoggerFactory.getLogger(BatchExecutorDataJob.class);  
      
    @Resource  
    private JobService jobService ;  
   
    //定义一个容量为10000的阻塞队列，BlockingQueue线程安全可以多个生产者同时put  
    private BlockingQueue<Job> dataQueue = new LinkedBlockingQueue<>(10000);  
   
    //  
    private List<job> list = new ArrayList<job>();  
   
    //put任务的方法，供生产者调用  
    public void recordJob(Job job) \{  
        try \{  
            dataQueue.put(job);  
        \} catch (InterruptedException e) \{  
            LOGGER.info("批量更新Job入队列异常");  
            Thread.currentThread().interrupt();  
        \}  
    \}  
   
    //初始化即调用  
    @PostConstruct  
    private void init() \{  
        Thread thread = new Thread(() -> \{  
            LOGGER.info("启动批量更新守护线程，启动时间\{\}", new Date(System.currentTimeMillis()));  
            while (Boolean.TRUE) \{  
                Job poll = null;  
                boolean pollTimeOut = false;  
                long startTime;  
                long endTime;  
                try \{  
                    // poll时设置超时时间为2秒  
                    poll = dataQueue.poll(2, TimeUnit.SECONDS);  
                \} catch (InterruptedException e) \{  
                    LOGGER.info("批量更新Job异常");  
                    Thread.currentThread().interrupt();  
                \}  
   
                if (null != poll) \{  
                    // poll到任务添加到List中  
                    list.add(poll);  
                \} else \{  
                    // poll超时，设置超时标志位  
                    pollTimeOut = true;  
                \}  
   
                // 如果任务List等于5000或poll超时且List中还有任务就批量更新  
                if (list.size() == 300||   
                    (pollTimeOut && !CollectionUtils.isEmpty(list)))\{  
                    startTime = System.currentTimeMillis();  
                    jobService.batchUpdateByPrimaryKeySelective(list);                
                    LOGGER.info("Job任务批量更新\{\}条任务,耗时\{\}毫秒", list.size(),       
                    System.currentTimeMillis()-startTime);  
                    list.clear();  
                \}  
            \}  
        \});  
   
        thread.setName("job-batchUpdate-deamon");  
        // 设置启动的线程为守护线程  
        thread.setDaemon(true);  
        thread.start();  
    \}  
\}  
生产者方法就不说了，主要说消费者方法：

1、@PostConstruct作用是在Bean初始化之前执行消费者方法  
2、poll = dataQueue.poll(2, TimeUnit.SECONDS)，使用阻塞队列的poll方法，设置poll超时时间，当超过时间返回一个null值  
3、list.size() == 300，如果List中值等于300了就执行批量更新  
4、pollTimeOut && !CollectionUtils.isEmpty(list)，如果poll超时了，说明当前生产者暂时没有生产任务或不再生产任务，把List       中剩余的任务批量更新  
5、thread.setDaemon(true)，设置线程为守护线程，直到jvm停了才停止

最后提交代码，查看log日志，使用批量入库耗时比较长。查找各种原因后jdbcUrl 还需要添加配置

rewriteBatchedStatements=true  
添加上后重新部署项目，入库耗时瞬间在1000ms以下了。  
\---------------------   
  
原文：https://blog.csdn.net/qq\_36245532/article/details/88190511

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