我们知道线程池原理如下：
如果核心线程数未满，将会直接创建线程执行任务。
如果核心线程数已满，将会把任务放入到队列中。
如果队列已满，将会再创建线程执行任务。
如果最大线程数已满，队列也已满，那么将会执行拒绝策略。
由于我们上述线程池设置的核心线程数为 5，很快就到达了最大核心线程数，后续任务只能被加入队列。
为了后续任务不被线程池拒绝，我们可以采用如下方案：
将队列容量设置成很大，包含整个文件所有行数
将最大线程数设置成很大，数量大于件所有行数
以上两种方案都存在同样的问题，第一种是相当于将文件所有内容加载到内存，将会占用过多内存。
而第二种创建过多的线程，同样也会占用过多内存。
一旦内存占用过多，GC 无法清理，就可能会引起频繁的?Full GC，甚至导致?OOM，导致程序导入速度过慢。
解决这个问题，我们可以如下两种解决方案：
CountDownLatch?批量执行
JDK 提供的?CountDownLatch，可以让主线程等待子线程都执行完成之后，再继续往下执行。
利用这个特性，我们可以改造多线程导入的代码,主体逻辑如下：
try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream(file), "UTF-8")) { // 存储每个任务执行的行数 List<String> lines = Lists.newArrayList(); // 存储异步任务 List<ConvertTask> tasks = Lists.newArrayList(); while (iterator.hasNext()) { String line = iterator.nextLine(); lines.add(line); // 设置每个线程执行的行数 if (lines.size() == 1000) { // 新建异步任务，注意这里需要创建一个 List tasks.add(new ConvertTask(Lists.newArrayList(lines))); lines.clear(); } if (tasks.size() == 10) { asyncBatchExecuteTask(tasks); }
} // 文件读取结束，但是可能还存在未被内容 tasks.add(new ConvertTask(Lists.newArrayList(lines))); // 最后再执行一次 asyncBatchExecuteTask(tasks);}
复制代码
这段代码中，每个异步任务将会导入 1000 行数据，等积累了 10 个异步任务，然后将会调用?asyncBatchExecuteTask?使用线程池异步执行。
/** * 批量执行任务 * * @param tasks */private static void asyncBatchExecuteTask(List<ConvertTask> tasks) throws InterruptedException { CountDownLatch countDownLatch = new CountDownLatch(tasks.size()); for (ConvertTask task : tasks) { task.setCountDownLatch(countDownLatch); executorService.submit(task); } // 主线程等待异步线程 countDownLatch 执行结束 countDownLatch.await(); // 清空，重新添加任务 tasks.clear();}
复制代码
asyncBatchExecuteTask?方法内将会创建?CountDownLatch，然后主线程内调用?await方法等待所有异步线程执行结束。
ConvertTask?异步任务逻辑如下：
/** * 异步任务 * 等数据导入完成之后，一定要调用 countDownLatch.countDown() * 不然，这个主线程将会被阻塞， */private static class ConvertTask implements Runnable {
private CountDownLatch countDownLatch;
private List<String> lines;
public ConvertTask(List<String> lines) { this.lines = lines; }
public void setCountDownLatch(CountDownLatch countDownLatch) { this.countDownLatch = countDownLatch; }
@Override public void run() { try { for (String line : lines) { convertToDB(line); } } finally { countDownLatch.countDown(); } }}
复制代码
ConvertTask任务类逻辑就非常简单，遍历所有行，将其导入到数据库中。所有数据导入结束，调用?countDownLatch#countDown。
一旦所有异步线程执行结束，调用?countDownLatch#countDown，主线程将会被唤醒，继续执行文件读取。
虽然这种方式解决上述问题，但是这种方式，每次都需要积累一定任务数才能开始异步执行所有任务。
另外每次都需要等待所有任务执行结束之后，才能开始下一批任务，批量执行消耗的时间等于最慢的异步任务消耗的时间。
这种方式线程池中线程存在一定的闲置时间，那有没有办法一直压榨线程池，让它一直在干活呢？
扩展线程池
回到最开始的问题，文件读取导入，其实就是一个生产者-消费者消费模型。
主线程作为生产者不断读取文件，然后将其放置到队列中。
异步线程作为消费者不断从队列中读取内容，导入到数据库中。
一旦队列满载，生产者应该阻塞，直到消费者消费任务。
其实我们使用线程池的也是一个生产者-消费者消费模型，其也使用阻塞队列。
那为什么线程池在队列满载的时候，不发生阻塞？
这是因为线程池内部使用?offer?方法，这个方法在队列满载的时候不会发生阻塞，而是直接返回 。
那我们有没有办法在线程池队列满载的时候，阻塞主线程添加任务？
其实是可以的，我们自定义线程池拒绝策略，当队列满时改为调用?BlockingQueue.put?来实现生产者的阻塞。
RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { if (!executor.isShutdown()) { try { executor.getQueue().put(r); } catch (InterruptedException e) { // should not be interrupted } }
}};
复制代码
这样一旦线程池满载，主线程将会被阻塞。
使用这种方式之后，我们可以直接使用上面提到的多线程导入的代码。
ExecutorService executorService = new ThreadPoolExecutor( 5, 10, 60, TimeUnit.MINUTES, new ArrayBlockingQueue<>(100), new ThreadFactoryBuilder().setNameFormat("test-%d").build(), (r, executor) -> { if (!executor.isShutdown()) { try { // 主线程将会被阻塞 executor.getQueue().put(r); } catch (InterruptedException e) { // should not be interrupted } }
复制代码
评论