# Java多线程处理文件的技术详解与实现方案

## 引言：多线程文件处理的必要性

在现代软件开发中，文件处理是常见的业务需求。当面对大规模文件或需要高性能处理的场景时，传统的单线程文件处理方式往往无法满足性能要求。Java多线程技术为文件处理提供了强大的并发能力，能够显著提升I/O密集型操作和CPU密集型文件处理的效率。本文将深入探讨Java多线程处理文件的各种技术方案、最佳实践及常见问题解决方案。

## 基础概念与设计模式

### 生产者-消费者模式

在文件处理中，生产者-消费者模式是最常用的多线程设计模式之一。

```java

import java.io.*;

import java.nio.file.*;

import java.util.concurrent.*;

import java.util.*;

public class ProducerConsumerFileProcessor {

    

    // 文件行记录

    public static class FileRecord {

        private final String fileName;

        private final int lineNumber;

        private final String content;

        

        public FileRecord(String fileName, int lineNumber, String content) {

            this.fileName = fileName;

            this.lineNumber = lineNumber;

            this.content = content;

        }

        

        // Getter方法...

    }

    

    // 生产者：读取文件

    public static class FileReaderProducer implements Runnable {

        private final BlockingQueue<FileRecord> queue;

        private final List<Path> files;

        private volatile boolean running = true;

        

        public FileReaderProducer(BlockingQueue<FileRecord> queue, 

                                 List<Path> files) {

            this.queue = queue;

            this.files = files;

        }

        

        @Override

        public void run() {

            try {

                for (Path file : files) {

                    if (!running) break;

                    

                    try (BufferedReader reader = Files.newBufferedReader(file)) {

                        String line;

                        int lineNumber = 0;

                        

                        while ((line = reader.readLine()) != null) {

                            if (!running) break;

                            

                            FileRecord record = new FileRecord(

                                file.getFileName().toString(),

                                ++lineNumber,

                                line

                            );

                            

                            // 将记录放入队列，如果队列满则等待

                            queue.put(record);

                            

                            // 模拟处理延迟（实际应用中不需要）

                            // Thread.sleep(1);

                        }

                    } catch (IOException e) {

                        System.err.println("读取文件失败: " + file + ", 错误: " + e.getMessage());

                    }

                }

            } catch (InterruptedException e) {

                Thread.currentThread().interrupt();

                System.out.println("生产者线程被中断");

            } finally {

                System.out.println("生产者线程结束");

            }

        }

        

        public void stop() {

            running = false;

        }

    }

    

    // 消费者：处理文件内容

    public static class FileProcessorConsumer implements Runnable {

        private final BlockingQueue<FileRecord> queue;

        private final String name;

        private volatile boolean running = true;

        private int processedCount = 0;

        

        public FileProcessorConsumer(BlockingQueue<FileRecord> queue, 

                                    String name) {

            this.queue = queue;

            this.name = name;

        }

        

        @Override

        public void run() {

            try {

                while (running) {

                    // 从队列获取记录，如果队列空则等待

                    FileRecord record = queue.poll(100, TimeUnit.MILLISECONDS);

                    

                    if (record != null) {

                        processRecord(record);

                        processedCount++;

                    } else if (Thread.interrupted()) {

                        break;

                    }

                }

            } catch (InterruptedException e) {

                Thread.currentThread().interrupt();

                System.out.println(name + " 消费者线程被中断");

            } finally {

                System.out.printf("%s 消费者线程结束，处理了 %d 条记录%n", 

                    name, processedCount);

            }

        }

        

        private void processRecord(FileRecord record) {

            // 实际的文件处理逻辑

            String processedContent = record.getContent().toUpperCase();

            

            // 这里可以进行各种处理：数据分析、格式转换、写入数据库等

            if (processedCount % 1000 == 0) {

                System.out.printf("%s: 处理 %s 第%d行: %s%n",

                    name, record.getFileName(), 

                    record.getLineNumber(),

                    processedContent.substring(0, 

                        Math.min(20, processedContent.length())));

            }

        }

        

        public void stop() {

            running = false;

        }

        

        public int getProcessedCount() {

            return processedCount;

        }

    }

    

    // 主控制器

    public static class FileProcessingController {

        private final BlockingQueue<FileRecord> queue;

        private final ExecutorService executor;

        private final List<FileReaderProducer> producers;

        private final List<FileProcessorConsumer> consumers;

        

        public FileProcessingController(int queueCapacity, 

                                       int producerThreads,

                                       int consumerThreads) {

            this.queue = new LinkedBlockingQueue<>(queueCapacity);

            this.executor = Executors.newFixedThreadPool(

                producerThreads + consumerThreads);

            this.producers = new ArrayList<>();

            this.consumers = new ArrayList<>();

        }

        

        public void processFiles(List<Path> files) {

            // 创建生产者

            FileReaderProducer producer = new FileReaderProducer(queue, files);

            producers.add(producer);

            executor.submit(producer);

            

            // 创建多个消费者

            for (int i = 0; i < 3; i++) {

                FileProcessorConsumer consumer = 

                    new FileProcessorConsumer(queue, "Consumer-" + (i + 1));

                consumers.add(consumer);

                executor.submit(consumer);

            }

        }

        

        public void shutdown() throws InterruptedException {

            // 停止生产者

            for (FileReaderProducer producer : producers) {

                producer.stop();

            }

            

            // 等待队列清空

            while (!queue.isEmpty()) {

                Thread.sleep(100);

            }

            

            // 停止消费者

            for (FileProcessorConsumer consumer : consumers) {

                consumer.stop();

            }

            

            // 关闭线程池

            executor.shutdown();

            boolean terminated = executor.awaitTermination(10, TimeUnit.SECONDS);

            

            if (!terminated) {

                System.out.println("线程池未完全终止，强制关闭");

                executor.shutdownNow();

            }

            

            // 输出统计信息

            int totalProcessed = consumers.stream()

                .mapToInt(FileProcessorConsumer::getProcessedCount)

                .sum();

            System.out.println("总共处理记录数: " + totalProcessed);

        }

    }

    

    // 使用示例

    public static void main(String[] args) throws Exception {

        // 获取要处理的文件列表

        List<Path> files = Files.list(Paths.get("input"))

            .filter(Files::isRegularFile)

            .filter(p -> p.toString().endsWith(".txt"))

            .limit(10)

            .toList();

        

        if (files.isEmpty()) {

            System.out.println("未找到.txt文件");

            return;

        }

        

        System.out.println("找到 " + files.size() + " 个文件需要处理");

        

        // 创建控制器并处理文件

        FileProcessingController controller = 

            new FileProcessingController(1000, 1, 3);

        

        try {

            controller.processFiles(files);

            Thread.sleep(5000); // 模拟处理时间

        } finally {

            controller.shutdown();

        }

    }

}

```

