Xử lý tệp song song: ForkJoin, Parallel Streams

1. Stream song song (parallelStream): đơn giản và tiện lợi

Nếu bạn đã làm việc với Stream API, bạn biết việc lọc, biến đổi và thu thập các collection thuận tiện như thế nào. Điểm cộng quan trọng: bất kỳ stream nào cũng có thể chuyển sang song song chỉ với một dòng — bật parallel(), và các phần tử của collection sẽ được xử lý đồng thời.

Điều này đặc biệt hữu ích cho các thao tác độc lập trên tập tệp: đếm dòng, tìm chuỗi con, sao chép, nén, v.v.

Ví dụ: đếm dòng song song trong tất cả tệp của một thư mục

Cách 1: tuần tự

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("Tổng số dòng trong tất cả các log: " + totalLines);
    }
}

Ghi chú: mọi thứ được thực hiện lần lượt, từng tệp một. Nếu tệp nhiều và lớn, bạn sẽ phải chờ lâu.

Cách 2: song song!

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() // day la 'phep mau'!
                .mapToLong(file -> {
                    try (Stream<String> lines = Files.lines(file)) {
                        return lines.count();
                    } catch (IOException e) {
                        e.printStackTrace();
                        return 0L;
                    }
                })
                .sum();
            System.out.println("Tổng số dòng trong tất cả các log: " + totalLines);
        }
    }
}

Ghi chú: dòng mấu chốt — .parallel(). Trên bộ xử lý đa nhân, chương trình thường chạy nhanh hơn đáng kể.

Hoạt động như thế nào?

• parallel() biến một stream thường thành stream song song. Bên dưới sử dụng common ForkJoinPool (số luồng mặc định bằng số lõi).
• Mỗi tệp được xử lý độc lập, kết quả được gộp qua các phép toán kết thúc (ví dụ, sum()).
• Nếu số tệp ít — có thể không có tăng tốc; nếu hàng trăm — lợi ích thường rõ rệt.

Quan trọng!

• Stream song song không tự làm nhanh các thao tác I/O; chúng cho phép thực hiện nhiều thao tác đồng thời. Trên thiết bị nhanh (SSD) điều này giúp ích, trên thiết bị chậm (HDD) bạn có thể chạm trần băng thông đĩa.

2. ForkJoinPool: “chia để trị” trong thực tế

ForkJoin — framework cho tính toán song song theo mô hình “chia để trị”: chia một bài toán lớn thành các bài toán con, thực thi chúng song song và hợp nhất kết quả. Việc này do một pool đặc biệt quản lý — ForkJoinPool. Chính pool này được dùng “hậu trường” của stream song song, nhưng bạn cũng có thể quản lý trực tiếp để linh hoạt hơn.

Mô hình này đặc biệt phù hợp với cấu trúc đệ quy (cây thư mục), mảng dữ liệu lớn và các tác vụ dễ phân rã thành những phần độc lập.

Ví dụ: tìm kiếm đệ quy trong cây thư mục

Tìm tất cả tệp ".txt" (bao gồm cả thư mục lồng nhau) và đếm tổng số dòng.

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(); // Khoi chay tac vu con
                    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();
        }
        // Thu thap ket qua tu cac tac vu con
        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("Tổng số dòng trong tất cả tệp .txt: " + totalLines);
    }
}

Diễn giải:

• Đối với mỗi thư mục, tạo một tác vụ riêng (FolderLineCounter), với thư mục con — các tác vụ con riêng (fork()).
• Các tệp được đếm tại chỗ, kết quả được cộng sau khi join() tất cả tác vụ con.

Ưu điểm của ForkJoin là gì?

• Hoạt động hiệu quả với các cây phân cấp lớn (cây thư mục).
• Tận dụng tối đa các lõi CPU.
• Cho phép kiểm soát chính xác việc song song hóa và ranh giới tác vụ.

3. Các kịch bản sử dụng thực tế

Xử lý tệp hàng loạt

Ví dụ, cần sao chép hàng nghìn ảnh vào thư mục dự phòng.

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("Tất cả tệp đã được sao chép!");
    }
}

Ghi chú: mỗi tệp được sao chép trong một luồng riêng. Khi số lượng tệp lớn, tăng tốc sẽ rõ rệt.

Nén/giải nén song song

Tương tự, bạn có thể song song hóa việc nén, tính lại hash, chuyển đổi định dạng ảnh, v.v. qua parallelStream() hoặc ForkJoinPool riêng.

4. Lưu ý và hạn chế quan trọng

• Các thao tác I/O không phải lúc nào cũng được lợi từ tính song song. Nếu đĩa hoặc mạng là “điểm nghẽn”, hàng trăm tác vụ song song chỉ làm tăng cạnh tranh tài nguyên.
• Đừng khởi chạy quá nhiều luồng. Mặc định, stream song song dùng common pool ForkJoinPool.commonPool() với mức song song ≈ số lõi. Có thể thay đổi qua thuộc tính "java.util.concurrent.ForkJoinPool.common.parallelism", nhưng hãy làm có chủ đích.
• Đừng quên đồng bộ. Nếu nhiều luồng ghi vào cùng một tệp/đối tượng — hãy dùng đồng bộ và hàng đợi; với các tệp độc lập — không cần đồng bộ.

5. Làm quen nhanh với FileChannel và truy cập theo vị trí

Đối với các kịch bản nâng cao (ví dụ, đọc song song các phần khác nhau của một tệp lớn) hãy dùng java.nio.channels.FileChannel, hỗ trợ đọc/ghi theo vị trí.

Ví dụ: đọc các phần khác nhau của tệp ở các luồng khác nhau

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);
            // Xu ly du lieu
            System.out.println("Đã đọc " + buffer.position() + " byte từ vị trí " + position);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Ghi chú: hãy chạy nhiều tác vụ như vậy — mỗi tác vụ đọc một vùng riêng. Nhưng cẩn thận: không phải mọi ổ đĩa và hệ thống tệp đều “ưa thích” tải song song cao.

6. Các lỗi thường gặp khi xử lý tệp song song

Lỗi số 1: ghi song song vào cùng một tệp mà không đồng bộ. Dữ liệu bị trộn lẫn và hỏng. Hãy dùng hàng đợi, đệm và đồng bộ hóa ghi.

Lỗi số 2: quá nhiều song song. Stream song song trên phần cứng yếu/tệp nhỏ gây chi phí phụ và có thể làm chậm thực thi.

Lỗi số 3: bỏ qua lỗi I/O. Trong stream song song, ngoại lệ rất dễ bị “bỏ lỡ” — hãy xử lý chúng bên trong lambda, ghi log và tính đến các lỗi.

Lỗi số 4: không đóng tài nguyên. Luôn dùng try-with-resources cho stream/kênh; nếu không, bạn sẽ gặp rò rỉ và lỗi khó hiểu.

Lỗi số 5: trông chờ “phép màu” từ parallel(). Tính song song chỉ tăng tốc khi khối lượng công việc đủ lớn và tài nguyên sẵn sàng (CPU, ổ đĩa nhanh). Bản thân lời gọi parallel() không phải “viên đạn bạc”.