Bài hôm qua kết thúc ở một pool chiếm dụng: ForkJoinPool.commonPool(). Hôm nay là chính nó — thiết kế bên trong, và cách dùng cho đúng loại bài toán mà nó sinh ra để giải.

ForkJoinPool không phải một pool luồng thông thường. Nó được thiết kế cho chia để trị: bài toán tự tách thành bài toán con cùng dạng, cho tới khi đủ nhỏ thì giải thẳng.

Khuôn cơ bản

class Tong extends RecursiveTask<Long> {
    final int tu, den;

    protected Long compute() {
        if (den - tu <= NGUONG) {
            long s = 0;
            for (int i = tu; i < den; i++) s += du[i] * 31L % 7;
            return s;                                   // đủ nhỏ: giải thẳng
        }
        int giua = (tu + den) >>> 1;
        Tong trai = new Tong(tu, giua);
        Tong phai = new Tong(giua, den);
        trai.fork();                                    // đẩy nửa trái cho luồng khác
        long kqPhai = phai.compute();                   // tự làm nửa phải
        return trai.join() + kqPhai;                    // gộp
    }
}

RecursiveTask<T> khi có kết quả trả về, RecursiveAction khi không.

Chú ý (tu + den) >>> 1 chứ không phải (tu + den) / 2. Với mảng rất lớn, phép cộng có thể tràn int thành số âm, và phép chia cho 2 giữ nguyên dấu âm còn dịch bit không dấu thì không. Đây là lỗi từng tồn tại chín năm trong Arrays.binarySearch của chính thư viện chuẩn.

Ngưỡng chia nhỏ: chỗ dễ sai nhất

Đây là câu hỏi mà tài liệu thường trả lời rất mơ hồ. Tôi đo bằng mảng 50 triệu phần tử trên máy 16 nhân:

  tuần tự một luồng : 28 ms

  ngưỡng          thời gian   nhanh hơn       số tác vụ
  100                 52 ms      0,54 lần        999.999
  1.000                6 ms      4,67 lần         99.999
  10.000               3 ms      9,33 lần          9.999
  100.000              2 ms     14,00 lần            999
  1.000.000            2 ms     14,00 lần             99
  10.000.000           3 ms      9,33 lần              9
  50.000.000          27 ms      1,04 lần              1

Đường cong hình chữ U, và hai đầu đều tệ.

Ngưỡng 100 chậm hơn cả chạy tuần tự — 52 ms so với 28 ms. Một triệu tác vụ, mỗi tác vụ chỉ cộng 100 số. Chi phí tạo đối tượng, đẩy vào hàng đợi, đánh thức luồng, gộp kết quả lớn hơn hẳn công việc thật.

Ngưỡng bằng cả mảng thì không có song song, đúng bằng chạy tuần tự.

Khoảng tốt nằm ở giữa và khá rộng: từ 100.000 tới 1.000.000 đều cho 14 lần, tức là 99 tới 999 tác vụ. Đây là tin tốt — bạn không cần tinh chỉnh, chỉ cần đừng chọn sai một bậc độ lớn.

Quy tắc tôi rút ra: nhắm khoảng vài trăm tới vài nghìn tác vụ lá, đủ nhiều để ăn trộm việc cân bằng được tải, đủ ít để chi phí quản lý không đáng kể. Với 16 nhân thì vài trăm là quá đủ.

Và như bảng cho thấy, con số 14 lần trên 16 nhân đã là gần trần lý thuyết.

Ăn trộm việc

Đây là thứ làm ForkJoinPool khác pool thường.

Pool thường có một hàng đợi chung: mọi luồng lấy việc từ đó, nên chúng tranh nhau ở đúng một chỗ.

ForkJoinPool cho mỗi luồng một hàng đợi hai đầu riêng. Luồng đẩy và lấy việc của chính mình ở đầu hàng — không tranh chấp với ai. Khi hết việc, nó đi trộm từ đuôi hàng đợi của luồng khác.

Trộm từ đuôi là chi tiết tinh tế: đuôi chứa những tác vụ cũ nhất, tức là to nhất (vì tác vụ được chia nhỏ dần theo thời gian). Trộm một tác vụ to thì đáng công đi trộm; và vì chủ hàng đang làm ở đầu kia nên hai bên ít đụng nhau.

Tôi thử một tải lệch hẳn — cây 32 lá, trong đó nhánh đầu tiên nặng gấp 30 lần các nhánh còn lại:

  32 lá, nhánh đầu nặng gấp 30 lần -> 29 ms
  số luồng thực sự tham gia: 15 / 15 luồng của pool

Cả 15 luồng đều vào việc, dù công việc chia rất không đều. Không có ăn trộm việc thì luồng nhận nhánh nặng phải gánh hết trong khi 14 luồng kia ngồi chơi.

Đây là lý do ForkJoinPool hợp với bài toán mà bạn không biết trước phần nào nặng: duyệt cây thư mục, phân tích cú pháp, tìm kiếm trên đồ thị.

Ba cách viết fork/join, chênh nhau 12 lần

Cùng bài toán, chỉ khác thứ tự gọi:

// Cách A — SAI
trai.fork();
return trai.join() + phai.compute();

