关于java:30G-上亿数据的超大文件如何快速导入生产环境

Hello，大家好，我是楼下小黑哥~

如果给你一个蕴含一亿行数据的超大文件，让你在一周之内将数据转化导入生产数据库，你会如何操作？

下面的问题其实是小黑哥前段时间接到一个实在的业务需要，将一个老零碎历史数据通过线下文件的形式迁徙到新的生产零碎。

~~因为老板们曾经敲定了新零碎上线工夫，所以只留给小黑哥一周的工夫将历史数据导入生产零碎。~~

因为工夫紧，而数据量又超大，所以小黑哥设计的过程想到一下解决办法：

拆分文件
多线程导入

欢送关注我的公众号：小黑十一点半，取得日常干货推送。如果您对我的专题内容感兴趣，也能够关注我的博客：studyidea.cn

首先咱们能够写个小程序，或者应用拆分命令 split 将这个超大文件拆分一个个小文件。

-- 将一个大文件拆分成若干个小文件，每个文件 100000 行
split -l 100000 largeFile.txt -d -a 4 smallFile_

这里之所以抉择先将大文件拆分，次要思考到两个起因：

第一如果程序间接读取这个大文件，假如读取一半的时候，程序忽然宕机，这样就会间接失落文件读取的进度，又须要从新结尾读取。

而文件拆分之后，一旦小文件读取完结，咱们能够将小文件挪动一个指定文件夹。

这样即便应用程序宕机重启，咱们从新读取时，只须要读取残余的文件。

第二，一个文件，只能被一个应用程序读取，这样就限度了导入的速度。

而文件拆分之后，咱们能够采纳多节点部署的形式，程度扩大。每个节点读取一部分文件，这样就能够成倍的放慢导入速度。

当咱们拆分完文件，接着咱们就须要读取文件内容，进行导入。

之前拆分的时候，设置每个小文件蕴含 10w 行的数据。因为放心一下子将 10w 数据读取利用中，导致堆内存占用过高，引起频繁的 Full GC，所以上面采纳流式读取的形式，一行一行的读取数据。

当然了，如果拆分之后文件很小，或者说利用的堆内存设置很大，咱们能够间接将文件加载到利用内存中解决。这样相对来说简略一点。

逐行读取的代码如下：

File file = ...
try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream(file), "UTF-8")) {while (iterator.hasNext()) {String line=iterator.nextLine();
        convertToDB(line);
    }

}

下面代码应用 commons-io 中的 LineIterator类，这个类底层应用了 BufferedReader 读取文件内容。它将其封装成迭代器模式，这样咱们能够很不便的迭代读取。

如果以后应用 JDK1.8，那么上述操作更加简略，咱们能够间接应用 JDK 原生的类 Files将文件转成 Stream 形式读取，代码如下：

Files.lines(Paths.get("文件门路"), Charset.defaultCharset()).forEach(line -> {convertToDB(line);
});

其实认真看下 Files#lines底层源码，其实原理跟下面的 LineIterator相似，同样也是封装成迭代器模式。

上述读取的代码写起来不难，然而存在效率问题，次要是因为只有单线程在导入，上一行数据导入实现之后，能力持续操作下一行。

为了放慢导入速度，那咱们就多来几个线程，并发导入。

多线程咱们天然将会应用线程池的形式，相干代码革新如下：

File file = ...;
ExecutorService executorService = new ThreadPoolExecutor(
        5,
        10,
        60,
        TimeUnit.MINUTES,
              // 文件数量，假如文件蕴含 10W 行
        new ArrayBlockingQueue<>(10*10000),
               // guava 提供
        new ThreadFactoryBuilder().setNameFormat("test-%d").build());
try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream(file), "UTF-8")) {while (iterator.hasNext()) {String line = iterator.nextLine();
        executorService.submit(() -> {convertToDB(line);
        });
    }

}

上述代码中，每读取到一行内容，就会间接交给线程池来执行。

咱们晓得线程池原理如下：

如果外围线程数未满，将会间接创立线程执行工作。
如果外围线程数已满，将会把工作放入到队列中。
如果队列已满，将会再创立线程执行工作。
如果最大线程数已满，队列也已满，那么将会执行回绝策略。

因为咱们上述线程池设置的外围线程数为 5，很快就达到了最大外围线程数，后续工作只能被退出队列。

为了后续工作不被线程池回绝，咱们能够采纳如下计划：

将队列容量设置成很大，蕴含整个文件所有行数
将最大线程数设置成很大，数量大于件所有行数

以上两种计划都存在同样的问题，第一种是相当于将文件所有内容加载到内存，将会占用过多内存。

而第二种创立过多的线程，同样也会占用过多内存。

一旦内存占用过多，GC 无奈清理，就可能会引起频繁的 Full GC，甚至导致 OOM，导致程序导入速度过慢。

解决这个问题，咱们能够如下两种解决方案：

CountDownLatch 批量执行
扩大线程池

`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}
            }

        });
File file = new File("文件门路");

try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream(file), "UTF-8")) {while (iterator.hasNext()) {String line = iterator.nextLine();
        executorService.submit(() -> convertToDB(line));
    }
}

一个超大的文件，咱们能够采纳拆分文件的形式，将其拆分成多份文件，而后部署多个应用程序进步读取速度。

另外读取过程咱们还能够应用多线程的形式并发导入，不过咱们须要留神线程池满载之后，将会回绝后续工作。

咱们能够通过扩大线程池，自定义回绝策略，使读取主线程阻塞。

好了，明天文章内容就到这里，不晓得各位有没有其余更好的解决办法，欢送留言探讨。

欢送关注我的公众号：小黑十一点半，取得日常干货推送。如果您对我的专题内容感兴趣，也能够关注我的博客：studyidea.cn

关于java:30G-上亿数据的超大文件如何快速导入生产环境

拆分文件

多线程导入

多线程的引入存在的问题

`CountDownLatch` 批量执行

扩大线程池

小结

拆分文件

多线程导入

多线程的引入存在的问题

CountDownLatch 批量执行

扩大线程池

小结

`CountDownLatch` 批量执行