## 高性能文件读取方案

### 使用内存映射文件（MappedByteBuffer）

```java

import java.io.*;

import java.nio.*;

import java.nio.channels.*;

import java.nio.file.*;

import java.util.concurrent.*;

import java.util.concurrent.atomic.*;

public class MappedFileProcessor {

    

    /**

     * 使用内存映射文件进行多线程读取

     */

    public static class MappedFileReader {

        private final Path filePath;

        private final int threadCount;

        private final ExecutorService executor;

        

        public MappedFileReader(Path filePath, int threadCount) {

            this.filePath = filePath;

            this.threadCount = threadCount;

            this.executor = Executors.newFixedThreadPool(threadCount);

        }

        

        public void processFile() throws IOException, InterruptedException {

            long fileSize = Files.size(filePath);

            System.out.printf("文件: %s, 大小: %d bytes%n", 

                filePath.getFileName(), fileSize);

            

            // 打开文件通道

            try (FileChannel channel = FileChannel.open(filePath, 

                StandardOpenOption.READ)) {

                

                // 计算每个线程处理的数据块大小

                long chunkSize = fileSize / threadCount;

                // 确保块边界对齐到行边界（简单实现：追加到下一个换行符）

                chunkSize = alignToLineBoundary(channel, chunkSize);

                

                List<Future<ProcessResult>> futures = new ArrayList<>();

                long startPosition = 0;

                

                // 创建处理任务

                for (int i = 0; i < threadCount && startPosition < fileSize; i++) {

                    long endPosition = Math.min(startPosition + chunkSize, fileSize);

                    

                    // 如果是最后一个线程，处理到文件末尾

                    if (i == threadCount - 1) {

                        endPosition = fileSize;

                    } else {

                        endPosition = alignToLineBoundary(channel, endPosition);

                    }

                    

                    FileChunkTask task = new FileChunkTask(

                        filePath, startPosition, endPosition, i);

                    futures.add(executor.submit(task));

                    

                    startPosition = endPosition;

                }

                

                // 收集结果

                int totalLines = 0;

                int totalWords = 0;

                

                for (Future<ProcessResult> future : futures) {

                    ProcessResult result = future.get();

                    totalLines += result.getLineCount();

                    totalWords += result.getWordCount();

                }

                

                System.out.printf("处理完成: %d 行, %d 单词%n", totalLines, totalWords);

                

            } finally {

                executor.shutdown();

                executor.awaitTermination(10, TimeUnit.SECONDS);

            }

        }

        

        private long alignToLineBoundary(FileChannel channel, long position) 

                throws IOException {

            if (position >= channel.size()) {

                return channel.size();

            }

            

            ByteBuffer buffer = ByteBuffer.allocate(1024);

            channel.read(buffer, position);

            buffer.flip();

            

            // 查找下一个换行符

            while (buffer.hasRemaining()) {

                if (buffer.get() == '\n') {

                    return position + buffer.position();

                }

            }

            

            return position; // 未找到换行符，返回原位置

        }

        

        // 处理结果类

        static class ProcessResult {

            private final int lineCount;

            private final int wordCount;

            

            public ProcessResult(int lineCount, int wordCount) {

                this.lineCount = lineCount;

                this.wordCount = wordCount;

            }

            

            public int getLineCount() { return lineCount; }

            public int getWordCount() { return wordCount; }

        }

        

        // 文件块处理任务

        static class FileChunkTask implements Callable<ProcessResult> {

            private final Path filePath;

            private final long start;

            private final long end;

            private final int taskId;

            

            public FileChunkTask(Path filePath, long start, 

                               long end, int taskId) {

                this.filePath = filePath;

                this.start = start;

                this.end = end;

                this.taskId = taskId;

            }

            

            @Override

            public ProcessResult call() throws IOException {

                int lineCount = 0;

                int wordCount = 0;

                

                try (FileChannel channel = FileChannel.open(filePath, 

                    StandardOpenOption.READ)) {

                    

                    long size = end - start;

                    MappedByteBuffer buffer = channel.map(

                        FileChannel.MapMode.READ_ONLY, start, size);

                    

                    // 处理内存映射区域

                    StringBuilder lineBuilder = new StringBuilder();

                    

                    for (long i = 0; i < size; i++) {

                        char c = (char) buffer.get();

                        

                        if (c == '\n' || i == size - 1) {

                            // 处理完整的一行

                            if (c != '\n') {

                                lineBuilder.append(c);

                            }

                            

                            String line = lineBuilder.toString();

                            if (!line.trim().isEmpty()) {

                                lineCount++;

                                wordCount += countWords(line);

                            }

                            

                            lineBuilder.setLength(0);

                        } else {

                            lineBuilder.append(c);

                        }

                    }

                    

                    System.out.printf("任务 %d: 处理了 %,d bytes, %,d 行%n",

                        taskId, size, lineCount);

                }

                

                return new ProcessResult(lineCount, wordCount);

            }

            

            private int countWords(String line) {

                return line.trim().isEmpty() ? 0 : line.trim().split("\\s+").length;

            }www.kw.zizhitong.cn>www.rz.zizhitong.cn>

        }

    }

    

    // 使用示例

    public static void main(String[] args) {

        Path largeFile = Paths.get("large_file.txt");

        

        try {

            MappedFileReader reader = new MappedFileReader(largeFile, 4);

            reader.processFile();

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}

```

