最近在看处理一个Bug的时候，发现了自己之前写的代码的一些问题，还好后来自己提出了优化，修改掉了。特作此记录。
项目背景：
我需要根据集合中的数据记录，将每一条数据封装为对应的报文，然后调用发送给外部接口，最后解析对应的返回报文，最终将数据落库。
由于数据库的数据记录比较多，所以这个操作是一个十分耗时的操作，所以我第一反应就是使用多线程进行处理，最终我决定使用线程池进行该任务的操作。
下列代码我根据真实的项目代码抽象而来。
Task任务
1、定义一个任务类，实现Callable接口，用于接收返回值
2、每一个任务内部定义了一个计数器，用于任务完成的标记
3、内部随机获取一个五以内的值，用于后期的线程睡眠，代表业务执行时间
4、返回对应的执行结果，此处用休眠时间代替
public class Task implements Callable<Integer> {
private CountDownLatch num;
public Task(CountDownLatch num) {
this.num = num;
}
@Override
public 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使用的小伙伴，可以参考下面的文章——