ForkJoinPool và RecursiveTask: các tác vụ đệ quy

1. ForkJoinPool: đó là gì và dùng để làm gì

ForkJoinPool là một pool luồng đặc biệt, hiện thực cách tiếp cận “chia để trị” (divide and conquer). Mục tiêu của nó là phân song song công việc một cách hiệu quả nhất khi có thể tách một tác vụ lớn thành nhiều tác vụ con độc lập, thực thi song song và sau đó hợp nhất kết quả.

• Fork (chia) — tác vụ được tách thành các tác vụ con.
• Join (hợp nhất) — kết quả của các tác vụ con được gom thành kết quả cuối.

ForkJoinPool là trái tim của parallel streams trong Java: khi bạn viết list.parallelStream(), bên trong chính là nó. Nhưng bạn cũng có thể dùng trực tiếp để có nhiều quyền kiểm soát hơn.

Khi nào ForkJoinPool đặc biệt hữu ích

ForkJoinPool phát huy tác dụng trên những tác vụ dễ dàng chia thành các phần độc lập: ví dụ xử lý các mảng rất lớn, khi mỗi đoạn được xử lý riêng, rồi hợp nhất kết quả.

• Tác vụ dễ chia thành các tác vụ con độc lập: sắp xếp, tìm kiếm, tính tổng.
• Các tác vụ con có khối lượng xấp xỉ nhau và không phụ thuộc lẫn nhau.
• Cần tận dụng tất cả lõi CPU để đạt tốc độ tối đa.

+---------------------+
|     Tac vu lon      |
+---------------------+
          |
          v
+---------+---------+
|  Tac vu con 1     |
|  Tac vu con 2     |
|  ...              |
+-------------------+
          |
          v
+---------+---------+
|     Ket qua        |
+-------------------+

Đó chính là cách “chia để trị” hoạt động: chia nhỏ — tính song song — hợp nhất.

2. RecursiveTask và RecursiveAction: hai mặt của cùng một vấn đề

Trong ForkJoinPool, các tác vụ được tạo thông qua các lớp chuyên dụng có khả năng chia nhỏ thành tác vụ con và hợp nhất kết quả. RecursiveTask<T> trả về kết quả, còn RecursiveAction thì không. Trên thực tế, RecursiveTask thường được dùng hơn, chẳng hạn để trả về tổng, giá trị lớn nhất hoặc số lượng.

Để tạo một tác vụ như vậy, hãy kế thừa và hiện thực phương thức compute(). Trong đó mô tả logic: nếu tác vụ nhỏ — giải ngay; nếu lớn — chia thành các tác vụ con, khởi chạy chúng song song bằng fork() và hợp nhất kết quả bằng join(). Từ đó hình thành song song đệ quy một cách tự nhiên.

3. Cú pháp và ví dụ: tính tổng mảng song song

Giả sử chúng ta có một mảng số lớn và muốn nhanh chóng tính tổng tất cả phần tử.

Bước 1. Lớp tác vụ

import java.util.concurrent.RecursiveTask;

public class ArraySumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 1_000; // Ngưỡng để chia tác vụ
    private final int[] array;
    private final int start, end;

    public ArraySumTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        // Nếu tác vụ nhỏ — tính trực tiếp
        if (end - start <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += array[i];
            }
            return sum;
        } else {
            // Chia tác vụ thành hai tác vụ con
            int mid = (start + end) / 2;
            ArraySumTask leftTask = new ArraySumTask(array, start, mid);
            ArraySumTask rightTask = new ArraySumTask(array, mid, end);

            // Khởi chạy các tác vụ con song song
            leftTask.fork(); // Bất đồng bộ
            long rightResult = rightTask.compute(); // Đồng bộ
            long leftResult = leftTask.join(); // Chờ tác vụ bên trái hoàn thành

            // Hợp nhất kết quả
            return leftResult + rightResult;
        }
    }
}

• Nếu tác vụ nhỏ (nhỏ hơn ngưỡng THRESHOLD) — tính tổng bằng vòng lặp thông thường.
• Nếu lớn — chia làm hai; một cái khởi chạy bất đồng bộ bằng fork(), cái còn lại tính đồng bộ bằng compute(), sau đó hợp nhất qua join().

Bước 2. Chạy tác vụ qua ForkJoinPool

import java.util.concurrent.ForkJoinPool;

public class ForkJoinSumDemo {
    public static void main(String[] args) {
        int[] numbers = new int[10_000_000];
        for (int i = 0; i < numbers.length; i++) {
            numbers[i] = 1; // Đơn giản — tổng phải bằng độ dài mảng
        }

        ForkJoinPool pool = new ForkJoinPool(); // Mặc định — theo số lõi

        ArraySumTask task = new ArraySumTask(numbers, 0, numbers.length);

        long result = pool.invoke(task); // Chạy tác vụ

        System.out.println("Tổng các phần tử của mảng: " + result);
    }
}

Nó hoạt động như thế nào?

• ForkJoinPool tự quyết định dùng bao nhiêu luồng (thường — theo số lõi CPU).
• Tác vụ được tự động chia thành các tác vụ con, mỗi cái có thể chạy trên một lõi riêng.
• Hiệu năng thường cao hơn so với mã tuần tự (đặc biệt trên dữ liệu lớn và hệ thống đa lõi).

4. ForkJoinPool hoạt động thế nào: đôi chút “hậu trường”

Work-Stealing

ForkJoinPool hiện thực cơ chế “work-stealing”: nếu một luồng hết việc, nó “lấy” công việc từ luồng khác. Điều này giúp cân bằng tải hiệu quả và tận dụng tối đa các lõi.

Thuật toán cơ bản