## 文件写入的线程安全方案

### 使用分段写入和锁机制

```java

import java.io.*;

import java.nio.file.*;

import java.util.concurrent.*;

import java.util.concurrent.atomic.*;

import java.util.concurrent.locks.*;

public class ThreadSafeFileWriter {

    

    /**

     * 线程安全的文件写入器（单个文件）

     */

    public static class ConcurrentFileWriter implements AutoCloseable {

        private final Path filePath;

        private final FileChannel fileChannel;

        private final ReentrantLock writeLock;

        private final AtomicLong bytesWritten;

        private final AtomicInteger writeCount;

        

        public ConcurrentFileWriter(Path filePath) throws IOException {

            this.filePath = filePath;

            this.writeLock = new ReentrantLock(true); // 公平锁

            this.bytesWritten = new AtomicLong(0);

            this.writeCount = new AtomicInteger(0);

            

            // 创建或打开文件

            this.fileChannel = FileChannel.open(filePath,

                StandardOpenOption.CREATE,

                StandardOpenOption.WRITE,

                StandardOpenOption.APPEND);

        }

        

        /**

         * 写入数据（线程安全）

         */

        public void write(String data) throws IOException {

            byte[] bytes = data.getBytes();

            

            writeLock.lock();

            try {

                ByteBuffer buffer = ByteBuffer.wrap(bytes);

                int written = fileChannel.write(buffer);

                

                bytesWritten.addAndGet(written);

                writeCount.incrementAndGet();

                

                // 可选的日志记录

                if (writeCount.get() % 1000 == 0) {

                    System.out.printf("已写入 %,d 次, 总共 %,d bytes%n",

                        writeCount.get(), bytesWritten.get());

                }

            } finally {

                writeLock.unlock();

            }

        }

        

        /**

         * 批量写入（更高效）

         */

        public void writeBatch(List<String> dataList) throws IOException {

            writeLock.lock();

            try {

                for (String data : dataList) {

                    byte[] bytes = data.getBytes();

                    ByteBuffer buffer = ByteBuffer.wrap(bytes);

                    fileChannel.write(buffer);

                    

                    bytesWritten.addAndGet(bytes.length);

                }

                writeCount.addAndGet(dataList.size());

            } finally {

                writeLock.unlock();

            }

        }

        

        /**

         * 使用内存映射进行高效写入

         */

        public void writeWithMapping(String data, long position) throws IOException {

            byte[] bytes = data.getBytes();

            

            writeLock.lock();

            try {

                // 确保文件足够大

                long requiredSize = position + bytes.length;

                if (fileChannel.size() < requiredSize) {

                    fileChannel.position(requiredSize - 1);

                    fileChannel.write(ByteBuffer.wrap(new byte[]{0}));

                }

                

                MappedByteBuffer buffer = fileChannel.map(

                    FileChannel.MapMode.READ_WRITE, position, bytes.length);

                buffer.put(bytes);

                

                bytesWritten.addAndGet(bytes.length);

                writeCount.incrementAndGet();

            } finally {

                writeLock.unlock();

            }

        }

        

        /**

         * 获取写入统计信息

         */

        public WriteStats getStats() {

            return new WriteStats(

                writeCount.get(),

                bytesWritten.get(),

                filePath.toString()

            );

        }

        

        @Override

        public void close() throws IOException {

            writeLock.lock();

            try {

                if (fileChannel.isOpen()) {

                    fileChannel.force(true); // 强制写入磁盘

                    fileChannel.close();

                }

            } finally {

                writeLock.unlock();

            }

        }

        

        // 写入统计信息类

        public static class WriteStats {

            private final int writeCount;

            private final long bytesWritten;

            private final String filePath;

            

            public WriteStats(int writeCount, long bytesWritten, String filePath) {

                this.writeCount = writeCount;

                this.bytesWritten = bytesWritten;

                this.filePath = filePath;

            }

            

            @Override

            public String toString() {

                return String.format(

                    "文件: %s, 写入次数: %,d, 总字节数: %,d",

                    filePath, writeCount, bytesWritten

                );

            }

        }

    }

    

    /**

     * 多文件写入管理器

     */

    public static class MultiFileWriterManager {

        private final ConcurrentMap<String, ConcurrentFileWriter> writers;

        private final ExecutorService executor;

        private final String outputDir;

        

        public MultiFileWriterManager(String outputDir, int threadCount) {

            this.writers = new ConcurrentHashMap<>();

            this.executor = Executors.newFixedThreadPool(threadCount);

            this.outputDir = outputDir;

            

            // 确保输出目录存在

            try {

                Files.createDirectories(Paths.get(outputDir));

            } catch (IOException e) {

                throw new RuntimeException("无法创建输出目录: " + outputDir, e);

            }

        }

        

        /**

         * 异步写入文件

         */

        public CompletableFuture<Void> writeAsync(String fileName, String content) {

            return CompletableFuture.runAsync(() -> {

                try {

                    writeSync(fileName, content);

                } catch (IOException e) {

                    throw new CompletionException(e);

                }

            }, executor);

        }

        

        /**

         * 同步写入文件（线程安全）

         */

        public synchronized void writeSync(String fileName, String content) 

                throws IOException {

            ConcurrentFileWriter writer = writers.computeIfAbsent(

                fileName, key -> {

                    try {

                        Path filePath = Paths.get(outputDir, fileName);

                        return new ConcurrentFileWriter(filePath);

                    } catch (IOException e) {

                        throw new RuntimeException("无法创建文件写入器: " + fileName, e);

                    }

                });

            

            writer.write(content);

        }

        

        /**

         * 批量异步写入

         */

        public CompletableFuture<Void> writeBatchAsync(

                String fileName, List<String> contents) {

            return CompletableFuture.runAsync(() -> {

                try {

                    ConcurrentFileWriter writer = writers.get(fileName);

                    if (writer == null) {

                        writer = new ConcurrentFileWriter(

                            Paths.get(outputDir, fileName));

                        writers.put(fileName, writer);

                    }

                    writer.writeBatch(contents);

                } catch (IOException e) {

                    throw new CompletionException(e);

                }

            }, executor);

        }

        

        /**

         * 关闭所有写入器

         */

        public void shutdown() throws IOException {

            executor.shutdown();

            

            try {

                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {

                    executor.shutdownNow();

                }

            } catch (InterruptedException e) {

                executor.shutdownNow();

                Thread.currentThread().interrupt();

            }

            

            // 关闭所有文件写入器

            for (ConcurrentFileWriter writer : writers.values()) {

                try {

                    writer.close();

                } catch (IOException e) {

                    System.err.println("关闭写入器时出错: " + e.getMessage());

                }

            }

            

            // 输出统计信息

            System.out.println("写入统计:");

            writers.forEach((fileName, writer) -> {

                System.out.println(writer.getStats());

            });

        }

    }

    

    // 使用示例

    public static void main(String[] args) {

        MultiFileWriterManager manager = 

            new MultiFileWriterManager("output", 4);

        

        try {

            // 模拟并发写入

            List<CompletableFuture<Void>> futures = new ArrayList<>();

            

            for (int i = 0; i < 100; i++) {

                String fileName = "file_" + (i % 5) + ".txt";

                String content = "数据行 " + i + " - " + 

                    System.currentTimeMillis() + "\n";

                

                futures.add(manager.writeAsync(fileName, content));

            }

            

            // 等待所有写入完成

            CompletableFuture.allOf(

                futures.toArray(new CompletableFuture[0])

            ).join();

            

            System.out.println("所有写入任务完成");

            

        } finally {

            try {

                manager.shutdown();

            } catch (IOException e) {

                e.printStackTrace();

            }

        }

    }

}

```

