解决大文件的OOM的问题

痛定思痛。 2022-03-19 01:52 1277阅读 0赞

下面就为大家带来一篇完美解决java读取大文件内存溢出的问题。本人觉得挺不错的，现在就分享给大家，也给大家做个参考。

一、传统方式  
1、在内存中读取文件内容  
读取文件行的标准方式是在内存中读取，Guava 和Apache Commons IO都提供了如下所示快速读取文件行的方法：

Files.readLines(new File(path), Charsets.UTF\_8);  
FileUtils.readLines(new File(path));

实际上是使用BufferedReader或者其子类LineNumberReader来读取的。

传统方式的问题：是文件的所有行都被存放在内存中，当文件足够大时很快就会导致程序抛出OutOfMemoryError 异常。

2、问题思考：  
我们通常不需要把文件的所有行一次性地放入内存中，相反，我们只需要遍历文件的每一行，然后做相应的处理，  
处理完之后把它扔掉。所以我们可以通过行迭代方式来读取，而不是把所有行都放在内存中。

二、大文件读取处理方式

不重复读取与不耗尽内存的情况下处理大文件：

1、文件流方式：使用java.util.Scanner类扫描文件的内容，一行一行连续地读取  
   
FileInputStream inputStream = null;  
Scanner sc = null;  
try \{  
 inputStream = new FileInputStream(path);  
 sc = new Scanner(inputStream, UTF-8);  
 while (sc.hasNextLine()) \{  
  String line = sc.nextLine();  
  // System.out.println(line);  
  \}  
\}catch(IOException e)\{  
  logger.error(e);  
\}finally \{  
  if (inputStream != null) \{  
  inputStream.close();  
  \}  
  if (sc != null) \{  
    sc.close();  
   \}  
\}

该方案将会遍历文件中的所有行，允许对每一行进行处理，而不保持对它的引用。总之没有把它们存放在内存中！

2、Apache Commons IO流：使用Commons IO库实现，利用该库提供的自定义LineIterator

LineIterator it = FileUtils.lineIterator(theFile, UTF-8);  
try \{  
 while (it.hasNext()) \{  
 String line = it.nextLine();  
 // do something with line  
  \}  
\} finally \{  
 LineIterator.closeQuietly(it);  
\}  
   
该方案由于整个文件不是全部存放在内存中，这也就导致相当保守的内存消耗。

三、针对多个大文件的处理(对指定路径下的文件处理)  
1、采用多线程处理文件  
首先创建自定义线程池，然后获取文件列表，用循环遍历文件列表，获取每一个文件，最后将读取文件的任务放入事先创建的线程池中。

2、读取文件  
采用Apache Commons IO流实现:  
举例说明，处理思想是分批处理，有效控制处理速率。代码如下：

int corePoolSize = 5;// 核心线程数，可以采用Redis动态配置  
int eachDealNum = 1000;// 到期券文件处理每次处理数据量,可以采用Redis动态配置  
long sleepTime = 1000L;// 到期券文件处理每次处理数据的休眠时间(以毫秒为单位),可以采用Redis动态配置

long count=0;// 计数  
List<TInfoPasVoucherRec> listPasVoucherRecs = new ArrayList<TInfoPasVoucherRec>(eachDealNum);  
String tempString;  
TInfoPasVoucherRec tInfoPasVoucherRec = null;  
LineIterator it = FileUtils.lineIterator(theFile, UTF-8);  
try \{  
 while (it.hasNext()) \{  
    count++;// 每读取一行数据，计数器+1  
    tempString = lineIterator.nextLine();  
    String\[\] srt = tempString.split("\\\\,");  
    String productNo = srt\[0\];// 产品号  
    String batchNo = srt\[1\];// 券批次号  
    ..........  
    tInfoPasVoucherRec = new TInfoPasVoucherRec();  
    tInfoPasVoucherRec.setProductNo(productNo);  
    tInfoPasVoucherRec.setBatchNo(batchNo);  
    ..........  
    listPasVoucherRecs.add(tInfoPasVoucherRec);// 读数据到List  
          
    // 发出一个信号让GC抽时间回收调上面new的对象即释放内存,防止出现内存泄漏或溢出问题  
    tInfoPasVoucherRec = null;  
    // 每eachDealNum条数据扔线程池处理  
    if (count % eachDealNum == 0) \{  
        if (!listPasVoucherRecs.isEmpty()) \{  
            // 多线程处理到期券  
            threadDeal(listPasVoucherRecs, pasVoucherRecService, pasFlowRecService, mobileHService,  
                    producerManager, batchNoLists, redBagBatchNoList, expireRemindFlag, fileName);  
            listPasVoucherRecs = new ArrayList<TInfoPasVoucherRec>(eachDealNum);  
        \} else \{  
            LOG.info("没有可执行的数据");  
        \}  
    \}  
  \}  
   
    // while循环外的判断，为了防止上面判断后剩下最后少于eachDealNum条的数据没有被处理  
    if (count % eachDealNum != 0) \{  
        if (!listPasVoucherRecs.isEmpty()) \{  
            // 多线程处理到期券  
            threadDeal(listPasVoucherRecs);  
        \} else \{  
            // ....  
        \}  
    \}  
\} finally \{  
 // 关闭文件流  
 LineIterator.closeQuietly(it);  
\}

