1. 并行流（parallelStream）：简单且易用
如果你已经使用过 Stream API，就知道过滤、映射与收集集合有多便利。一个重要的加分项是：只需一行代码就能让任何流变成并行流——调用 parallel()，集合元素就会被并发处理。
这对独立的文件操作尤其有用：统计行数、查找子串、复制、压缩等。
示例：并行统计目录中所有文件的行数
方案 1：顺序执行
import java.nio.file.*;
import java.io.IOException;
import java.util.List;
public class LogLineCounter {
public static void main(String[] args) throws IOException {
Path logDir = Paths.get("logs");
long totalLines = 0;
try (DirectoryStream<Path> stream = Files.newDirectoryStream(logDir, "*.log")) {
for (Path file : stream) {
long lines = Files.lines(file).count();
totalLines += lines;
}
}
System.out.println("所有日志的总行数：" + totalLines);
}
}
说明：一切按顺序进行，一个文件接一个文件。如果文件很多且很大，就得等很久。
方案 2：并行！
import java.nio.file.*;
import java.io.IOException;
import java.util.stream.Stream;
public class LogLineCounterParallel {
public static void main(String[] args) throws IOException {
Path logDir = Paths.get("logs");
try (Stream<Path> files = Files.list(logDir)) {
long totalLines = files
.filter(path -> path.toString().endsWith(".log"))
.parallel() // 魔法就在这里！
.mapToLong(file -> {
try (Stream<String> lines = Files.lines(file)) {
return lines.count();
} catch (IOException e) {
e.printStackTrace();
return 0L;
}
})
.sum();
System.out.println("所有日志的总行数：" + totalLines);
}
}
}
说明：关键一行是 .parallel()。在多核处理器上，程序通常会明显更快。
它是如何工作的？
- parallel() 将普通流转换为并行流。底层使用公共的 ForkJoinPool（默认线程数等于 CPU 核心数）。
- 每个文件独立处理，结果通过终端操作聚合（例如 sum()）。
- 如果文件很少——可能没有加速；如果有数百个——通常会有明显收益。
重要！
- 并行流并不会让 I/O 操作本身更快；它让多个操作可以同时进行。在快速介质（SSD）上有帮助，在慢速介质（HDD）上可能会受到磁盘“瓶颈”的限制。
2. ForkJoinPool：“分而治之”的实战
ForkJoin 是一个“分而治之”的并行计算框架：把大任务拆成子任务并行执行，然后合并结果。由专门的线程池 ForkJoinPool 进行管理。它正是并行流“幕后”所用的实现，但你也可以直接使用它以获得更大的灵活性。
这种模型对递归结构（目录树）、大型数据数组以及容易分解为独立部分的任务特别有效。
示例：对目录树进行递归搜索
查找所有 ".txt" 文件（包括子文件夹），并统计总行数。
import java.nio.file.*;
import java.util.concurrent.*;
import java.util.*;
import java.io.IOException;
import java.util.stream.Collectors;
public class FolderLineCounter extends RecursiveTask<Long> {
private final Path dir;
public FolderLineCounter(Path dir) {
this.dir = dir;
}
@Override
protected Long compute() {
List<FolderLineCounter> subTasks = new ArrayList<>();
long lines = 0;
try (DirectoryStream<Path> stream = Files.newDirectoryStream(dir)) {
for (Path entry : stream) {
if (Files.isDirectory(entry)) {
FolderLineCounter task = new FolderLineCounter(entry);
task.fork(); // 启动子任务
subTasks.add(task);
} else if (entry.toString().endsWith(".txt")) {
try (Stream<String> fileLines = Files.lines(entry)) {
lines += fileLines.count();
} catch (IOException e) {
e.printStackTrace();
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
// 汇总子任务的结果
for (FolderLineCounter task : subTasks) {
lines += task.join();
}
return lines;
}
public static void main(String[] args) {
Path root = Paths.get("big_folder");
ForkJoinPool pool = new ForkJoinPool();
FolderLineCounter counter = new FolderLineCounter(root);
long totalLines = pool.invoke(counter);
System.out.println("所有 .txt 文件的总行数：" + totalLines);
}
}
这里发生了什么：
- 为每个文件夹创建一个独立任务（FolderLineCounter），对子目录创建各自的子任务（fork()）。
- 文件在当前层进行统计，在所有子任务 join() 之后汇总结果。
ForkJoin 的优势是什么？
- 对大型层级结构（目录树）效果显著。
- 最大化利用 CPU 核心。
- 可以精确控制并行化与任务边界。
3. 实用应用场景
批量文件处理
例如，需要把上千张照片复制到备份文件夹。
import java.nio.file.*;
import java.util.List;
import java.util.stream.Collectors;
public class ParallelFileCopier {
public static void main(String[] args) throws Exception {
Path sourceDir = Paths.get("photos");
Path destDir = Paths.get("photos_backup");
Files.createDirectories(destDir);
List<Path> files = Files.list(sourceDir)
.filter(Files::isRegularFile)
.collect(Collectors.toList());
files.parallelStream().forEach(file -> {
try {
Path destFile = destDir.resolve(file.getFileName());
Files.copy(file, destFile, StandardCopyOption.REPLACE_EXISTING);
} catch (Exception e) {
e.printStackTrace();
}
});
System.out.println("所有文件已复制完成！");
}
}
说明：每个文件在独立线程中复制。文件数量很大时，加速效果明显。
并行压缩/解压
同样可以通过 parallelStream() 或自定义 ForkJoinPool 来并行化压缩、哈希重算、图像格式转换等操作。
4. 重要注意事项与限制
- I/O 操作不一定总能从并行中获益。 如果磁盘或网络是“瓶颈”，同时跑上百个并行任务只会增加资源竞争。
- 不要启动过多线程。 默认情况下，并行流使用公共的 ForkJoinPool.commonPool()，其并行度 ≈ CPU 核心数。可以通过属性 "java.util.concurrent.ForkJoinPool.common.parallelism" 调整，但请谨慎操作。
- 别忘了同步。 如果多个线程写入同一个文件/对象——请使用同步与队列；对于彼此独立的文件——不需要同步。
5. 了解 FileChannel 与定位访问
在更高级的场景中（例如并行读取同一个大文件的不同部分），可以使用 java.nio.channels.FileChannel，它支持定位读/写。
示例：在不同线程中读取文件的不同区块
import java.nio.channels.FileChannel;
import java.nio.file.*;
import java.nio.ByteBuffer;
public class FileChunkReader implements Runnable {
private final Path path;
private final long position;
private final int size;
public FileChunkReader(Path path, long position, int size) {
this.path = path;
this.position = position;
this.size = size;
}
@Override
public void run() {
try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) {
ByteBuffer buffer = ByteBuffer.allocate(size);
channel.read(buffer, position);
// 处理数据
System.out.println("已读取 " + buffer.position() + " 字节，自位置 " + position);
} catch (Exception e) {
e.printStackTrace();
}
}
}
说明：同时启动多个这样的任务——每个任务读取自己的范围。不过要注意：并非所有磁盘和文件系统都能很好地承受强并行负载。
6. 并行文件处理中的常见错误
错误 1：未同步的并行写入同一个文件。 数据会交叉、破坏。请使用队列、缓冲与写入同步。
错误 2：并行度过高。 在较弱的硬件/很小的文件上，并行流引入的开销可能导致变慢。
错误 3：忽略 I/O 错误。 在并行流中异常很容易被“吞掉”——请在 lambda 内处理、记录日志，并计及失败情况。
错误 4：资源未关闭。 始终为流/通道使用 try-with-resources，否则会出现泄漏与奇怪错误。
错误 5：对 parallel() 抱有“魔法”期待。 并行只有在工作量足够且资源（CPU、快速磁盘）可用时才会加速。单靠调用 parallel() 并不是银弹。
GO TO FULL VERSION