文件的多线程读写

1. 什么时候多线程有帮助

当有大量可以同时进行的工作时，就需要多线程。比如需要处理几十个文件——复制、重计或分析——把不同部分交给不同线程要比全部顺序执行更简单。这就像不只找一个朋友来整理你的照片档案，而是叫上五个，工作会更快也更有趣。

它在以下场景尤其有用：批量处理文件、按块下载或复制大文件，或者在读取数据后需要对不同部分并行计算。

但多线程并不总是有帮助。如果只有一个小文件，专门为它启动十几个线程毫无意义。如果磁盘或网络已经满载，新线程只会拖慢进度。而如果多个线程在没有同步的情况下同时写入同一个文件——可能会造成真正的混乱，数据损坏。

简单来说，多线程是一种工具。就像锤子：能钉钉子，也可能敲到手指。关键是知道何时以及如何使用它。

2. Java 中用于多线程 I/O 的工具

你可能已经知道，Java 有多种方式并行执行任务：

• 经典的 Thread —— 手动创建线程。
• 通过 ExecutorService 的线程池 —— 现代、灵活且方便的方法。
• CompletableFuture 和并行流（Stream API）——用于更高级的任务（后续课程会详细讲解）。

先从最简单的开始：在不同线程中处理多个文件。

示例 1：经典 Thread

public class FileCopyTask extends Thread {
    private final Path source;
    private final Path target;

    public FileCopyTask(Path source, Path target) {
        this.source = source;
        this.target = target;
    }

    @Override
    public void run() {
        try {
            Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);
            System.out.println("文件已复制: " + source);
        } catch (IOException e) {
            System.err.println("复制出错 " + source + ": " + e.getMessage());
        }
    }
}

// 在独立线程中启动多个复制任务
List<Path> filesToCopy = List.of(
    Path.of("log1.txt"), Path.of("log2.txt"), Path.of("log3.txt")
);
for (Path file : filesToCopy) {
    new FileCopyTask(file, Path.of("backup_" + file.getFileName())).start();
}

优点： 简单、易懂。

缺点： 手动管理大量线程不方便，无法控制同时运行的线程数。

示例 2：ExecutorService —— 线程池

ExecutorService 允许你将任务委托给线程池，由它自行决定同时使用多少线程。

import java.util.concurrent.*;

public class MultiFileCopier {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(4); // 最多 4 个线程

        List<Path> filesToCopy = List.of(
            Path.of("log1.txt"), Path.of("log2.txt"), Path.of("log3.txt")
        );
        for (Path file : filesToCopy) {
            executor.submit(() -> {
                try {
                    Files.copy(file, Path.of("backup_" + file.getFileName()), StandardCopyOption.REPLACE_EXISTING);
                    System.out.println("已复制: " + file);
                } catch (IOException e) {
                    System.err.println("错误: " + file + " " + e.getMessage());
                }
            });
        }

        executor.shutdown(); // 不再接收新任务
        executor.awaitTermination(1, TimeUnit.MINUTES); // 等待所有任务完成
    }
}

优点：

• 易于扩展（可设置所需线程数）。
• 便于控制任务结束（方法 shutdown()、awaitTermination(...)）。
• 适合处理成百上千个文件。

3. 多线程 I/O 的问题与限制

资源竞争

如果在没有同步的情况下尝试从多个线程同时读写同一个文件——结果将不可预测。这就像两个人同时在书的同一页上写字：只会变成一团糟。为协调访问，请使用例如 synchronized、显式锁，或单独的写入线程。

文件系统和操作系统的限制

• 并非所有文件系统都能很好地支持对同一文件的并发写入。
• 操作系统可能会限制同时打开的文件数量。
• 机械硬盘（尤其是 HDD）在大量随机访问时表现很差。

写入共享资源时的同步

如果多个线程向同一个文件（例如日志）写入，务必使用同步（例如通过 synchronized、锁，或专门的写入线程）。

小文件时效率低

对于小文件，创建线程和线程切换的开销可能会超过并行带来的收益。

4. 实践示例

并行复制文件

假设我们有一个装满日志的文件夹，需要将它们复制到归档目录。

import java.nio.file.*;
import java.util.List;
import java.util.concurrent.*;

public class ParallelFileCopier {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(4);

        List<Path> filesToCopy = List.of(
            Path.of("log1.txt"), Path.of("log2.txt"), Path.of("log3.txt")
            // ... 按需添加任意数量的文件
        );

        for (Path file : filesToCopy) {
            executor.submit(() -> {
                try {
                    Path target = Path.of("archive", file.getFileName().toString());
                    Files.copy(file, target, StandardCopyOption.REPLACE_EXISTING);
                    System.out.println("已复制: " + file);
                } catch (IOException e) {
                    System.err.println("错误: " + file + " " + e.getMessage());
                }
            });
        }

        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.MINUTES);
    }
}

说明：

• 我们使用包含 4 个线程的线程池——这通常足以让磁盘保持忙碌，但不会让系统过载。
• 对于 1000 个文件，可以把线程池增至 8，但不要过大。

使用 Stream API 并行处理文件行

从 Java 8 起，你可以使用并行流来处理文件内容：

import java.nio.file.*;
import java.io.IOException;

