工作中线程池使用不当的问题记录（get是阻塞式的）

文章目录

Task任务
有问题的版本
修改后的版本一
修改后的版本二

最近在看处理一个Bug的时候，发现了自己之前写的代码的一些问题，还好后来自己提出了优化，修改掉了。特作此记录。

项目背景：

我需要根据集合中的数据记录，将每一条数据封装为对应的报文，然后调用发送给外部接口，最后解析对应的返回报文，最终将数据落库。

由于数据库的数据记录比较多，所以这个操作是一个十分耗时的操作，所以我第一反应就是使用多线程进行处理，最终我决定使用线程池进行该任务的操作。

下列代码我根据真实的项目代码抽象而来。

Task任务

1、定义一个任务类，实现Callable接口，用于接收返回值

2、每一个任务内部定义了一个计数器，用于任务完成的标记

3、内部随机获取一个五以内的值，用于后期的线程睡眠，代表业务执行时间

4、返回对应的执行结果，此处用休眠时间代替

public class Task implements Callable<Integer> {private CountDownLatch num;public Task(CountDownLatch num) {this.num = num;}@Overridepublic Integer call() throws Exception {int i = new Random().nextInt(5);TimeUnit.SECONDS.sleep(i);System.out.println("业务执行时间" + i);num.countDown();return i;}
}

有问题的版本

1、记录程序执行开始的时间

2、自定义一个线程池，最大线程数和核心线程数都使用CPU密集型进行大小的设置

3、设置待处理的集合中数据的条数，并实例化一个对应大小的计数器

4、使用for循环，使用线程池依次执行对应的任务，然后将任务的执行结果添加到指定的集合中

5、打印对应的执行时间和结果集合的数据信息

public class OldThreadPool {public static void main(String[] args) {long startTime = System.currentTimeMillis();int nThreads = Runtime.getRuntime().availableProcessors();ExecutorService threadPool = new ThreadPoolExecutor(nThreads,nThreads,100,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(10000),Executors.defaultThreadFactory(),new ThreadPoolExecutor.CallerRunsPolicy());// 集合中的数据条数int listSize = 5;CountDownLatch countDownLatch = new CountDownLatch(listSize);List<Integer> restList = new ArrayList<>();try {for (int i = 0; i < listSize; i++) {Future<Integer> submit = threadPool.submit(new Task(countDownLatch));restList.add(submit.get());}} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();} finally {threadPool.shutdown();}long endTime = System.currentTimeMillis();System.out.println("花费时间:" + (endTime - startTime));System.out.println("返回集合数据" + JSON.toJSONString(restList));System.out.println("返回集合长度" + restList.size());}
}

看似没有问题，其实最开始我也是这么觉得的。但是，电灯点火——其实不燃。

其是背后的原因是因为get方法是阻塞式的，他会等你线程池的任务得到结果以后才往下走，否则会一直卡住，直至获取返回结果。

修改后的版本一

在上面代码的基础上：

1、将线程任务都返回到一个集合中

2、然后依次遍历任务的集合，获取对应的返回结果

public class OldThreadPool {public static void main(String[] args) {long startTime = System.currentTimeMillis();int nThreads = Runtime.getRuntime().availableProcessors();ExecutorService threadPool = new ThreadPoolExecutor(nThreads,nThreads,100,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(10000),Executors.defaultThreadFactory(),new ThreadPoolExecutor.CallerRunsPolicy());// 集合中的数据条数int listSize = 5;CountDownLatch countDownLatch = new CountDownLatch(listSize);List<Integer> restList = new ArrayList<>();try {// 任务集合List<Future<Integer>> taskList = new ArrayList<>();for (int i = 0; i < listSize; i++) {Future<Integer> submit = threadPool.submit(new Task(countDownLatch));taskList.add(submit);}for (Future<Integer> future : taskList) {restList.add(future.get());}} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();} finally {threadPool.shutdown();}long endTime = System.currentTimeMillis();System.out.println("花费时间:" + (endTime - startTime));System.out.println("返回集合数据" + JSON.toJSONString(restList));System.out.println("返回集合长度" + restList.size());}
}

但是我个人并不是很倾向于这种方式，我觉得使用这种方式其实并没有解决get是阻塞的方式获取数据，它仅仅是因为第一个4s的任务是最长的，所以get方法卡在了获取第一个4s结果的位置，当第一个4s的任务执行完成后，其余get方法能够获取到结果，继续向下执行。由于是并行执行，且后续任务的执行时间都小于等于第一个任务，所以当第一个任务执行完成了，后面的任务get能够直接获取到结果。以至于最终从时间上看，达到了并行的效果。
所以我真实的处理的版本是下面的解决方式，当然上面的方式要用也能用

修改后的版本二

主要是最近刚学了CompletionService）

1、记录程序开始时间

2、自定义一个线程池然后将线程池封装到CompletionService对象中

3、初始化任务个数个计数器

4、使用completionService对象（内部封装了线程池）来执行对应的Task任务

5、等到计数器归零以后，代表所有的任务都执行完成继续向下执行任务

6、依次获取对应的返回结果

7、打印返回信息

public class FutureServiceTest {public static void main(String[] args) {long startTime = System.currentTimeMillis();int nThreads = Runtime.getRuntime().availableProcessors();ExecutorService threadPool = new ThreadPoolExecutor(nThreads,nThreads,100,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(1000),Executors.defaultThreadFactory(),new ThreadPoolExecutor.CallerRunsPolicy());CompletionService<Integer> compl = new ExecutorCompletionService<>(threadPool);int listSize = 5;CountDownLatch countDownLatch = new CountDownLatch(listSize);List<Integer> restList = new ArrayList<>();for (int i = 0; i < listSize; i++) {compl.submit(new Task(countDownLatch));}try {countDownLatch.await();for (int i = 0; i < listSize; i++) {Future<Integer> poll = compl.poll(500, TimeUnit.MILLISECONDS);if (poll != null) {restList.add(poll.get());}}} catch (ExecutionException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();} finally {threadPool.shutdown();}long endTime = System.currentTimeMillis();System.out.println("花费时间:" + (endTime - startTime));System.out.println("返回集合数据" + JSON.toJSONString(restList));System.out.println("返回集合长度" + restList.size());}
}

其实版本2和版本1在本质上没有什么区别，都能用。版本2区别与版本1的修改，只是引入了一个CompletionService类，这个类有个好处就是获取对应的结果的时候，队列中的数据是按照先完成的排在前面进行处理的。

有兴趣了解CompletionService甚至是CompletableFuture使用的小伙伴，可以参考下面的文章——

CompletableFuture异步任务的简单使用