java.util.concurrent
Tự quản lý từng đối tượng Thread và từng khoá vừa rườm rà vừa dễ sai. Kể từ Java 5, gói java.util.concurrent đã cung cấp các trừu tượng bậc cao giúp viết mã đa luồng an toàn và gọn hơn nhiều. Bài này giới thiệu ba thành phần thực dụng nhất: bộ điều phối với bể luồng, cặp Callable/Future để lấy kết quả, và hàng đợi chặn cho mẫu sản xuất tiêu thụ.
Hãy nghĩ tới một máy chủ web nhận hàng nghìn yêu cầu mỗi giây. Nếu mỗi yêu cầu tạo một Thread mới rồi huỷ ngay sau khi xử lý xong, chi phí tạo và thu hồi luồng sẽ ngốn phần lớn tài nguyên, và số luồng có thể bùng lên tới mức máy ảo cạn bộ nhớ. Vấn đề thật không phải là chạy song song mà là chạy song song có kiểm soát: giới hạn số luồng, tái dùng chúng, phối hợp việc chuyền dữ liệu giữa các luồng mà không dính lỗi tranh chấp. Đó chính là những bài toán mà gói java.util.concurrent giải sẵn cho ta.
Cách tiếp cận của gói này là nâng mức trừu tượng. Ở bài trước, đơn vị ta thao tác là luồng: ta tạo nó, chờ nó, ngắt nó. Từ đây, đơn vị ta thao tác là tác vụ (task): ta chỉ mô tả việc cần làm rồi nộp cho hệ thống, còn chuyện luồng nào chạy nó và chạy khi nào là việc của thư viện.
ExecutorService và bể luồng
Thay vì tự tạo luồng cho mỗi việc, ta nộp công việc cho một bộ điều phối (executor) quản lý sẵn một bể luồng (thread pool). Bể luồng tái dùng một số luồng cố định để chạy lần lượt các tác vụ trong hàng đợi, nhờ vậy tránh được chi phí tạo và huỷ luồng liên tục, đồng thời khống chế số luồng chạy đồng thời.
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ViDuExecutor {
public static void main(String[] args) {
ExecutorService pool = Executors.newFixedThreadPool(3); // 3 worker threads
for (int i = 1; i <= 6; i++) {
int id = i;
pool.submit(() -> System.out.println("Tac vu " + id
+ " chay tren " + Thread.currentThread().getName()));
}
pool.shutdown(); // no new tasks; finish queued ones then stop
}
}
Lớp tiện ích Executors cung cấp vài loại bể luồng cho các nhu cầu khác nhau. Chọn đúng loại giúp cân bằng giữa thông lượng và tài nguyên:
newFixedThreadPool(n): cố địnhnluồng, các tác vụ dư xếp hàng chờ; hợp khi muốn khống chế chặt mức song song.newCachedThreadPool(): tạo luồng theo nhu cầu và tái dùng luồng rảnh; hợp với nhiều tác vụ ngắn nhưng có thể sinh rất nhiều luồng khi tải cao.newSingleThreadExecutor(): một luồng duy nhất chạy tuần tự các tác vụ theo thứ tự nộp.newScheduledThreadPool(n): chạy tác vụ theo lịch hoặc lặp lại định kỳ.
Việc tắt bộ điều phối cũng cần làm đúng. shutdown chặn nhận tác vụ mới nhưng vẫn chạy hết hàng đợi, còn shutdownNow cố dừng ngay và trả về các tác vụ chưa chạy. Muốn chờ cho mọi tác vụ hoàn tất trước khi đi tiếp, gọi awaitTermination với một hạn thời gian.
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
public class TatBoDieuPhoi {
static void tatAnToan(ExecutorService pool) throws InterruptedException {
pool.shutdown(); // stop accepting new tasks
if (!pool.awaitTermination(10, TimeUnit.SECONDS)) // wait up to 10 seconds
pool.shutdownNow(); // force stop if it overruns
}
}
Từ Java 19 trở đi, ExecutorService cài giao tiếp AutoCloseable, nên có thể đặt trong try-with-resources để tự đóng và tự chờ kết thúc khi ra khỏi khối. Đây là cách viết gọn và an toàn được khuyến nghị cho mã mới.
Chọn kích thước bể ra sao? Quy tắc ngón tay cái: với tác vụ nặng tính toán (CPU bound) thì số luồng xấp xỉ số nhân, lấy bằng Runtime.getRuntime().availableProcessors(), vì thêm luồng nữa chỉ tốn thêm chi phí chuyển ngữ cảnh chứ không tăng thông lượng. Với tác vụ chủ yếu ngồi chờ vào ra (IO bound) như gọi mạng hay đọc đĩa, có thể để số luồng lớn hơn số nhân khá nhiều, vì phần lớn thời gian chúng ngủ chứ không giành CPU.
Callable và Future
Giao tiếp Runnable không trả về kết quả. Khi tác vụ cần trả ra một giá trị, ta dùng Callable<V>: phương thức call của nó trả về kiểu V và được phép ném ngoại lệ. Khi nộp một Callable cho bộ điều phối, ta nhận lại một Future<V> làm phiếu hẹn lấy kết quả; gọi get trên Future sẽ chờ cho tới khi tác vụ xong rồi trả về giá trị.
import java.util.concurrent.*;
public class ViDuFuture {
public static void main(String[] args) throws Exception {
ExecutorService pool = Executors.newFixedThreadPool(2);
Callable<Integer> task = () -> { // returns a value
int s = 0;
for (int i = 1; i <= 100; i++) s += i;
return s; // 1 + 2 + ... + 100
};
Future<Integer> f = pool.submit(task); // a ticket for the result
System.out.println("Tong = " + f.get()); // get() waits for completion
pool.shutdown();
}
}
Điểm cần khắc sâu: get là lời gọi chặn. Luồng nào gọi get sẽ đứng yên cho tới khi tác vụ hoàn tất, nên vị trí đặt get quyết định chương trình còn song song hay không. Nộp một tác vụ rồi get ngay, sau đó mới nộp tác vụ kế tiếp, là cách chắc chắn biến bể luồng thành một hàng chạy tuần tự. Nguyên tắc đúng là nộp hết trước, thu kết quả sau.
Bản get(timeout, unit) chờ có hạn và ném TimeoutException khi quá giờ, giúp không treo vô hạn vì một tác vụ hỏng. Nếu tác vụ ném ngoại lệ, ngoại lệ đó được gói trong ExecutionException và bật ra đúng lúc ta gọi get, muốn biết nguyên nhân gốc thì gọi getCause. Với cả một danh sách tác vụ, invokeAll nộp tất cả và chỉ trả về khi mọi tác vụ đã xong.
import java.util.*;
import java.util.concurrent.*;
public class ThuKetQuaDungCach {
public static void main(String[] args) throws Exception {
ExecutorService pool = Executors.newFixedThreadPool(4);
List<Callable<Long>> viec = new ArrayList<>();
for (int i = 0; i < 4; i++) {
int tu = i * 250 + 1, den = (i + 1) * 250; // 4 disjoint ranges
viec.add(() -> {
long s = 0;
for (int k = tu; k <= den; k++) s += (long) k * k;
return s;
});
}
long tong = 0;
for (Future<Long> f : pool.invokeAll(viec)) { // submit all, then collect
try {
tong += f.get(); // already done, no blocking
} catch (ExecutionException e) {
System.out.println("Tac vu loi: " + e.getCause());
}
}
System.out.println("Tong binh phuong = " + tong); // 333833500
pool.shutdown();
}
}
CompletableFuture sơ lược
Future có một hạn chế: muốn biết kết quả thì phải chủ động gọi get và chịu chặn. CompletableFuture<T> lật ngược quan hệ đó, cho ta khai báo trước chuỗi việc cần làm khi kết quả sẵn sàng, rồi luồng gọi đi tiếp ngay. Đây là nền tảng của phong cách lập trình bất đồng bộ trong Java hiện đại.
Ba phương thức đủ để bắt đầu. supplyAsync chạy một hàm sinh giá trị trên bể luồng nền. thenApply gắn thêm một bước biến đổi kết quả. thenCombine gộp kết quả của hai chuỗi độc lập lại với nhau. Không lời gọi nào trong số đó chặn; chỉ join ở cuối mới chờ.
import java.util.concurrent.CompletableFuture;
public class ViDuCompletableFuture {
static double giaSanPham(String ma) { return 100.0; } // pretend: call service A
static double tyGia() { return 25000.0; } // pretend: call service B
public static void main(String[] args) {
CompletableFuture<Double> gia =
CompletableFuture.supplyAsync(() -> giaSanPham("SP01")); // runs async
CompletableFuture<Double> ty =
CompletableFuture.supplyAsync(() -> tyGia()); // runs in parallel
CompletableFuture<String> ketQua = gia
.thenCombine(ty, (g, t) -> g * t) // combine both results
.thenApply(vnd -> "Gia = " + vnd + " VND"); // transform, still no blocking
System.out.println("main van chay tiep...");
System.out.println(ketQua.join()); // only here we wait
}
}
Hai lời gọi dịch vụ chạy song song thay vì nối đuôi nhau, và mã đọc như một dây chuyền xử lý chứ không phải một mớ get rải rác. Khi cần bắt lỗi, dùng exceptionally để cung cấp giá trị thay thế, hoặc handle để xử lý cả hai nhánh thành công và thất bại.
Hàng đợi chặn
Hàng đợi chặn (blocking queue) là cấu trúc dữ liệu an toàn cho đa luồng, dùng để chuyền dữ liệu giữa các luồng. Điểm đặc biệt: khi hàng đợi rỗng, luồng lấy ra tự động chờ cho tới khi có phần tử; khi hàng đợi đầy, luồng bỏ vào tự động chờ cho tới khi có chỗ trống. Nhờ đó nó là nền tảng tự nhiên cho mẫu sản xuất tiêu thụ (producer consumer), nơi một nhóm luồng sinh dữ liệu và một nhóm khác xử lý dữ liệu mà không cần ta tự viết đồng bộ.
import java.util.concurrent.*;
public class SanXuatTieuThu {
public static void main(String[] args) {
BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5); // capacity 5
Runnable producer = () -> {
try { for (int i = 1; i <= 5; i++) queue.put(i); } // blocks if full
catch (InterruptedException e) { Thread.currentThread().interrupt(); }
};
Runnable consumer = () -> {
try { for (int i = 1; i <= 5; i++)
System.out.println("Tieu thu " + queue.take()); } // blocks if empty
catch (InterruptedException e) { Thread.currentThread().interrupt(); }
};
new Thread(producer).start();
new Thread(consumer).start();
}
}
Chọn cài đặt nào? ArrayBlockingQueue có sức chứa cố định đặt sẵn lúc tạo, nên nó đóng vai một van điều tiết: khi luồng tiêu thụ chậm, hàng đợi đầy và luồng sản xuất bị chặn lại, không thể sinh dữ liệu tới mức tràn bộ nhớ. LinkedBlockingQueue mặc định không giới hạn sức chứa, gọn hơn nhưng mất đúng cơ chế điều tiết đó.
Làm sao báo cho luồng tiêu thụ biết là hết hàng? Vì take chặn vô hạn trên hàng đợi rỗng, luồng tiêu thụ sẽ treo mãi nếu ta không nói gì. Kỹ thuật quy ước là viên thuốc độc (poison pill): sản xuất xong thì bỏ vào hàng đợi một giá trị đặc biệt mà mọi luồng tiêu thụ đều hiểu là tín hiệu dừng, và bỏ đúng số viên bằng số luồng tiêu thụ.
Biến nguyên tử
Khi nhiều luồng cùng tăng một biến đếm, phép count++ không an toàn: nó gồm ba bước đọc, cộng, ghi, và hai luồng có thể xen kẽ khiến một lần tăng bị mất. Cách cũ là bọc trong khối synchronized, nhưng gói java.util.concurrent.atomic cho một lựa chọn nhẹ hơn: các lớp như AtomicInteger, AtomicLong, AtomicReference bảo đảm mỗi thao tác diễn ra trọn vẹn như một khối không thể chia cắt, thường dựa trên chỉ thị phần cứng chứ không cần khoá.
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.*;
public class ViDuAtomic {
public static void main(String[] args) throws InterruptedException {
AtomicInteger dem = new AtomicInteger(0); // safe counter
ExecutorService pool = Executors.newFixedThreadPool(8);
for (int i = 0; i < 1000; i++) {
pool.submit(dem::incrementAndGet); // atomic ++, no lost update
}
pool.shutdown();
pool.awaitTermination(5, TimeUnit.SECONDS);
System.out.println("Tong = " + dem.get()); // always 1000
}
}
Phương thức incrementAndGet, getAndAdd, compareAndSet đều nguyên tử, nên đây là lựa chọn gọn cho các bộ đếm và cờ trạng thái chia sẻ mà không phải tự viết đồng bộ. Với trường hợp phức tạp hơn một biến đơn, hãy quay lại với khoá hoặc các cấu trúc dữ liệu đồng thời như ConcurrentHashMap.
ConcurrentHashMap
HashMap thường không an toàn cho đa luồng: hai luồng cùng ghi có thể làm hỏng cấu trúc bên trong. Người ta từng bọc nó bằng Collections.synchronizedMap, nhưng cách này khoá toàn bộ bản đồ cho mỗi thao tác, biến mọi truy cập thành xếp hàng một. ConcurrentHashMap cho nhiều luồng đọc và ghi ở các vùng khác nhau cùng lúc, nên thông lượng cao hơn hẳn dưới tải nặng.
Điều dễ sai là tưởng cứ dùng ConcurrentHashMap thì mọi thứ tự khắc an toàn. Từng thao tác đơn lẻ như put hay get là nguyên tử, nhưng một chuỗi thao tác thì không: đoạn if (!map.containsKey(k)) map.put(k, v); gồm hai bước và hai luồng có thể cùng lọt qua bước kiểm tra. Vì vậy hãy dùng các thao tác gộp nguyên tử mà lớp này cung cấp: putIfAbsent, computeIfAbsent, merge.
import java.util.concurrent.*;
public class DemTuSongSong {
public static void main(String[] args) throws Exception {
String[] vanBan = {"java", "thread", "java", "pool", "java", "thread"};
ConcurrentHashMap<String, Integer> dem = new ConcurrentHashMap<>();
ExecutorService pool = Executors.newFixedThreadPool(4);
for (String tu : vanBan) {
pool.submit(() -> dem.merge(tu, 1, Integer::sum)); // atomic read-modify-write
}
pool.shutdown();
pool.awaitTermination(5, TimeUnit.SECONDS);
System.out.println(dem); // java=3, thread=2, pool=1 (print order not guaranteed)
}
}
Lời gọi merge(tu, 1, Integer::sum) nghĩa là: nếu khoá chưa có thì đặt giá trị 1, nếu đã có thì lấy giá trị cũ cộng 1. Toàn bộ chuỗi đọc, cộng, ghi này diễn ra nguyên tử, nên dù bốn luồng cùng đụng vào một khoá thì không lần tăng nào bị mất.
CountDownLatch
Có những lúc ta cần một luồng đứng chờ tới khi một nhóm việc khác hoàn tất, chẳng hạn chờ mọi mô đun nạp xong rồi mới mở giao diện, hoặc chờ tất cả luồng thợ tính xong rồi mới tổng hợp. CountDownLatch là cái chốt cửa cho tình huống đó: khởi tạo với một con số, mỗi việc xong thì gọi countDown để giảm một, còn luồng chờ gọi await và chỉ được đi tiếp khi số đếm về 0.
import java.util.concurrent.*;
public class ChoKhoiDong {
public static void main(String[] args) throws InterruptedException {
int soModule = 3;
CountDownLatch chot = new CountDownLatch(soModule); // count starts at 3
ExecutorService pool = Executors.newFixedThreadPool(soModule);
for (int i = 1; i <= soModule; i++) {
int id = i;
pool.submit(() -> {
try {
Thread.sleep(200L * id); // pretend: loading work
System.out.println("Module " + id + " san sang");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
chot.countDown(); // always count down
}
});
}
chot.await(); // blocks until count is 0
System.out.println("Tat ca module da san sang, mo giao dien");
pool.shutdown();
}
}
Hai chi tiết đáng nhớ. Thứ nhất, luôn gọi countDown trong khối finally: nếu một tác vụ ném ngoại lệ mà quên giảm chốt, số đếm không bao giờ về 0 và luồng chờ treo vĩnh viễn. Thứ hai, chốt là thiết bị dùng một lần, đếm về 0 rồi thì không đặt lại được; muốn một rào chắn tái sử dụng qua nhiều vòng, hãy dùng CyclicBarrier.
- Quên gọi
shutdown: bể luồng không tự tắt, chương trình treo không thoát vì luồng của bể vẫn sống. - Dùng
count++chia sẻ giữa nhiều luồng mà không đồng bộ: mất cập nhật do tranh chấp; hãy dùngAtomicIntegerhoặc khoá. - Nuốt
InterruptedExceptionbằng khốicatchrỗng: hãy khôi phục cờ ngắt bằngThread.currentThread().interrupt(). - Gọi
future.get()ngay sausubmittrong vòng lặp: vô tình biến song song thành tuần tự vì mỗi lần chờ một tác vụ xong mới nộp tiếp; hãy nộp hết rồi mới thu kết quả.
Ưu tiên dùng java.util.concurrent. Tự dựng luồng và khoá thủ công chỉ nên dành cho lúc học cơ chế hoặc khi thư viện không đáp ứng nhu cầu đặc thù. Mã dùng bộ điều phối và hàng đợi chặn vừa ngắn hơn vừa ít lỗi đồng bộ hơn.
Bài tập thực hành
Bài 1. Dùng ExecutorService với bể bốn luồng để tính song song tổng bình phương các số từ 1 đến 1000, chia thành bốn khoảng. Mỗi khoảng là một Callable<Long>; thu kết quả qua các Future rồi cộng lại và in ra.
Gợi ý
Tạo Executors.newFixedThreadPool(4). Mỗi Callable<Long> nhận một cận đầu và một cận cuối, cộng i * i trong khoảng rồi trả về tổng. Nộp cả bốn tác vụ, giữ lại bốn Future<Long> trong một danh sách, sau đó cộng các giá trị get lại. Nhớ gọi shutdown khi xong.
Bài 2. Cài đặt mẫu sản xuất tiêu thụ bằng BlockingQueue: hai luồng sản xuất bỏ vào hàng đợi các số nguyên, ba luồng tiêu thụ lấy ra và in kèm tên luồng. Dùng một giá trị quy ước báo hết để các luồng tiêu thụ biết khi nào dừng.
Gợi ý
Dùng một ArrayBlockingQueue<Integer>. Sau khi cả hai luồng sản xuất xong, bỏ vào ba phần tử báo hết (mỗi luồng tiêu thụ một cái), ví dụ giá trị quy ước -1. Mỗi luồng tiêu thụ lặp take; khi gặp -1 thì dừng. Cách này tránh việc luồng tiêu thụ chờ mãi trên hàng đợi rỗng.
Bài 3. So sánh Runnable và Callable<V>. Khi nào bắt buộc phải dùng Callable?
Gợi ý
Runnable có phương thức run không trả về giá trị và không được khai báo ném ngoại lệ kiểm tra. Callable<V> có phương thức call trả về kiểu V và được phép ném ngoại lệ. Bắt buộc dùng Callable khi tác vụ cần trả ra một kết quả để gọi get qua Future, hoặc khi thân tác vụ có thể ném ngoại lệ kiểm tra cần truyền ngược về nơi nộp tác vụ.
Bài 4. Cho một mảng chuỗi, hãy đếm số lần xuất hiện của mỗi chuỗi bằng bốn luồng, dùng ConcurrentHashMap và CountDownLatch để luồng main chỉ in kết quả sau khi mọi luồng đã đếm xong.
Lời giải
Dùng merge để cộng dồn nguyên tử, countDown trong finally để chốt luôn về 0 dù tác vụ có lỗi, và await để main chờ.
import java.util.concurrent.*;
public class DemTanSuat {
public static void main(String[] args) throws InterruptedException {
String[] du = {"java", "thread", "java", "pool", "java", "thread", "future"};
ConcurrentHashMap<String, Integer> dem = new ConcurrentHashMap<>();
CountDownLatch chot = new CountDownLatch(du.length);
ExecutorService pool = Executors.newFixedThreadPool(4);
for (String tu : du) {
pool.submit(() -> {
try {
dem.merge(tu, 1, Integer::sum); // atomic increment per key
} finally {
chot.countDown(); // always release the latch
}
});
}
chot.await(); // wait for all 7 tasks
// java=3, thread=2, pool=1, future=1 (print order not guaranteed)
System.out.println(dem);
pool.shutdown();
}
}
Nếu thay merge bằng dem.put(tu, dem.getOrDefault(tu, 0) + 1) thì chương trình lại sai, vì đọc rồi ghi là hai bước tách rời và hai luồng có thể xen kẽ làm mất một lần tăng.
Tự kiểm tra
Khi nào nên ưu tiên ExecutorService thay vì tự tạo Thread?
Khi ứng dụng có nhiều tác vụ và muốn khống chế số luồng chạy đồng thời. ExecutorService quản lý một bể luồng tái dùng, tránh chi phí tạo và huỷ luồng liên tục, cho phép nộp tác vụ vào hàng đợi và lấy kết quả qua Future. Tự tạo Thread chỉ phù hợp cho ví dụ nhỏ hoặc khi học cơ chế bên dưới.
Câu hỏi tự kiểm
- 1Vì sao gọi future.get() ngay sau mỗi submit trong một vòng lặp lại là sai?
- 2Khác biệt giữa Runnable và Callable<V> là gì?
- 3Vì sao nên dùng dem.merge(tu, 1, Integer::sum) thay cho dem.put(tu, dem.get(tu) + 1) trên một ConcurrentHashMap?
- 4Kỹ thuật viên thuốc độc (poison pill) dùng để làm gì?
Tóm tắt
Đơn vị làm việc trong java.util.concurrent là tác vụ chứ không phải luồng. Nộp tác vụ cho ExecutorService, chọn loại bể theo nhu cầu, và luôn shutdown khi xong. Dùng Runnable khi tác vụ không trả gì, Callable<V> khi cần một giá trị hoặc cần ném ngoại lệ kiểm tra; Future.get là lời gọi chặn nên phải nộp hết trước rồi mới thu kết quả. CompletableFuture cho phép nối chuỗi xử lý bất đồng bộ mà không chặn, với supplyAsync, thenApply, thenCombine.
Để chuyền dữ liệu giữa luồng, dùng BlockingQueue với mẫu sản xuất tiêu thụ và một viên thuốc độc để báo hết. Để chia sẻ trạng thái, dùng các lớp nguyên tử cho bộ đếm và cờ, ConcurrentHashMap với merge hay computeIfAbsent cho bản đồ, CountDownLatch khi cần chờ một nhóm việc hoàn tất.
Trong ứng dụng thật, hãy ưu tiên java.util.concurrent với ExecutorService, Callable/Future và hàng đợi chặn để mã vừa gọn vừa an toàn. Bộ điều phối quản lý bể luồng giúp tái dùng luồng và khống chế mức song song; Callable/Future cho phép lấy kết quả tính toán; hàng đợi chặn lo phần đồng bộ giữa luồng sản xuất và luồng tiêu thụ.