public class ParallelLineProcessing {
    public static void main(String[] args) throws IOException {
        Path path = Path.of("biglog.txt");

        // Files.lines 返回 Stream<String> —— 文件行的流
        Files.lines(path)
            .parallel() // 转为并行流
            .filter(line -> line.contains("ERROR"))
            .forEach(line -> System.out.println("错误: " + line));
    }
}

重要：

• 并行流能加速的是需要大量计算的处理（CPU 受限），而不是读取本身（I/O 受限）。
• 如果处理逻辑很简单（例如仅通过 System.out.println 打印），可能不会有提升。

读/写同一大文件的不同片段

Java 允许借助 FileChannel 和位置方法同时读写同一文件的不同区域。这属于进阶内容，但原理很简单：每个线程处理自己的一段文件。

import java.nio.channels.FileChannel;
import java.nio.file.*;
import java.io.*;
import java.nio.ByteBuffer;

public class FileChunkReader implements Runnable {
    private final Path path;
    private final long position;
    private final long size;

    public FileChunkReader(Path path, long position, long 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((int) size);
            channel.read(buffer, position);
            System.out.println("已读取从位置 " + position + " 开始、大小为 " + size + " 的片段");
            // 这里可以处理 buffer
        } catch (IOException e) {
            System.err.println("读取片段出错: " + e.getMessage());
        }
    }
}

// 运行示例：用 4 个线程按 1 MB 分块读取文件
Path file = Path.of("bigdata.bin");
long fileSize = Files.size(file);
long chunkSize = 1024 * 1024; // 1 MB
int chunks = (int) Math.ceil((double) fileSize / chunkSize);

ExecutorService executor = Executors.newFixedThreadPool(4);

for (int i = 0; i < chunks; i++) {
    long position = i * chunkSize;
    long size = Math.min(chunkSize, fileSize - position);
    executor.submit(new FileChunkReader(file, position, size));
}

executor.shutdown();
executor.awaitTermination(10, TimeUnit.MINUTES);

说明：

• 每个线程读取自己的文件片段，互不干扰。
• 这种方法常用于种子下载器和各类下载器。

写入共享文件时的同步

如果多个线程向同一个文件（例如日志）写入，必须进行同步，否则会得到“杂乱无章”的行：

import java.io.*;

public class SafeLogger {
    private final Writer writer;

    public SafeLogger(String filename) throws IOException {
        this.writer = new BufferedWriter(new FileWriter(filename, true));
    }

    public synchronized void log(String message) throws IOException {
        writer.write(message);
        writer.write(System.lineSeparator());
        writer.flush();
    }

    public void close() throws IOException {
        writer.close();
    }
}

说明：

• 方法 log 使用了 synchronized，以确保任一时刻只有一个线程写入文件。
• 这可行，但在线程很多时可能成为瓶颈——更好的做法是写入不同文件后再合并。

5. 什么时候不该使用多线程

多线程很有诱惑力：线程越多，速度就越快！但在实践中并非总是如此。如果你只处理少量小文件，顺序执行更简单也更可靠。你为启动线程和进行协调付出的时间，往往得不偿失。

有时问题根本不在磁盘，而在网络——此时增加线程并不会加速，因为瓶颈（bottleneck）在别处。另一个陷阱是并行写入同一个文件。如果你缺乏同步方面的经验，最好别尝试：很可能得到损坏的数据。

最后，如果你的磁盘或文件系统不适合被几十个线程同时“拉扯”，多线程不仅无济于事，反而会让情况更糟。

简而言之，如果你觉得“线程越多越好”，多数情况下并非如此。有时一个安稳的线程比十个匆忙的线程完成得更干净、更快、更可靠。

6. 面向进阶任务的 FileChannel 简介

来自 java.nio.channels 的 FileChannel 是用于低层文件操作的工具，允许按任意位置读写数据。借此可以实现并行下载或按块处理大文件等。

示例：

try (FileChannel channel = FileChannel.open(Path.of("bigfile.bin"), StandardOpenOption.READ)) {
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    long position = 0;
    int bytesRead = channel.read(buffer, position); // 读取从位置 0 开始的 1024 字节
    // 处理 buffer
}

重要：

• FileChannel 不是线程安全的——如果多个线程共享同一个通道，需要自行实现同步。
• 要并行处理，更简单的方法是为每个线程分别打开一个通道。

7. 多线程 I/O 的常见错误

错误 1：对同一文件的写入缺乏同步。
结果是数据损坏、出现奇怪字符，甚至文件无法读取。务必同步访问或写入不同文件。

错误 2：线程过多。
如果你为复制 1000 个文件而开启 1000 个线程，计算机可能会“生气”（OutOfMemoryError、卡顿、崩溃）。请使用线程池（ExecutorService）并限制线程数量。

错误 3：未关闭流/文件。
每个打开的流都是操作系统资源。如果不关闭，可能会遇到 “Too many open files” 错误。使用 try-with-resources，或别忘记调用 close()。

错误 4：程序过早退出。
如果没有等待所有线程结束（例如未调用 executor.awaitTermination(...)），程序可能在文件复制完成前就退出。

错误 5：未考虑位置就并行写入同一文件区域。
如果多个线程写入同一文件的同一区域，数据会被混淆。进行按位置写入时，请使用通道并明确划分区间。