我最近致力于外部合并排序算法(外部排序)的实现,我的实现需要使用多线程方法。
我尝试使用 ForkJoinPool 而不是使用 Java 中较旧的实现,例如 Thread 和 ExecutorService。该算法的第一步需要读取一个文件,每x行收集并发送以排序并写入文件。此操作(排序和保存)可以在主线程读取下一批时在单独的线程中完成。我已经写了一个方法来做到这一点(见下文)。
我担心的是,实际的并行工作在我使用时并没有开始,ForkJoinPool.commonPool().submit(()->SortAndWriteToFile(lines, fileName))
而是只有在我task.join()
在循环完成后调用时才开始。这意味着在一个足够大的循环中,我将整理要运行的任务,但没有任何时间运行它们。当我使用它invoke
而不是submit
它时,我似乎无法控制join
意志在哪里,也不能保证在继续之前完成所有工作。
有没有更正确的方法来实现这一点?
我的代码如下。列出了该方法和两个实用方法。我希望这不会太长。
protected int generateSortedFiles (String originalFileName, String destinationFilePrefix) {
//Number of accumulated sorted blocks of size blockSize
int blockCount = 0;
//hold bufferSize number of lines from the file
List<String> bufferLines = new ArrayList<String>();
List<ForkJoinTask<?>> taskList = new ArrayList<ForkJoinTask<?>>();
//Open file to read
try (Stream<String> fileStream = Files.lines(Paths.get(originalFileName))) {
//Iterate over BufferSize lines to add them to list.
Iterator<String> lineItr = fileStream.iterator();
while(lineItr.hasNext()) {
//Add bufferSize lines to List
for (int i=0;i<bufferSize;i++) {
if (lineItr.hasNext()) {
bufferLines.add(lineItr.next());
}
}
//submit the task to sort and write to file in a separate thread
String fileName= destinationFilePrefix+blockCount+".csv";
List<String> lines = Collections.unmodifiableList(bufferLines);
taskList.add(ForkJoinPool.commonPool().submit(
()->SortAndWriteToFile(lines, fileName)));
blockCount++;
bufferLines = new ArrayList<String>();
}
} catch (IOException e) {
System.out.println("read from file " +originalFileName + "has failed due to "+e);
} catch (ArrayIndexOutOfBoundsException e) {
System.out.println("the index prodived was not available in the file "
+originalFileName+" and the error is "+e);
}
flushParallelTaskList(taskList);
return blockCount;
}
/**
* This method takes lines, sorts them and writes them to file
* @param lines the lines to be sorted
* @param fileName the filename to write them to
*/
private void SortAndWriteToFile(List<String> lines, String fileName) {
//Sort lines
lines = lines.stream()
.parallel()
.sorted((e1,e2) -> e1.split(",")[indexOfKey].compareTo(e2.split(",")[indexOfKey]))
.collect(Collectors.toList());
//write the sorted block of lines to the destination file.
writeBuffer(lines, fileName);
}
/**
* Wait until all the threads finish, clear the list
* @param writeList
*/
private void flushParallelTaskList (List<ForkJoinTask<?>> writeList) {
for (ForkJoinTask<?> task:writeList) {
task.join();
}
writeList.clear();
}