/\*\*  
 \* 多线程处理到期券  
 \*/  
private static void threadDeal(List<TInfoPasVoucherRec> listPasVoucherRecs) \{  
    // 避免大并发  
    try \{  
        Thread.sleep(sleepTime);// 线程休眠  
    \} catch (Exception e) \{  
        LOG.error("线程休眠异常:\{\}", e);  
    \}  
      
    try \{  
        int listSize = listPasVoucherRecs.size();  
        /\*\*  
         \* 创建自定义线程池  
         \* 通常核心线程数可以设为CPU数量+1，而最大线程数可以设为CPU的数量\*2+1。  
         \*/  
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize + 1, corePoolSize \* 2 + 1, 200L,  
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(listSize));  
          
        for (TInfoPasVoucherRec pasVoucherRec : listPasVoucherRecs) \{  
            threadPoolExecutor.execute(new PasVoucherDataDisposeThread(pasVoucherRec));  
        \}  
        // 关闭线程池  
        threadPoolExecutor.shutdown();  
    \} catch (Exception e) \{  
        // ....  
    \}  
    // 释放内存,防止出现内存泄漏或溢出问题  
    listPasVoucherRecs = null;  
\}

三、Java 线程池介绍  
参考https://www.cnblogs.com/zhujiabin/p/5404771.html或者https://www.cnblogs.com/dolphin0520/p/3932921.html  
1、线程池的作用：  
线程池作用就是限制系统中执行线程的数量。  
根据系统的环境情况，可以自动或手动设置线程数量，达到运行的最佳效果；少了浪费了系统资源，  
多了造成系统拥挤效率不高。用线程池控制线程数量，其他线程排 队等候。一个任务执行完毕，  
再从队列的中取最前面的任务开始执行。若队列中没有等待进程，线程池的这一资源处于等待。当一个新任务需要运行时，  
如果线程池 中有等待的工作线程，就可以开始运行了；否则进入等待队列。

2、为什么要用线程池:

(1)、减少了创建和销毁线程的次数，每个工作线程都可以被重复利用，可执行多个任务。

(2)、可以根据系统的承受能力，调整线程池中工作线线程的数目，防止因为消耗过多的内存，  
而把服务器累趴下(每个线程需要大约1MB内存，线程开的越多，消耗的内存也就越大，最后死机)。

3、Java通过Executors提供四种线程池，分别为：

newCachedThreadPool--创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。  
newFixedThreadPool---创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。  
newScheduledThreadPool---创建一个定长线程池，支持定时及周期性任务执行。  
newSingleThreadExecutor---创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

(1)、newSingleThreadExecutor  
创建一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，  
那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。

(2)、newFixedThreadPool  
创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，  
如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。

(3)、 newCachedThreadPool  
创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，  
那么就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。  
此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者说JVM）能够创建的最大线程大小。

(4)、newScheduledThreadPool  
创建一个大小无限的线程池。此线程池支持定时以及周期性执行任务的需求。

4、建议采用自定义线程池处理多线程，参考https://blog.csdn.net/qq\_25806863/article/details/71126867

