ExecutorService, Callable, Future: chạy tác vụ

1. ExecutorService: quản lý luồng một cách chuyên nghiệp

Vì sao không nên chỉ đơn giản tạo luồng bằng new Thread

Ban đầu, đa luồng trông rất đơn giản:

Thread t = new Thread(() -> {
    // làm gì đó
});
t.start();

Cách này hoạt động, nhưng nhanh chóng trở thành gánh nặng khi số lượng tác vụ tăng lên. Mỗi lần gọi new Thread() tạo ra một luồng mới, và hàng chục hay hàng trăm luồng sẽ bắt đầu quá tải hệ thống. Hơn nữa việc quản lý chúng không tiện: phải theo dõi khi nào chúng kết thúc, làm gì khi có lỗi, dừng và tái sử dụng ra sao.

Đây là lúc ExecutorService xuất hiện — bộ điều phối luồng thông minh. Bạn chỉ việc giao tác vụ cho nó, còn nó sẽ tự quyết định dùng luồng nào và khi nào thực thi. Kết quả là mọi thứ chạy nhanh hơn, ổn định hơn và đỡ nhức đầu.

ExecutorService hoạt động như thế nào

ExecutorService hoạt động theo nguyên tắc đơn giản nhưng hiệu quả.

• Bên trong có pool luồng — tập hợp các luồng làm việc được tạo sẵn (cố định hoặc động).
• Tác vụ được đưa vào hàng đợi và được các luồng rảnh xử lý.
• Dịch vụ quản lý vòng đời: bạn có thể chờ hoàn thành, dừng pool một cách an toàn và giải phóng tài nguyên.

Tạo ExecutorService

Cách phổ biến nhất — dùng các phương thức factory của lớp Executors:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

ExecutorService executor = Executors.newFixedThreadPool(4); // 4 luồng

• newFixedThreadPool(N) — pool gồm N luồng (phù hợp với đa số tác vụ).
• newCachedThreadPool() — pool động, tạo luồng khi cần (cẩn thận: có thể cạn bộ nhớ khi tác vụ đổ dồn).
• newSingleThreadExecutor() — một luồng (thực thi tuần tự).

Ví dụ: chạy Runnable qua ExecutorService

executor.submit(() -> {
    System.out.println("Xin chào từ pool luồng!");
});

Sau khi bạn làm xong với ExecutorService, cần kết thúc nó đúng cách:

executor.shutdown(); // Cấm thêm tác vụ mới, chờ các tác vụ hiện tại kết thúc

Quan trọng: Nếu không gọi shutdown(), chương trình có thể không thoát — các luồng trong pool sẽ chờ tác vụ mới.

2. Runnable vs Callable: tác vụ có nhiều loại

Trước Java 5, nếu bạn muốn chạy gì đó trong một luồng, bạn viết một triển khai của giao diện Runnable. Đây là tác vụ không trả về giá trị và không ném checked exception.

Runnable task = () -> {
    System.out.println("Chỉ làm việc, không trả về gì cả!");
};
executor.submit(task);

Callable: tác vụ có kết quả (và có thể ném ngoại lệ)

Đôi khi bạn muốn tác vụ không chỉ “làm gì đó” mà còn trả về kết quả — ví dụ tổng các số, kết quả tính toán, dữ liệu từ server. Đó là lý do có giao diện Callable<T>.

import java.util.concurrent.Callable;

Callable<Integer> sumTask = () -> {
    int sum = 0;
    for (int i = 1; i <= 100; i++) sum += i;
    return sum;
};

• Phương thức call() trả về kết quả kiểu T.
• Phương thức call() có thể ném checked exception.

Ví von: Runnable — “đi rửa bát” (kết quả không quan trọng), Callable — “đi lấy trà và cho biết trà nóng bao nhiêu” (kết quả quan trọng).

Chạy Callable: để lấy kết quả, hãy dùng executor.submit(...). Nó sẽ trả về đối tượng Future<T>.

3. Future: lời hứa về kết quả

Future là “lời hứa” sẽ trả kết quả trong tương lai. Khi bạn gửi tác vụ vào ExecutorService, bạn nhận được Future, từ đó sau này có thể lấy kết quả, biết tác vụ đã xong chưa, hoặc hủy nó.

Các phương thức chính của Future

• T get() — lấy kết quả (chờ đến khi tác vụ hoàn tất).
• boolean isDone() — tác vụ đã hoàn tất chưa.
• boolean cancel(boolean mayInterruptIfRunning) — cố gắng hủy tác vụ.
• boolean isCancelled() — tác vụ đã bị hủy chưa.

Ví dụ: chạy Callable và lấy kết quả

import java.util.concurrent.*;

public class ParallelSumApp {
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        Callable<Integer> sumTask = () -> {
            int sum = 0;
            for (int i = 1; i <= 100; i++) sum += i;
            return sum;
        };

        Future<Integer> future = executor.submit(sumTask);

        System.out.println("Tác vụ đã khởi chạy, có thể làm việc khác...");

        // Lấy kết quả (phương thức sẽ chặn luồng nếu tác vụ chưa hoàn tất)
        Integer result = future.get();
        System.out.println("Kết quả tính toán: " + result);

        executor.shutdown();
    }
}

• Tác vụ được gửi vào pool luồng.
• Trong khi tác vụ chạy, luồng chính có thể làm việc khác.
• Khi cần kết quả, gọi future.get() — luồng sẽ chờ nếu tác vụ còn đang chạy.
• Ngay khi tác vụ kết thúc, kết quả sẽ được trả về.

4. Thực hành: nhiều tác vụ, chờ hoàn tất

