美文网首页
springboot @EnableAsync 多线程

springboot @EnableAsync 多线程

作者: 你的大英雄 | 来源:发表于2019-12-11 16:05 被阅读0次

    在处理大数据或实时数据时,如果在主线程频繁创建大量对象,这些对象使用完后成为游离对象,不会立即被GC。当创建速度大于销毁速度时,可能导致内存持续上涨,最后内存溢出。
    可以开启多线程来处理,线程内的对象会在执行结束后尽快的销毁,均分内存累加的负担,保证内存占用的稳定性。

    springboot的多线程使用

    1.配置@EnableAsync

    package com.cdgs.data.config;
    import java.util.concurrent.Executor;
    import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
    import org.springframework.aop.interceptor.SimpleAsyncUncaughtExceptionHandler;
    import org.springframework.context.annotation.ComponentScan;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.AsyncConfigurer;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    @Configuration
    @EnableAsync(proxyTargetClass=true)//利用@EnableAsync注解开启异步任务支持
    @ComponentScan("com.cdgs.data.service") //必须加此注解扫描包
    public class CustomMultiThreadingConfig implements AsyncConfigurer{
    
        @Override
        public Executor getAsyncExecutor() {
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            taskExecutor.setCorePoolSize(10);
            taskExecutor.setMaxPoolSize(20);
            taskExecutor.setQueueCapacity(500);
          //当提交的任务个数大于QueueCapacity,就需要设置该参数,但spring提供的都不太满足业务场景,可以自定义一个,也可以注意不要超过QueueCapacity即可
          //taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());        
            taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
            taskExecutor.setAwaitTerminationSeconds(10);
            taskExecutor.setThreadNamePrefix("ES-IMOPRT-");
            taskExecutor.initialize();  
            return taskExecutor;
        }
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
            return new SimpleAsyncUncaughtExceptionHandler();
        }
    }
    

    2 配置@Async,修饰类时表示类里所有方法都是多线程异步执行

        @Async
        public Future<Integer> jdbcToElasticsearch(Pageable pageable) {
            //通过实现ApplicationContextAware得到applicationContext,进而获取spring管理的bean
            BaseinfoRepository repository = (BaseinfoRepository)SpringBeanUtil.getBean(BaseinfoRepository.class);
            List<ZrHisEnterpriseBaseinfo> list = new ArrayList<>() ;
    
            //实际查询数据库等业务代码...
    
            elasticsearchOperation.bulkSave(list);
            return new AsyncResult<>(list.size());
        }
    

    3 调用,注意:调用方法不能和异步方法在同一类里

       //oralce批量导入es
        @GetMapping("/baseinfo")
        public String importBaseinfo(@PageableDefault(size=1000)Pageable pageable) {
            long count = autoImportService.count();
            int size = pageable.getPageSize();
            long loops = count%size>0?count/size+1:count/size;
            ArrayList<Future<Integer>> futureList = new ArrayList<>();
            for (int i = 0; i < loops; i++) {
                //异步执行任务,返回参数使用Future封装接收
                Future<Integer> future = autoImportService.jdbcToElasticsearch(new PageRequest(pageable.getPageNumber() + i, size));
                futureList.add(future);
            }
            importCount = checkTaskDone(futureList);
            System.out.println(importCount);
        }
    
        public static long checkTaskDone(ArrayList<Future<Integer>> futureList) {
            //判断异步调用的方法是否全都执行完了
            while(true) {
                int doneSize = 0;
                for ( Future<Integer> future  : futureList) {
                    //该异步方法是否执行完成
                    if(future.isDone()) {
                        doneSize++;
                    }
                }
                //如果异步方法全部执行完,跳出循环
                if(doneSize == futureList.size()) {
                    break;
                }
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }//每隔2秒判断一次
            }
            
            long importCount = 0;
            for ( Future<Integer> future  : futureList) {
                try {
                    importCount += future.get();
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }
            return importCount;
        }
    

    4 调用该方法,观察内存能够稳定在一定范围

    文章来源 :半缘_1ec0(https://www.jianshu.com/u/9ce90913f230)

    相关文章

      网友评论

          本文标题:springboot @EnableAsync 多线程

          本文链接:https://www.haomeiwen.com/subject/brzogctx.html