## 文件加密/解密的并行处理

```java

import javax.crypto.*;

import javax.crypto.spec.*;

import java.io.*;

import java.nio.file.*;

import java.security.*;

import java.util.concurrent.*;

import java.util.concurrent.atomic.*;

public class ParallelFileEncryptor {

    

    private static final String ALGORITHM = "AES";

    private static final String TRANSFORMATION = "AES/CBC/PKCS5Padding";

    private static final int BUFFER_SIZE = 8192;

    

    /**

     * 并行文件加密处理器

     */

    public static class ParallelFileProcessor {

        private final ExecutorService executor;

        private final AtomicInteger processedFiles;

        private final AtomicLong processedBytes;

        private final SecretKey secretKey;

        private final byte[] iv;

        

        public ParallelFileProcessor(int threadCount) 

                throws NoSuchAlgorithmException {

            this.executor = Executors.newFixedThreadPool(threadCount);

            this.processedFiles = new AtomicInteger(0);

            this.processedBytes = new AtomicLong(0);

            

            // 生成加密密钥（实际应用中应从安全的地方获取）

            KeyGenerator keyGen = KeyGenerator.getInstance(ALGORITHM);

            keyGen.init(128);

            this.secretKey = keyGen.generateKey();

            this.iv = new byte[16]; // 初始化向量

            new SecureRandom().nextBytes(iv);

        }www.mv.zizhitong.cn>www.bn.zizhitong.cn

        

        /**

         * 并行处理多个文件

         */

        public void processFiles(List<Path> inputFiles, Path outputDir, 

                               boolean encrypt) 

                throws IOException, InterruptedException {

            

            // 创建输出目录

            Files.createDirectories(outputDir);

            

            List<Future<ProcessResult>> futures = new ArrayList<>();

            

            // 提交处理任务

            for (Path inputFile : inputFiles) {

                Path outputFile = outputDir.resolve(

                    inputFile.getFileName() + 

                    (encrypt ? ".encrypted" : ".decrypted"));

                

                FileProcessorTask task = new FileProcessorTask(

                    inputFile, outputFile, encrypt, 

                    secretKey, iv,

                    processedFiles, processedBytes);

                

                futures.add(executor.submit(task));

            }

            

            // 等待所有任务完成

            for (Future<ProcessResult> future : futures) {

                try {

                    ProcessResult result = future.get();

                    System.out.printf("处理完成: %s -> %s (%,d bytes)%n",

                        result.getInputFile(),

                        result.getOutputFile(),

                        result.getFileSize());

                } catch (ExecutionException e) {

                    System.err.println("处理失败: " + e.getCause().getMessage());

                }

            }

            

            System.out.printf("总共处理了 %d 个文件, %,d bytes%n",

                processedFiles.get(), processedBytes.get());

        }

        

        public void shutdown() {

            executor.shutdown();

            try {

                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {

                    executor.shutdownNow();

                }

            } catch (InterruptedException e) {

                executor.shutdownNow();

                Thread.currentThread().interrupt();

            }

        }

        

        // 处理结果类

        static class ProcessResult {

            private final String inputFile;

            private final String outputFile;

            private final long fileSize;

            

            public ProcessResult(String inputFile, String outputFile, long fileSize) {

                this.inputFile = inputFile;

                this.outputFile = outputFile;

                this.fileSize = fileSize;

            }

            

            public String getInputFile() { return inputFile; }

            public String getOutputFile() { return outputFile; }

            public long getFileSize() { return fileSize; }

        }

        

        // 文件处理任务

        static class FileProcessorTask implements Callable<ProcessResult> {

            private final Path inputFile;

            private final Path outputFile;

            private final boolean encrypt;

            private final SecretKey secretKey;

            private final byte[] iv;

            private final AtomicInteger fileCounter;

            private final AtomicLong byteCounter;

            

            public FileProcessorTask(Path inputFile, Path outputFile, 

                                   boolean encrypt, SecretKey secretKey, 

                                   byte[] iv, AtomicInteger fileCounter,

                                   AtomicLong byteCounter) {

                this.inputFile = inputFile;

                this.outputFile = outputFile;

                this.encrypt = encrypt;

                this.secretKey = secretKey;

                this.iv = iv;

                this.fileCounter = fileCounter;

                this.byteCounter = byteCounter;

            }

            

            @Override

            public ProcessResult call() throws Exception {

                long fileSize = Files.size(inputFile);

                

                if (encrypt) {

                    encryptFile(inputFile, outputFile, fileSize);

                } else {

                    decryptFile(inputFile, outputFile, fileSize);

                }www.es.zizhitong.cn>www.px.zizhitong.cn

                

                fileCounter.incrementAndGet();

                byteCounter.addAndGet(fileSize);

                

                return new ProcessResult(

                    inputFile.getFileName().toString(),

                    outputFile.getFileName().toString(),

                    fileSize

                );

            }

            

            private void encryptFile(Path input, Path output, long fileSize) 

                    throws Exception {

                Cipher cipher = Cipher.getInstance(TRANSFORMATION);

                cipher.init(Cipher.ENCRYPT_MODE, secretKey, new IvParameterSpec(iv));

                

                try (InputStream in = Files.newInputStream(input);

                     OutputStream out = Files.newOutputStream(output);

                     CipherOutputStream cipherOut = new CipherOutputStream(out, cipher)) {

                    

                    byte[] buffer = new byte[BUFFER_SIZE];

                    int bytesRead;

                    long totalRead = 0;

                    

                    while ((bytesRead = in.read(buffer)) != -1) {

                        cipherOut.write(buffer, 0, bytesRead);

                        totalRead += bytesRead;

                        

                        // 显示进度（每10%）

                        if (fileSize > 0 && totalRead % (fileSize / 10) == 0) {

                            int percent = (int) ((totalRead * 100) / fileSize);

                            System.out.printf("加密 %s: %d%%%n", 

                                input.getFileName(), percent);

                        }

                    }

                }

            }

            

            private void decryptFile(Path input, Path output, long fileSize) 

                    throws Exception {

                Cipher cipher = Cipher.getInstance(TRANSFORMATION);

                cipher.init(Cipher.DECRYPT_MODE, secretKey, new IvParameterSpec(iv));

                

                try (InputStream in = Files.newInputStream(input);

                     CipherInputStream cipherIn = new CipherInputStream(in, cipher);

                     OutputStream out = Files.newOutputStream(output)) {

                    

                    byte[] buffer = new byte[BUFFER_SIZE];

                    int bytesRead;

                    long totalRead = 0;

                    

                    while ((bytesRead = cipherIn.read(buffer)) != -1) {

                        out.write(buffer, 0, bytesRead);

                        totalRead += bytesRead;

                        

                        // 显示进度

                        if (fileSize > 0 && totalRead % (fileSize / 10) == 0) {

                            int percent = (int) ((totalRead * 100) / fileSize);

                            System.out.printf("解密 %s: %d%%%n", 

                                input.getFileName(), percent);

                        }

                    }

                }

            }

        }

    }

    

    // 使用示例

    public static void main(String[] args) {

        try {

            // 获取要处理的文件

            List<Path> files = Files.list(Paths.get("data"))

                .filter(Files::isRegularFile)

                .filter(p -> p.toString().endsWith(".txt"))

                .limit(5)

                .toList();

            

            if (files.isEmpty()) {

                System.out.println("未找到文件");

                return;

            }

            

            System.out.println("找到 " + files.size() + " 个文件");

            

            // 创建并行处理器

            ParallelFileProcessor processor = new ParallelFileProcessor(3);

            

            try {

                // 加密文件

                System.out.println("开始加密文件...");

                processor.processFiles(files, Paths.get("encrypted"), true);

                

                // 获取加密后的文件

                List<Path> encryptedFiles = Files.list(Paths.get("encrypted"))

                    .filter(Files::isRegularFile)

                    .toList();

                

                // 解密文件

                System.out.println("\n开始解密文件...");

                processor.processFiles(encryptedFiles, Paths.get("decrypted"), false);

                

            } finally {

                processor.shutdown();

            }

            

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}

```