Thường cần chạy ngay nhiều tác vụ và đợi cho đến khi tất cả hoàn tất. Ví dụ, bạn xử lý một mảng dữ liệu, chia thành các phần và tính tổng từng phần trong các tác vụ riêng.

Ví dụ: tính tổng phần tử mảng theo từng phần

import java.util.*;
import java.util.concurrent.*;

public class ParallelArraySum {
    public static void main(String[] args) throws Exception {
        int[] array = new int[1000];
        Arrays.setAll(array, i -> i + 1); // Điền các số từ 1 đến 1000

        ExecutorService executor = Executors.newFixedThreadPool(4);

        int chunkSize = array.length / 4;
        List<Future<Integer>> futures = new ArrayList<>();

        for (int i = 0; i < 4; i++) {
            int from = i * chunkSize;
            int to = (i == 3) ? array.length : (i + 1) * chunkSize;

            Callable<Integer> sumTask = () -> {
                int sum = 0;
                for (int j = from; j < to; j++) sum += array[j];
                System.out.println("Tổng từ " + from + " đến " + (to - 1) + " = " + sum);
                return sum;
            };

            futures.add(executor.submit(sumTask));
        }

        int totalSum = 0;
        for (Future<Integer> f : futures) {
            totalSum += f.get(); // Chờ từng tác vụ lần lượt
        }

        System.out.println("Tổng chung: " + totalSum);

        executor.shutdown();
    }
}

Ở đây mảng được chia thành 4 phần. Với mỗi phần, tạo một tác vụ (Callable) để tính tổng. Tất cả tác vụ được gửi vào ExecutorService, trả về các Future. Cuối cùng gom kết quả của tất cả tác vụ và cộng lại.

Trong bài toán thực tế tiện dùng invokeAll để chờ tất cả tác vụ hoàn tất cùng lúc.

5. Xử lý lỗi khi làm việc với Future

Khi bạn gọi future.get(), nếu tác vụ kết thúc với ngoại lệ, nó sẽ được ném dưới dạng ExecutionException. Điều này quan trọng: nếu trong tác vụ có vấn đề, bạn sẽ chỉ biết khi gọi get().

Ví dụ: xử lý ngoại lệ

Callable<Integer> errorTask = () -> {
    throw new IllegalArgumentException("Có gì đó không ổn!");
};

Future<Integer> badFuture = executor.submit(errorTask);

try {
    badFuture.get();
} catch (ExecutionException e) {
    System.out.println("Tác vụ kết thúc với lỗi: " + e.getCause());
}

• Ngoại lệ được ném bên trong tác vụ.
• Khi gọi get() nó được “bọc” trong ExecutionException.
• Có thể lấy nguyên nhân thực sự qua getCause().

6. Những điểm hữu ích

Cách hủy một tác vụ

Future<?> f = executor.submit(() -> {
    while (true) {
        // Chạy vô hạn
        if (Thread.currentThread().isInterrupted()) {
            System.out.println("Tôi được yêu cầu kết thúc!");
            break;
        }
    }
});

Thread.sleep(100); // Chờ một chút
f.cancel(true); // Thử hủy tác vụ

• cancel(true) cố gắng ngắt tác vụ nếu nó chưa hoàn tất.
• Bên trong tác vụ nên kiểm tra Thread.currentThread().isInterrupted() và kết thúc đúng cách.

shutdown vs shutdownNow

shutdown() — dừng mềm: cấm thêm tác vụ mới và cho phép các tác vụ hiện tại kết thúc bình thường. Dùng thường xuyên nhất.

shutdownNow() — dừng cứng: cố gắng ngắt các luồng đang hoạt động và trả về danh sách các tác vụ chưa kịp khởi chạy. Hãy dùng cẩn trọng.

invokeAll và invokeAny

invokeAll(Collection<Callable<T>> tasks) chạy tất cả các tác vụ được truyền vào và chờ cho đến khi tất cả hoàn tất. Trả về danh sách Future.

invokeAny(Collection<Callable<T>> tasks) chỉ chờ tác vụ đầu tiên hoàn tất thành công, trả về kết quả của nó và hủy các tác vụ còn lại. Hữu ích khi quan trọng là câu trả lời thành công đầu tiên.

7. Các lỗi thường gặp khi làm việc với ExecutorService, Callable và Future

Lỗi #1: Không đóng ExecutorService. Nếu quên gọi shutdown(), chương trình có thể “treo” sau khi main kết thúc vì các luồng trong pool chờ tác vụ mới.

Lỗi #2: Chờ kết quả ngay sau khi gửi tác vụ. Nếu ngay sau submit() gọi get(), bạn sẽ không có lợi ích của bất đồng bộ — luồng vẫn phải chờ. Hãy làm việc hữu ích song song và chỉ yêu cầu kết quả khi thật sự cần.

Lỗi #3: Bỏ qua ngoại lệ trong tác vụ. Nếu không xử lý ExecutionException khi gọi get(), bạn có thể bỏ lỡ các lỗi quan trọng xảy ra trong tác vụ.

Lỗi #4: Dùng chung biến có thể thay đổi mà không đồng bộ. Nếu nhiều tác vụ cùng làm việc với cùng một dữ liệu — cần đồng bộ hóa hoặc dùng các collection an toàn luồng.

Lỗi #5: Tạo quá nhiều luồng. Không nên tạo pool với số luồng vượt quá nhiều so với số nhân CPU — điều này thậm chí có thể làm chậm việc thực thi.

Lỗi #6: Quên hủy tác vụ. Nếu tác vụ không còn cần nữa, hãy hủy bằng cancel() để tránh lãng phí tài nguyên.