(一)、ThreadPoolExecutor是线程池的真正实现，他通过构造方法的一系列参数，来构成不同配置的线程池。常用的构造方法有下面四个：  
(1)、ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)  
(2)ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory)  
(3)、ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler)  
(4)、ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

(二)、构造方法参数说明

(1)、corePoolSize

核心线程数，默认情况下核心线程会一直存活，即使处于闲置状态也不会受存keepAliveTime限制。除非将allowCoreThreadTimeOut设置为true。

(2)、maximumPoolSize

线程池所能容纳的最大线程数。超过这个数的线程将被阻塞。当任务队列为没有设置大小的LinkedBlockingDeque时，这个值无效。

(3)、keepAliveTime

非核心线程的闲置超时时间，超过这个时间就会被回收。

(4)、unit

指定keepAliveTime的单位，如TimeUnit.SECONDS。当将allowCoreThreadTimeOut设置为true时对corePoolSize生效。

(5)、workQueue

线程池中的任务队列.

常用的有三种队列，SynchronousQueue,LinkedBlockingDeque,ArrayBlockingQueue。

(6)、threadFactory

线程工厂，提供创建新线程的功能。ThreadFactory是一个接口，只有一个方法

【关键词释义  
corePoolSize在很多地方被翻译成核心池大小，其实我的理解这个就是线程池的大小。举个简单的例子：  
假如有一个工厂，工厂里面有10个工人，每个工人同时只能做一件任务。因此只要当10个工人中有工人是空闲的，来了任务就分配给空闲的工人做；  
当10个工人都有任务在做时，如果还来了任务，就把任务进行排队等待；  
如果说新任务数目增长的速度远远大于工人做任务的速度，那么此时工厂主管可能会想补救措施，比如重新招4个临时工人进来；  
然后就将任务也分配给这4个临时工人做；  
如果说这14个工人做任务的速度还是不够，此时工厂主管可能就要考虑不再接收新的任务或者抛弃前面的一些任务了。  
当这14个工人当中有人空闲时，而新任务增长的速度又比较缓慢，工厂主管可能就考虑辞掉4个临时工了，只保持原来的10个工人，毕竟请额外的工人是要花钱的。  
这个例子中的corePoolSize就是10，而maximumPoolSize就是14（10+4）。  
也就是说corePoolSize就是线程池大小，maximumPoolSize在我看来是线程池的一种补救措施，即任务量突然过大时的一种补救措施。  
】

(三)、线程池规则

线程池的线程执行规则跟任务队列有很大的关系。  
下面都假设任务队列没有大小限制：  
(1)、如果线程数量 <= 核心线程数量，那么直接启动一个核心线程来执行任务，不会放入队列中。  
(2)、如果线程数量 > 核心线程数，但<=最大线程数，并且任务队列是LinkedBlockingDeque的时候，超过核心线程数量的任务会放在任务队列中排队。

(3)、如果线程数量 > 核心线程数，但<=最大线程数，并且任务队列是SynchronousQueue的时候，线程池会创建新线程执行任务，  
这些任务也不会被放在任务队列中。这些线程属于非核心线程，在任务完成后，闲置时间达到了超时时间就会被清除。

(4)、如果线程数量 > 核心线程数，并且>最大线程数，当任务队列是LinkedBlockingDeque，会将超过核心线程的任务放在任务队列中排队。  
也就是当任务队列是LinkedBlockingDeque并且没有大小限制时，线程池的最大线程数设置是无效的，他的线程数最多不会超过核心线程数。

(5)、如果线程数量 > 核心线程数，并且>最大线程数，当任务队列是SynchronousQueue的时候，会因为线程池拒绝添加任务而抛出异常。

(四)、任务队列大小有限制

(1)、当LinkedBlockingDeque塞满时，新增的任务会直接创建新线程来执行，当创建的线程数量超过最大线程数量时会抛异常。

(2)、SynchronousQueue没有数量限制。因为他根本不保持这些任务，而是直接交给线程池去执行。当任务数量超过最大线程数时会直接抛异常。

【注意】  
线程资源必须通过线程池提供，不允许在应用中自行显式创建线程。  
线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。  
说明：Executors 返回的线程池对象的弊端如下：高并发时，同步调用应该去考量锁的性能损耗。能用无锁数据结构，  
就不要用锁；能锁区块，就不要锁整个方法体；能用对象锁，就不要用类锁。