• Tác vụ chính được chia thành các tác vụ con.
• Các tác vụ con được đưa vào các hàng đợi chuyên biệt.
• Các luồng lấy việc từ hàng đợi của mình, và khi rỗng — “tìm” việc ở hàng xóm.
• Khi mọi thứ hoàn tất, kết quả được hợp nhất.

Sơ đồ hoạt động

flowchart TD
    A[Nhiệm vụ chính] --> B1[Nhiệm vụ con 1]
    A --> B2[Nhiệm vụ con 2]
    B1 --> C1[Nhiệm vụ nhỏ 1]
    B1 --> C2[Nhiệm vụ nhỏ 2]
    B2 --> C3[Nhiệm vụ nhỏ 3]
    B2 --> C4[Nhiệm vụ nhỏ 4]
    C1 --> D[Hợp nhất kết quả]
    C2 --> D
    C3 --> D
    C4 --> D

5. RecursiveAction — khi không cần trả về kết quả

Nếu chỉ cần thực thi song song mà không trả về kết quả, hãy dùng RecursiveAction. Ví dụ điển hình — song song hóa việc điền mảng, in ấn, sắp xếp “tại chỗ”, v.v.

import java.util.concurrent.RecursiveAction;

public class PrintTask extends RecursiveAction {
    private static final int THRESHOLD = 100;
    private final int[] array;
    private final int start, end;

    public PrintTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected void compute() {
        if (end - start <= THRESHOLD) {
            for (int i = start; i < end; i++) {
                System.out.print(array[i] + " ");
            }
        } else {
            int mid = (start + end) / 2;
            invokeAll(
                new PrintTask(array, start, mid),
                new PrintTask(array, mid, end)
            );
        }
    }
}

6. Thực hành: tìm giá trị lớn nhất trong mảng theo phương pháp song song

import java.util.concurrent.RecursiveTask;

public class MaxFindTask extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 1000;
    private final int[] array;
    private final int start, end;

    public MaxFindTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        if (end - start <= THRESHOLD) {
            int max = array[start];
            for (int i = start + 1; i < end; i++) {
                if (array[i] > max) max = array[i];
            }
            return max;
        } else {
            int mid = (start + end) / 2;
            MaxFindTask left = new MaxFindTask(array, start, mid);
            MaxFindTask right = new MaxFindTask(array, mid, end);
            left.fork();
            int rightResult = right.compute();
            int leftResult = left.join();
            return Math.max(leftResult, rightResult);
        }
    }
}

Chạy:

import java.util.concurrent.ForkJoinPool;

public class ForkJoinMaxDemo {
    public static void main(String[] args) {
        int[] array = new int[5_000_000];
        for (int i = 0; i < array.length; i++) {
            array[i] = (int)(Math.random() * 1_000_000);
        }

        ForkJoinPool pool = new ForkJoinPool();
        MaxFindTask task = new MaxFindTask(array, 0, array.length);

        int max = pool.invoke(task);

        System.out.println("Giá trị lớn nhất: " + max);
    }
}

7. Ưu điểm và hạn chế của ForkJoinPool

Ưu điểm

• Cân bằng tải tự động. Work-stealing cho phép sử dụng hiệu quả tất cả các lõi.
• Tiện lợi. Không cần tự tạo và quản lý luồng thủ công.
• Hiệu năng cao. Đặc biệt với tác vụ lớn và hệ thống đa lõi.
• Linh hoạt. Có thể chia tác vụ thành nhiều phần tùy ý.

Hạn chế

• Liên kết chặt giữa các tác vụ con. Nếu các tác vụ con thường xuyên chờ nhau, lợi ích sẽ giảm.
• Tác vụ quá nhỏ. Chi phí chia nhỏ/đồng bộ có thể “ăn” hết lợi thế của song song.
• Tác dụng phụ. Không được thay đổi biến dùng chung nếu không đồng bộ — sẽ dẫn đến race condition.
• Tính phù hợp. Phù hợp với các tác vụ có thể chia thành các phần độc lập.

8. Những lỗi thường gặp khi làm việc với ForkJoinPool và RecursiveTask

Lỗi №1: Chia tác vụ quá nhỏ. Nếu ngưỡng (THRESHOLD) quá thấp, sẽ sinh ra rất nhiều tác vụ li ti — chi phí tạo và đồng bộ sẽ vượt lợi ích của song song. Hãy thử nghiệm ngưỡng: giá trị tối ưu thường ở hàng nghìn hoặc hàng chục nghìn phần tử.

Lỗi №2: Dùng biến chung có thể thay đổi. Nếu các tác vụ con ghi vào biến chung mà không đồng bộ — bạn sẽ gặp race condition. Hãy trả về kết quả qua compute() và chỉ hợp nhất trong join().

Lỗi №3: Dùng fork/join không đúng cách. Quên gọi fork() hoặc join() — và tác vụ con sẽ không chạy song song hoặc kết quả bị “mất”. Hãy chú ý thứ tự lời gọi.

Lỗi №4: Chạy ForkJoinTask ngoài ForkJoinPool. Nếu chỉ gọi compute() trên tác vụ, nó sẽ chạy trong luồng hiện tại, không có song song. Để có “phép màu” thực sự, hãy dùng pool.invoke() hoặc pool.submit().

Lỗi №5: Bỏ qua ngoại lệ. Nếu có ngoại lệ xảy ra trong tác vụ, nó sẽ xuất hiện khi gọi join() hoặc invoke(). Đừng quên xử lý lỗi.

Lỗi №6: Dùng ForkJoinPool cho tác vụ bị chặn. ForkJoinPool không phù hợp với các tác vụ thường xuyên bị chặn (chờ I/O, v.v.). Trong các trường hợp đó, tốt hơn nên sử dụng ExecutorService.