java - 使用多线程处理大数组的替代方法

Question

我是 Java 多线程的新手。我已经在 J​​ava 中实现了一个多线程程序来处理数组，需要您的帮助和建议来优化它并在可能的情况下重构它。

场景
我们得到一个巨大的 csv 文件，它有超过 1000 行，我们需要处理它。所以我基本上将它们转换为数组，拆分它们并传递给执行程序，输入将是数组的子集。现在我将数组拆分为 20 个相等的子集并传递给 20 个线程执行。大约需要 2 分钟，这很好。如果不使用多线程，则需要 30 分钟。

需要帮助

我在下面给出了我的代码的快照。虽然它工作正常，但我想知道是否有任何方法可以对其进行更多标准化和重构。现在它看起来很笨拙。更具体地说，如果我可以参数化它，而不是创建单独的线程运行器，那就太好了。

代码

private static void ProcessRecords(List<String[]> inputCSVData)    
{    

// Do some operation    
}    


**In the main program**    

public static void main(String[] args)throws ClassNotFoundException, SQLException, IOException, InterruptedException         
{            
    int size = csvData.size();        
    // Split the array        
    int firstArraySize = (size / 20);        
    int secondArrayEndIndex = (firstArraySize * 2) - 1;        

    csvData1 = csvData.subList(1, firstArraySize);    
    csvData2 = csvData.subList(firstArraySize, secondArrayEndIndex);    
    // ....    and so on    

    Thread thread1 = new Thread(new Runnable() {    
    public void run() {    
    try {    
    ProcessRecords(csvData1);        
    } catch (ClassNotFoundException | SQLException | IOException e) {    
    // TODO Auto-generated catch block    
    e.printStackTrace();    
    }    
    }    
    });    

    Thread thread2 = new Thread(new Runnable() {    
    public void run() 
    {     
    try {    
    ProcessRecords(csvData2);            
    } catch (ClassNotFoundException | SQLException | IOException e) {    
    // TODO Auto-generated catch block    
    e.printStackTrace();        
    }    
    }    
    });    

    **and so on for 20 times**    

    thread1.start();     
    thread2.start();    
    //... For all remaining threads    
    // thread20.start();    

    thread1.join();    
    thread2.join();    
    //... For all remaining threads    
    // thread20.join();    

    }  

Answer 1

自 Java 7 以来，由于 Fork/Join Framework，您可以开箱即用地高效实现这种机制。 .从 Java 8 开始，您可以直接使用 Stream API更准确地说，并行流在后台使用 ForkJoinPool为了利用其 work-stealing提供最佳性能的算法。

在您的情况下，您可以像下一步一样逐行处理它:

csvData.parallelStream().forEach(MyClass::ProcessRecord);

使用 MyClass 类的方法 ProcessRecord 类型:

private static void ProcessRecord(String[] inputCSVData){
    // Do some operation
}

默认情况下，并行流将使用大小对应于 Runtime.getRuntime().availableProcessors() 的common ForkJoinPool对于 IO 很少的任务来说足够了，如果你有 IO 的任务，你想增加池的大小，只需将初始任务提供给你的自定义 ForkJoinPool，并行流将使用你的游泳池而不是公共(public)游泳池。

ForkJoinPool forkJoinPool = new ForkJoinPool(20);
forkJoinPool.submit(() -> csvData.parallelStream().forEach(MyClass::ProcessRecord)).get();