## 性能优化与监控

```java

import java.io.*;

import java.nio.file.*;

import java.util.concurrent.*;

import java.util.concurrent.atomic.*;

import java.util.concurrent.locks.*;

import java.time.*;

public class PerformanceOptimizedFileProcessor {

    

    /**

     * 带性能监控的文件处理器

     */

    public static class MonitoredFileProcessor {

        private final ExecutorService executor;

        private final PerformanceMonitor monitor;

        private final int batchSize;

        

        public MonitoredFileProcessor(int threadCount, int batchSize) {

            this.executor = new ThreadPoolExecutor(

                threadCount, threadCount,

                0L, TimeUnit.MILLISECONDS,

                new LinkedBlockingQueue<>(1000),

                new ThreadFactory() {

                    private final AtomicInteger counter = new AtomicInteger(0);

                    

                    @Override

                    public Thread newThread(Runnable r) {

                        Thread thread = new Thread(r, 

                            "file-processor-" + counter.incrementAndGet());

                        thread.setDaemon(true);

                        return thread;

                    }

                },

                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略

            );

            

            this.monitor = new PerformanceMonitor();

            this.batchSize = batchSize;

        }

        

        /**

         * 处理文件目录

         */

        public void processDirectory(Path directory, String filePattern) 

                throws IOException, InterruptedException {

            

            monitor.start();

            

            // 查找文件

            try (DirectoryStream<Path> stream = 

                 Files.newDirectoryStream(directory, filePattern)) {

                

                List<Path> fileList = new ArrayList<>();

                for (Path file : stream) {

                    if (Files.isRegularFile(file)) {

                        fileList.add(file);

                    }

                }

                

                System.out.printf("找到 %d 个文件需要处理%n", fileList.size());

                

                // 分批处理

                List<CompletableFuture<Void>> futures = new ArrayList<>();

                List<Path> batch = new ArrayList<>();

                

                for (Path file : fileList) {

                    batch.add(file);

                    

                    if (batch.size() >= batchSize) {

                        futures.add(processBatch(new ArrayList<>(batch)));

                        batch.clear();

                    }

                }

                

                // 处理最后一批

                if (!batch.isEmpty()) {

                    futures.add(processBatch(batch));

                }

                

                // 等待所有批处理完成

                CompletableFuture.allOf(

                    futures.toArray(new CompletableFuture[0])

                ).join();

                

            } finally {

                monitor.stop();

                monitor.printReport();

            }

        }

        

        private CompletableFuture<Void> processBatch(List<Path> batch) {

            return CompletableFuture.runAsync(() -> {

                for (Path file : batch) {

                    try {

                        processFile(file);

                        monitor.incrementProcessedFiles();

                    } catch (IOException e) {

                        monitor.incrementFailedFiles();

                        System.err.println("处理文件失败: " + file + ", 错误: " + e.getMessage());

                    }

                }

            }, executor);

        }

        

        private void processFile(Path file) throws IOException {

            long startTime = System.nanoTime();

            long fileSize = Files.size(file);

            

            // 实际的文件处理逻辑

            try (BufferedReader reader = Files.newBufferedReader(file)) {

                String line;

                int lineCount = 0;

                

                while ((line = reader.readLine()) != null) {

                    // 模拟处理

                    String processed = processLine(line);

                    lineCount++;

                }

                

                monitor.addProcessedBytes(fileSize);

                monitor.addProcessingTime(System.nanoTime() - startTime);

                

                if (lineCount % 1000 == 0) {

                    System.out.printf("处理文件: %s (%,d 行)%n", 

                        file.getFileName(), lineCount);

                }

            }

        }

        

        private String processLine(String line) {

            // 实际的处理逻辑

            return line.toUpperCase();

        }

        

        public void shutdown() {

            executor.shutdown();

            try {

                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {

                    executor.shutdownNow();

                }

            } catch (InterruptedException e) {

                executor.shutdownNow();

                Thread.currentThread().interrupt();

            }

        }

        

        /**

         * 性能监控器

         */

        static class PerformanceMonitor {

            private final AtomicLong processedFiles = new AtomicLong();

            private final AtomicLong failedFiles = new AtomicLong();

            private final AtomicLong processedBytes = new AtomicLong();

            private final AtomicLong totalProcessingTime = new AtomicLong(); // 纳秒

            private final AtomicInteger activeThreads = new AtomicInteger();

            private final LongAdder totalLines = new LongAdder();

            

            private Instant startTime;

            private Instant endTime;

            

            public void start() {

                this.startTime = Instant.now();

            }

            

            public void stop() {

                this.endTime = Instant.now();

            }

            

            public void incrementProcessedFiles() {

                processedFiles.incrementAndGet();

            }

            

            public void incrementFailedFiles() {

                failedFiles.incrementAndGet();

            }

            

            public void addProcessedBytes(long bytes) {

                processedBytes.addAndGet(bytes);

            }

            

            public void addProcessingTime(long nanos) {

                totalProcessingTime.addAndGet(nanos);

            }

            

            public void incrementActiveThreads() {

                activeThreads.incrementAndGet();

            }

            

            public void decrementActiveThreads() {

                activeThreads.decrementAndGet();

            }

            

            public void addLines(int lines) {

                totalLines.add(lines);

            }

            

            public void printReport() {

                if (startTime == null || endTime == null) {

                    System.out.println("监控未启动或未停止");

                    return;

                }

                

                Duration duration = Duration.between(startTime, endTime);

                long totalSeconds = duration.getSeconds();

                long filesProcessed = processedFiles.get();

                long bytesProcessed = processedBytes.get();

                long totalNanos = totalProcessingTime.get();

                

                System.out.println("\n=== 性能报告 ===");

                System.out.printf("总时间: %d 秒%n", totalSeconds);

                System.out.printf("处理文件数: %,d%n", filesProcessed);

                System.out.printf("失败文件数: %,d%n", failedFiles.get());

                System.out.printf("处理总字节数: %,d%n", bytesProcessed);

                System.out.printf("处理总行数: %,d%n", totalLines.sum());

                

                if (filesProcessed > 0) {

                    System.out.printf("平均文件大小: %,d bytes%n", 

                        bytesProcessed / filesProcessed);

                    System.out.printf("平均处理时间: %,d ms/文件%n", 

                        (totalNanos / filesProcessed) / 1_000_000);

                }

                

                if (totalSeconds > 0) {

                    double throughput = (double) bytesProcessed / totalSeconds / 1024 / 1024;

                    System.out.printf("吞吐量: %.2f MB/秒%n", throughput);

                }

                

                System.out.printf("平均处理速度: %,d bytes/秒%n", 

                    totalSeconds > 0 ? bytesProcessed / totalSeconds : 0);

            }

        }

    }

    

    /**

     * 带流量控制的文件处理器

     */

    public static class RateLimitedFileProcessor {

        private final ExecutorService executor;

        private final RateLimiter rateLimiter;

        private final Semaphore concurrentLimit;

        

        public RateLimitedFileProcessor(int threadCount, 

                                       int maxConcurrentFiles,

                                       double filesPerSecond) {

            this.executor = Executors.newFixedThreadPool(threadCount);

            this.rate