// Cách B — tạm được
trai.fork(); phai.fork();
return trai.join() + phai.join();

// Cách C — khuyến nghị
trai.fork();
long r = phai.compute();
return trai.join() + r;

  fork rồi join NGAY     : 12 ms
  fork cả hai rồi join   :  1 ms
  fork một, tự làm một   :  1 ms

Cách A hỏng vì join() gọi quá sớm. Nó đẩy nhánh trái đi rồi lập tức ngồi chờ nhánh trái xong, trong khi nhánh phải còn chưa bắt đầu. Luồng hiện tại không làm gì trong lúc chờ. Mã trông rất giống cách C, chỉ đảo thứ tự hai vế của phép cộng — và chậm hơn 12 lần.

Đây là lỗi rất dễ mắc vì a.join() + b.compute() đọc lên nghe hoàn toàn tự nhiên. Nhớ nguyên tắc: fork mọi thứ trước, join sau cùng.

Cách B và C ngang nhau ở phép đo này. Cách C được khuyến nghị vì nó tiết kiệm một lần đẩy vào hàng đợi: thay vì giao cả hai nửa cho pool rồi ngồi chờ, luồng hiện tại tự làm một nửa. Với cây sâu, khoản tiết kiệm ấy cộng dồn — dù ở quy mô này tôi không đo được khác biệt.

Chặn trong ForkJoinPool

Bài hôm qua đã nói: đừng chặn trong pool chung. Nhưng đôi khi bạn buộc phải, và ForkJoinPool có lối thoát chính thức:

  8 tác vụ ngủ 500ms, chặn thẳng      : 1506 ms  (pool 3 luồng)
  cùng thế, bọc trong ManagedBlocker  :  503 ms

Ba lần nhanh hơn, và con số 503 ms cho thấy cả 8 tác vụ chạy đồng thời.

ManagedBlocker báo cho pool biết "tôi sắp chặn" để pool tạm tạo thêm luồng bù, giữ nguyên mức song song:

ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
    boolean xong = false;
    public boolean block() throws InterruptedException { Thread.sleep(500); xong = true; return true; }
    public boolean isReleasable() { return xong; }
});

Rườm rà, và tôi chỉ dùng khi không có lựa chọn khác — ví dụ khi phải chặn bên trong một parallelStream. Với mã mới trên Java 21, luồng ảo giải quyết cùng vấn đề gọn hơn nhiều.

Đáng biết: ConcurrentHashMap.computeIfAbsent và một số chỗ trong thư viện chuẩn đã dùng ManagedBlocker sẵn bên trong.

Nó nằm dưới parallelStream

IntStream.range(0, N).parallel().map(...).sum();

Dòng này chạy trên đúng pool chung vừa nói, với cùng cơ chế ăn trộm việc, và cùng cách chia đôi đệ quy. parallelStream chỉ là ForkJoinPool mặc một giao diện dễ dùng.

Nên mọi thứ trong bài đều áp dụng cho nó: chia nhỏ quá thì chậm hơn tuần tự, chặn bên trong thì giam pool chung của cả JVM, và nếu nguồn dữ liệu không chia đôi rẻ được (LinkedList, Stream.iterate) thì song song hoá gần như vô ích.

Muốn parallelStream chạy trên pool riêng, có một mẹo: gọi nó bên trong một tác vụ nộp vào pool của bạn.

new ForkJoinPool(4).submit(() -> danhSach.parallelStream().map(...).toList()).get();

Nó hoạt động vì stream song song dùng pool của luồng đang chạy nếu luồng đó thuộc một ForkJoinPool. Đây là hành vi không được ghi trong đặc tả, nên tôi coi nó là giải pháp tình thế chứ không phải nền móng.

Khi nào dùng

Hợp: bài toán chia đôi rẻ, các nửa độc lập, dữ liệu trong bộ nhớ, công việc thuần tính toán, và tổng khối lượng đủ lớn — dưới vài chục nghìn phần tử thì tuần tự thường thắng.

Không hợp: tác vụ có chặn I/O (dùng luồng ảo), tác vụ phụ thuộc nhau, dữ liệu chia không đều mà lại không chia đôi được, hoặc mỗi phần tử cần cập nhật một trạng thái chung — lúc đó tranh chấp ăn hết phần lợi.

Và câu hỏi thực tế nhất: đo trước đã. Bảng ở đầu bài cho thấy cùng một mã có thể nhanh hơn 14 lần hoặc chậm hơn 2 lần, chỉ khác nhau ở một hằng số.

Thử ba mươi giây

Lấy một parallelStream bất kỳ trong dự án của bạn, bỏ chữ .parallel() đi và đo lại.

Với bộ dữ liệu nhỏ hoặc thao tác nhẹ, bản tuần tự thường nhanh hơn — đúng như dòng "ngưỡng 100" ở đầu bài. Song song không miễn phí, và ba mươi giây đó cho bạn biết mình đang trả tiền hay đang được lợi.

Ngày mai: ConcurrentHashMap và các bộ sưu tập đồng thời — vì sao Collections.synchronizedMap không đủ, và computeIfAbsent khác get rồi put ở chỗ nào.