springboot并发编程CountDownLatch配合异步线程池使用

素颜马尾好姑娘i 2022-12-31 11:25 256阅读 0赞

### 1. 首先在项目配置异步线程池，如下： ###

@EnableAsync // 开启异步任务
    @Configuration
    public class TaskPoolConfig { 
        @Bean("taskExecutor") // 线程池名称
        public Executor taskExecutor() { 
            // 使用Spring封装的异步线程池
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(10);               // 初始化线程数
            executor.setMaxPoolSize(20);                // 最大线程数
            executor.setQueueCapacity(200);             // 缓冲队列
            executor.setKeepAliveSeconds(60);           // 允许空闲时间/秒
            executor.setThreadNamePrefix("taskExecutor-");// 线程池名前缀-方便日志查找
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            executor.setWaitForTasksToCompleteOnShutdown(true);
            executor.setAwaitTerminationSeconds(60);
            executor.initialize();   // 初始化
            return executor;
        }

上面我们通过使用ThreadPoolTaskExecutor创建了一个线程池，同时设置了以下这些参数：

*  核心线程数10：线程池创建时候初始化的线程数
 *  最大线程数20：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
 *  缓冲队列200：用来缓冲执行任务的队列  
    允许线程的空闲时间60秒：当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
 *  线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池
 *  线程池对拒绝任务的处理策略：这里采用了CallerRunsPolicy策略，当线程池没有处理能力的时候，该策略会直接在 execute 方法的调用线程中运行被拒绝的任务；如果执行程序已关闭，则会丢弃该任务  
    **说明**：setWaitForTasksToCompleteOnShutdown（true）该方法就是这里的关键，用来设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean，这样这些异步任务的销毁就会先于Redis线程池的销毁。同时，这里还设置了setAwaitTerminationSeconds(60)，该方法用来设置线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁，以确保应用最后能够被关闭，而不是阻塞住。

### 2.线程池的使用 ###

使用多线程，往往是创建Thread，或者是实现runnable接口，用到线程池的时候还需要创建Executors，spring中有十分优秀的支持，就是注解@EnableAsync就可以使用多线程，@Async加在线程任务的方法上（需要异步执行的任务），定义一个线程任务，通过spring提供的ThreadPoolTaskExecutor就可以使用线程池。

线程池的使用在Spring中非常简单，只要设置两个注解就可以了

（1）@EnableAsync // 开启异步任务

（2）@Async(“taskExecutor”) // 申明为异步方法，指定线程池名称

注： @Async所修饰的函数不要定义为static类型，这样异步调用不会生效

@Slf4j
    @Component
    public class Task { 
     
        public static Random random = new Random();
     
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
     
        @Async("taskExecutor")
        public void doTaskOne() throws Exception { 
            log.info("开始做任务一");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            log.info(stringRedisTemplate.randomKey());
            log.info("完成任务一，耗时：" + (end - start) + "毫秒");
        }
     
        @Async("taskExecutor")
        public void doTaskTwo() throws Exception { 
            log.info("开始做任务二");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            log.info("完成任务二，耗时：" + (end - start) + "毫秒");
        }
     
        @Async("taskExecutor")
        public void doTaskThree() throws Exception { 
            log.info("开始做任务三");
            long start = System.currentTimeMillis();
            Thread.sleep(random.nextInt(10000));
            long end = System.currentTimeMillis();
            log.info("完成任务三，耗时：" + (end - start) + "毫秒");
        }
     
    }

简单测试如下：

@RunWith(SpringJUnit4ClassRunner.class)
    @SpringBootTest
    public class TaskTest { 
        @Autowired
        private Task task;
        @Test
        public void test() throws Exception { 
            task.doTaskOne();
            task.doTaskTwo();
            task.doTaskThree();
            Thread.currentThread().join();
        }
    }

测试结果如下：

2020-04-16 15:23:13.834  INFO 1828 --- [ taskExecutor-1] demo.spring.tasks.Task                   : 开始做任务一
    2020-04-16 15:23:13.834  INFO 1828 --- [ taskExecutor-2] demo.spring.tasks.Task                   : 开始做任务二
    2020-04-16 15:23:13.835  INFO 1828 --- [ taskExecutor-3] demo.spring.tasks.Task                   : 开始做任务三
    2020-04-16 15:23:17.539  INFO 1828 --- [ taskExecutor-2] demo.spring.tasks.Task                   : 完成任务二，耗时：3704毫秒
    2020-04-16 15:23:18.380  INFO 1828 --- [   scheduling-1] demo.spring.tasks.ScheduledTasks         : ScheduledTasks1 - The time is now 15:23:18
    2020-04-16 15:23:18.381  INFO 1828 --- [   scheduling-1] demo.spring.tasks.ScheduledTasks         : ScheduledTasks2 - The time is now 15:23:18
    2020-04-16 15:23:19.475  INFO 1828 --- [ taskExecutor-3] demo.spring.tasks.Task                   : 完成任务三，耗时：5640毫秒

### 3.配合使用countdownlatch ###

多用背景：多使用于一个方法内部需要拼接多个方法返回参数最终返回。如：远程调用其他系统接口，担心效率问题。可以采取多线程计数，最后拼装参数返回

未改造前代码如下：

AsyncTask

@Component
    public class AsyncTask { 
    
     
        /** * 异步调用第三方接口查询 */
        @Async
        public void queryTask(Entity entity){ 
           Data data = doPost();
           entity.setData(data);
        }
       
    }

service

public class Service { 
        
        @Autowired
        private AsyncTask asyncTask;
        @Autowired
        private EntityMapper mapper;
        
        public List<Entity> queryData(){ 
            
            List<Entity> list = mapper.selectAll();
            for(Entity entity:list){ 
                asyncTask.queryTask(entity);
            }
            
           return list;
        }
    }

这样就实现了异步请求接口，效率上提升了很多。但是，由于异步调用的原因，数据还没填充完就会返回，这显然不是我们想要的效果。我们必须等待AsyncTask中的所有线程结束后，再返回当前调用线程任务的方法

思路：在Service层的方法中实例化CountDownLatch并且制定线程个数，线程个数就是从本地数据库查询的list的长度，并且传入线程任务中，每个线程执行完毕就调用countDown()方法。最后在Service层中调用await()方法。这样在线程计数为零之前，Service的线程就会一直等待

service

public class Service { 
        
        @Autowired
        private AsyncTask asyncTask;
        @Autowired
        private EntityMapper mapper;
        
        public List<Entity> queryData(){ 
            
            List<Entity> list = mapper.selectAll();
            CountDownLatch latch = new CountDownLatch(list.size());
            for(Entity entity:list){ 
                asyncTask.queryTask(entity,latch);
            }
           latch.await();
           return list;
        }
    }

AsyncTask

@Component
    public class AsyncTask { 
     
     
        /** * 异步调用第三方接口查询 */
        @Async（'taskExecutor'）
        public void queryTask(Entity entity,CountDownLatch latch){ 
           Data data = doPost();
           entity.setData(data);
           latch.countDown();
        }
       
    }