Như vậy là sau tập đầu tiên của Thread Pool, bạn đã biết được rằng mình đang cố gắng giúp bạn ghi nhớ được ý nghĩa của kỹ thuật này như là một cái Hồ điều tiết Thread. Từ việc hiểu ý nghĩa đó, bạn cũng đã nắm được công năng và nắm được có bao nhiêu cách để có thể tổ chức nên cái hồ này.

Tuy nhiên tất cả cũng vẫn dừng lại ở lý thuyết. Việc hiểu tường tận hơn từng cách sử dụng Hồ, và làm quen với các ví dụ, chính là mục đích chính bắt đầu từ bài học hôm nay.

Mặc dù mình đã cố gắng trình bày hết mức có thể cách sử dụng cũng như ý nghĩa của các phương thức mà bộ ba các lớp của bài hôm nay mang lại, nhưng mình biết sẽ còn thiếu sót nhiều lắm. Một phần vì giới hạn của bài viết, tránh viết quá dài (thực ra nó cũng dài lắm rồi, và tốn rất nhiều thời gian để viết bài này), phần nữa vì mình cũng chưa có cơ hội sử dụng hết tất cả các kiến thức mà Thread Pool mang lại, vì chúng khá rộng lớn. Nên qua bài học nếu bạn nào có những đóng góp, xây dựng thì hãy liên lạc với mình qua các kênh mình liệt kê ở cuối bài viết hôm nay nhé.

Nào chúng ta cùng bắt đầu làm quen với cách sử dụng Thread Pool thông qua bộ ba tiện ích: Executors, Executor và ExecutorService.

Giới Thiệu Executors, Executor Và ExecutorService

Như đã giới thiệu từ bài học trước, bộ ba Executors, Executor và ExecutorService giúp bạn sử dụng ThreadPoolExecutor được dễ dàng hơn. Cụ thể chúng là gì thì mình có thể diễn đạt ra như sau.

• Executors: Đây được xem như một Helper Class. Nó cũng chỉ là một class thông thường thôi, nhưng lại cung cấp các phương thức hữu dụng, và dễ dàng nhất, để khởi tạo ra các ThreadPoolExecutor. Chúng ta gọi các lớp như vậy là Helper Class. Một lát bạn để ý xem Executors rất dễ sử dụng như thế nào nhé.
• Executor: Là một interface. Nó chỉ chứa mỗi phương thức execute(Runnable). Chính ThreadPoolExecutor phải implement interface này và hiện thực phương thức này, giúp bạn đưa một Runnable vào Thread Pool một cách dễ dàng.
• ExecutorService: Là lớp triển khai của của Executor, và vì vậy nó cũng là một interface. Nó cung cấp một số phương thức ràng buộc mở rộng hơn Executor mà lát nữa bạn sẽ được làm quen ở mục tiếp theo.
• Thực ra nếu nói sâu hơn nữa còn có ScheduledExecutorService, thằng này lại là lớp triển khai của ExecutorService trên đây. Lớp này cho phép bạn lên lịch (schedule) cho việc thực thi các tác vụ, tuy nhiên do bài viết hôm nay vốn đã rất dài rồi nên mình sẽ không dành thêm giấy mực để trình bày đến lớp này nhé.

Ví dụ sau cho thấy code của việc khai báo một Thread Pool có thể thực thi đồng thời 3 Thread một lúc.

Trước khi bắt đầu làm quen với các bài thực hành để biết rõ hơn về việc sử dụng Thread Pool. Thì mình mời các bạn cùng làm quen đến một số phương thức mà Helper Class Executors mang đến, để xem lớp này giúp bạn “triệu hồi” các thể loại Thread Pool nào nhé.

Làm Quen Với Executors

Như bạn biết, Executors là một Helper Class, nó chính là một lớp đắc lực để bạn có thể thi triển các ThreadPoolExecutor mà không cần biết tí gì về ThreadPoolExecutor cả. Chung quy lại thì Executors có các phương thức xây dựng sẵn hữu ích mà mình biết như sau (ngoài các phương thức này ra còn khá là nhiều các phương thức khác mà thú thiệt mình vẫn chưa có cơ hội tìm hiểu đến, nếu được bạn hãy tự mày mò tìm hiểu thêm nhé).

• newSingleThreadExecutor(): Giúp tạo ra một Thread Pool có khả năng thực thi 1 Thread trong đó. Như vậy nếu bạn dùng đến Thread Pool dạng này. Hồ điều tiết của bạn sẽ khá nhỏ, vì các Thread khi này được xem như được thực hiện tuần tự từng em một.
• newCachedThreadPool(): Thread Pool này mình chưa dùng bao giờ. Nghe nói là hệ thống sẽ tự quyết định số lượng Thread được thực thi trong Hồ. Có vài thông tin hay ho đối với Pool này, đó là Pool sẽ cache và sử dụng lại cấu trúc của Thread cũ đã xử lý xong để thực thi cho Thread mới. Ngoài ra nếu một Thread trong Pool này không được sử dụng trong vòng 60 giây sẽ bị gỡ ra khỏi cache. Những tính năng này giúp cho Pool được khởi tạo theo kiểu này tận dụng được hiệu năng của hệ thống, đồng thời cũng giúp tránh bị tình trạng nắm giữ resource của hệ thống quá lâu.
• newFixedThreadPool(int nThreads): Đây là Thread Pool thông dụng mà mình thấy. Phương thức này giúp tạo ra một Pool có thể chứa tối đa nThreads. Khi Pool đạt đến giá trị tối đa nThreads, các Thread còn lại sẽ được đưa vào hàng đợi và chờ đến khi có Thread trong Pool được xử lý xong mới được thực thi tiếp.

Chúng ta sẽ cùng làm quen với các thể loại Thread Pool được liệt kê trên đây mà lớp Executors cung cấp thông qua các bài thực hành sau. Các bạn cùng code với mình nhé.

Bài Thực Hành Số 1 – newSingleThreadExecutors

Như đã nói trên kia, phương thức newSingleThreadExecutors của Executors giúp tạo ra một Thread Pool mà chỉ có duy nhất 1 Thread được thực thi một lần.

Để kiểm chứng, chúng ta cùng xây dựng Thread Pool này. Nhưng trước hết, mình muốn các bạn xây dựng một Runnable để chúng ta có thể truyền chúng vào trong Thread Pool mà chúng ta sẽ xây dựng ở bước kế tiếp sau. Runnable này mình đặt tên là MyRunnable nhé.

public class MyRunnable implements Runnable {

 

// Tên của Runnable, giúp chúng ta phân biệt Runnable nào đang thực thi trong Thread Pool

private String name;

 

public MyRunnable(String name) {

// Khởi tạo Runnable với biến name truyền vào

this.name = name;

}

 

@Override

public void run() {

System.out.println(name + " đang thực thi...");

 

// Giả lập thời gian chạy của Runnable mất 2 giây

try {

Thread.sleep(2000);

} catch (InterruptedException e) {

e.printStackTrace();

}

 

System.out.println(name + " kết thúc.");

}

 

}

Trên đây là một Runnable bình thường thôi. Tại sao chúng ta lại phải sử dụng Runnable? Vì Thread Pool chỉ nhận các Runnable truyền vào (và Callable nữa mà bạn sẽ làm quen sau, nhưng Callable cũng giống Runnable mà thôi). Bạn nên nhớ là các Runnable khi chưa gọi phương thức start() thì vẫn chưa được hệ thống thực thi thành các Thread. Bạn có thể xem lại kiến thức này ở bài học về Thread này. Do đó việc chúng ta truyền các Runnable vào Thread Pool là để cho Thread Pool sắp xếp chúng vào hàng đợi. Đến khi quyết định thực thi Runnable nào, các Thread Pool sẽ tiến hành start() Runnable đó.

Tiếp theo sau là code ở phương thức main.

public static void main(String[] args) {

// Khai báo một Thread Pool thông qua newSingleThreadExecutor() của Executors

ExecutorService executorService = Executors.newSingleThreadExecutor();

 

// Khai báo 10 Runnable, và "quăng" chúng vào Thread Pool vừa khai báo

for (int i = 1; i <= 10; i++) {

MyRunnable myRunnable = new MyRunnable("Runnable " + i);

executorService.execute(myRunnable);

}

 

// Phương thức này sẽ được nói sau ở ExecutorService

executorService.shutdown();

}

Bạn có thể thấy, dòng đầu tiên của code trên khai báo một Thread Pool thông qua phương thức Executors.newSingleThreadExecutor(). Phương thức này của Executors giúp tạo một Thread Pool với chỉ duy nhất một Thread được thực thi. Thread Pool này được quản lý thông qua biến executorService, đây chính là lớp ExecutorService mà chúng ta sẽ tìm hiểu sau. Bạn chỉ cần biết thêm là phương thức executorService.execute(myRunnable) giúp lần lượt đưa các Runnable được khởi tạo vào trong Thread Pool và lần lượt thực thi sau đó.

Đến đây bạn có thể thực thi chương trình để xem, bạn có thể thấy cứ mỗi 2 giây, chỉ có 1 Thread được chạy và in ra console mà thôi. Hình dưới là kết quả ở console của máy mình.

Bài Thực Hành Số 2 – newFixedThreadPool

Thông qua bài thực hành trên, bạn đã hiểu cách làm việc với Thread Pool rồi. Bài thực hành số 2 này mình không nói nhiều. Bạn tự code lại theo khai báo mới của Thread Pool (như tô sáng ở code dưới) rồi trải nghiệm nhé. Lưu ý là chúng ta dùng lại MyRunnable và không thay đổi gì cả trong code của lớp này để cùng so sánh xem Thread Pool mới này sẽ cho ra kết quả khác như thế nào ở console.

public static void main(String[] args) {

// Khai báo một Thread Pool thông qua newFixedThreadPool(5) của Executors.

// Thread Pool này cho phép thực thi cùng một lúc 5 Thread

ExecutorService executorService = Executors.newFixedThreadPool(5);

 

// Khai báo 10 Runnable, và "quăng" chúng vào Thread Pool vừa khai báo

for (int i = 1; i <= 10; i++) {

MyRunnable myRunnable = new MyRunnable("Runnable " + i);

executorService.execute(myRunnable);

}

 

// Phương thức này sẽ được nói sau ở ExecutorService

executorService.shutdown();

}

Bài Thực Hành Số 3 – newCachedThreadPool

Cũng tương tự, bạn hãy thử dùng newCachedThreadPool để tạo một Thread Pool xem. Với cách này hệ thống sẽ thực thi hết các Thread trong phương thức main của chúng ta.

public static void main(String[] args) {

// Khai báo một Thread Pool thông qua newCachedThreadPool của Executors.

ExecutorService executorService = Executors.newCachedThreadPool();

 

// Khai báo 10 Runnable, và "quăng" chúng vào Thread Pool vừa khai báo

for (int i = 1; i <= 10; i++) {

MyRunnable myRunnable = new MyRunnable("Runnable " + i);

executorService.execute(myRunnable);

}

 

// Phương thức này sẽ được nói sau ở ExecutorService

executorService.shutdown();

}

Làm Quen Với Executor Và ExecutorService

Executor và ExecutorService là một. Tại sao thì mình đã có giải thích ở mục giới thiệu về chúng trên kia rồi. Do đặc tính này mà mình sẽ tập trung nói về ExecutorService, vì nó vừa bao gồm phương thức execute(Runnable) của Executor, vừa xây dựng thêm nhiều phương thức hữu ích khác nữa.

Bạn cũng có thể thấy rằng, thông qua các ví dụ trên đây, chúng ta đã cùng làm quen, và cùng hiểu sơ qua công dụng của ExecutorService rồi. Đó là nó giúp chúng ta đưa các Runnable vào bên trong Thread Pool thông qua phương thức execute(Runnable). Chính các Thread Pool sẽ quyết định thực thi các Runnable này theo kịch bản mà chúng được khai báo.

Sau đây là tất cả các phương thức mà ExecutorService cung cấp.

• execute(Runnable): Phương thức này bạn đã được làm quen thông qua các bài thực hành trên đây rồi nên mình không nói lại nữa. Nhưng bạn cũng nên biết, sau khi đã thực hành các bài trên. Đó là phương thức này được xem như việc đưa các Runnable vào Thread Pool và khởi chạy chúng theo kiểu bất đồng bộ. Đó là bạn sẽ không biết được khi nào các Runnable kết thúc, và các kết quả mà chúng trả về là gì.
• submit(Runnable) và submit(Callable): Phương thức submit() cho phép truyền vào hoặc là Runnable như cách bạn thực hành với execute() trên kia, hoặc là Callable. Về cơ bản thì Callable cũng như Runnable, chúng cũng có khả năng tạo ra một Thread. Nhưng Callable thì lại cho phép Thread này trả kết quả về một cách đồng bộ, khi mà Runnable lại không làm được điều đó. Một lát nữa bạn sẽ được trải nghiệm sử dụng Callable. Quay lại phương thức submit() khác với execute() như thế nào? Đó kà submit() có trả về kết quả cuối cùng thông qua lớp Future. Future này giúp bạn xác định xem Thread Pool này đã hoàn thành xong hay chưa. Một lát nữa bạn cũng sẽ được trải nghiệm kết quả Future này.
• invokeAny() và invokeAll(): Một cách sử dụng khác của Thread Pool, ngoài việc dùng execute() hay submit() để đưa vào Pool từng Runnable hay Callable. Với 2 phương thức này bạn có thể truyền vào chúng danh sách các Callable. Phương thức invokeAny() sẽ thực thi các Callable theo quy luật khai báo Thead Pool như chúng ta làm quen ở các bài ví dụ trên, nhưng khi có bất kỳ Callable nào hoàn thành trong danh sách các Callable truyền vào đó, Thread Pool sẽ chấm dứt các Thread còn lại, dù cho chúng đã được đưa vào Pool và đang chờ thực thi. invokeAll() thì ngược lại, nó sẽ thực thi tất cả các Callable và chờ nhận các kết quả trả về của các Callable này thông qua danh sách các đối tượng Future.
• shutdown() và shutdownNow(): Khi bạn đã thêm các Runnable hay Callable vào trong Thread Pool, bạn có thể gọi lệnh shutdown() để xem như đóng Thread Pool đó lại, Thread Pool lúc này sẽ từ chối nhận thêm task nữa. Tại sao phải gọi shutdown()? Bạn nên biết rằng một ExecutorService không tự động kết thúc khi chúng thực thi hết các Thread, nó vẫn ở đó và khiến ứng dụng của bạn vẫn chạy mặc dù các Thread trong Thread Pool đã hoàn thành. Và lệnh shutdown() vừa giúp đóng Thread Pool lại, vừa giúp ExecutorService cũng kết thúc luôn khi nó hoàn thành nhiệm vụ. Tương tự, shutdownNow() cũng có công năng như vậy, chỉ khác một chỗ phương thức này buộc ExecutorService kết thúc ngay khi được gọi, lúc này đây các Thread chưa được thực thi sẽ bị buộc phải kết thúc theo ExecutorService.

Chúng ta sẽ không thực hành với phương thức execute() nữa, vì các bài thực hành trước đã sử dụng rồi. Hãy cùng thực hành với các phương thức còn lại của ExecutorService nào.

Bài Thực Hành Số 4 – submit(Runnable)

Do bài thực hành này cũng dùng đến Runnable, nên mình sẽ vẫn dùng lại MyRunnable đã được xây dựng từ các bài thực hành trước. Bạn chú ý một chút cách gọi submit() để trả về danh sách các Future như sau.

public static void main(String[] args) throws InterruptedException {

ExecutorService executorService = Executors.newFixedThreadPool(5);

List<Future> listFuture = new ArrayList<Future>(); // Khởi tạo danh sách các Future

 

for (int i = 1; i <= 10; i++) {

MyRunnable myRunnable = new MyRunnable("Runnable " + i);

// Bước này chúng ta dùng submit() thay vì execute()

Future future = executorService.submit(myRunnable);

listFuture.add(future); // Từng Future sẽ quản lý một Runnable

}

 

for (Future future : listFuture) {

try {

// Khi Thread nào kết thúc, get() của Future tương ứng sẽ trả về null

System.out.println(future.get());

} catch (ExecutionException | InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

 

// Phương thức này đã nói ở trên đây rồi

executorService.shutdown();

}

Bạn thấy rằng kết quả sau đây có in ra các dòng “null”. Chúng là các kết quả của các lời gọi future.get(). Và vì phương thức submit() này của ExecutorService giúp trả về các kết quả theo kiểu đồng bộ, tức là khi Thread kết thúc thì null mới được trả về thông qua future.get(), nên dù cho bạn đã gọi chúng rất sớm, chúng vẫn sẽ chỉ in ra null khi nào Thread kết thúc mà thôi.

Bài Thực Hành Số 5 – submit(Callable)

Bài thực hành này chúng ta cũng sẽ làm quen với Callable. Cơ bản thì Callable cũng khá giống với Runnable, nó cũng giúp để khởi tạo Thread trong ThreadPool. Nhưng Callable hữu ích hơn Runnable ở chỗ nó cho phép trả về một kết quả mà bạn định nghĩa sẵn. Chúng ta cùng đến với code khai báo một Callable như sau, rồi cùng nhau nói về nó tiếp theo nữa nhé. Callable này mình đặt tên là MyCallable.

public class MyCallable implements Callable<String> {

 

// Tên của Callable, giúp chúng ta phân biệt Runnable nào đang thực thi trong

// Thread Pool

private String name;

 

public MyCallable(String name) {

// Khởi tạo Callable với biến name truyền vào

this.name = name;

}

 

@Override

public String call() throws Exception {

System.out.println(name + " đang thực thi...");

 

// Giả lập thời gian chạy của Callable mất 2 giây

try {

Thread.sleep(2000);

} catch (InterruptedException e) {

e.printStackTrace();

}

 

// Trả kết quả về là một kiểu String

return name;

}

}

Dòng đầu tiên bạn thấy có khai báo lớp MyCallable kế thừa từ Callable<String>. Cách viết Callable<String> mình sẽ nói rõ hơn khi nói về kiến thức Generic sau này. Cơ bản bạn có thể hiểu Callable<String> là một Callable với một ràng buộc trong code của bạn phải trả về kiểu dữ liệu là String. Nếu bạn muốn trả về kiểu dữ liệu khác thì cứ thay String bằng kiểu dữ liệu bạn muốn là được.

Tương tự, vì Callable<String> ràng buộc bạn phải trả về kiểu String, nên code bên trong nó bạn phải Override phương thức call() với kiểu trả về là String luôn. Phương thức call() này của Callable thay cho run() bên Runnable, chỉ khác là call() có đòi hỏi kết quả trả về. Lúc này chúng ta trả về biến name luôn và in dòng kết thúc ở bên ngoài nhé.

Với việc khai báo Callable như trên, mình sẽ tiến hành gọi như sau ở phương thức main. Bạn xem.

public static void main(String[] args) throws InterruptedException {

ExecutorService executorService = Executors.newFixedThreadPool(5);

List<Future<String>> listFuture = new ArrayList<Future<String>>(); // Khởi tạo danh sách các Future

 

for (int i = 1; i <= 10; i++) {

// Dùng Callable thay cho Runnable

MyCallable myCallable = new MyCallable("Callable " + i);

 

Future<String> future = executorService.submit(myCallable);

listFuture.add(future); // Từng Future sẽ quản lý một Callable

}

 

for (Future future : listFuture) {

try {

// Khi Thread nào kết thúc, get() của Future tương ứng sẽ trả về kết quả mà Callable return

System.out.println(future.get() + " kết thúc");

} catch (ExecutionException | InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

 

// Phương thức này đã nói ở trên đây rồi

executorService.shutdown();

}

Có một điều chắc bạn cũng sẽ hơi nhức đầu. Đó là thay vì khai báo List<Future> như bài thực hành trước (bài học về List, ArrayList gì đó, chính là các Collection mình cũng sẽ nói sau), thì bây giờ lại phải khai báo List<Future<String>>. Bạn nên biết rằng, chắc chắn cái đoạn <String> ở khai báo List này có liên quan mật thiết với <String> ở khai báo lớp Callable trên kia rồi, và như đã nói, mình sẽ nói sau ở bài về Generic. Bây giờ khi thực thi chương trình, kết quả không nằm ngoài tiên liệu của chúng ta. Có điều do phương thức future.get() có một độ trễ nhất định, nên việc in ra Callable nào kết thúc không ngay tức thời, nên nhìn vào kết quả console sau, bạn sẽ thấy dường như có nhiều Thread được thêm vào Thread Pool hơn khai báo 5 Thread ban đầu.

Bài Thực Hành Số 6 – invokeAny()

Do bài học đã quá dài, nên mục invokeAny() mình chỉ cho các bạn xem code, và kết quả thực thi chương trình. Ý nghĩa của phương thức này mình đã giải thích trên kia rồi nhé. Có khác một tí là ví dụ này mình không dùng đến MyCallable nữa, mình muốn khai báo các Callable trực tiếp trong vòng for bằng cách dùng đến lớp vô danh cho code được gọn nhẹ.

public static void main(String[] args) throws InterruptedException, ExecutionException {

// Khai báo một Thread Pool thông qua newSingleThreadExecutor() của Executors

ExecutorService executorService = Executors.newSingleThreadExecutor();

List<Callable<String>> listCallable = new ArrayList<Callable<String>>(); // Khởi tạo danh sách các Callable

 

for (int i = 1; i <= 5; i++) {

final int _i = i;

// Khởi tạo từng Callable

listCallable.add(new Callable<String>() {

 

@Override

public String call() throws Exception {

// Trả về kết quả ở mỗi Callable

return "Callable " + _i;

}

});

}

 

// Callable nào kết thúc ở đây cũng sẽ dừng luôn Thread Pool

String result = executorService.invokeAny(listCallable);

System.out.println("Result: " + result);

 

// Phương thức này đã nói ở trên đây rồi

executorService.shutdown();

}

Bài Thực Hành Số 7 – invokeAll()

public static void main(String[] args) throws InterruptedException, ExecutionException {

// Khai báo một Thread Pool thông qua newSingleThreadExecutor() của Executors

ExecutorService executorService = Executors.newSingleThreadExecutor();

List<Callable<String>> listCallable = new ArrayList<Callable<String>>(); // Khởi tạo danh sách các Callable

 

for (int i = 1; i <= 5; i++) {

final int _i = i;

// Khởi tạo từng Callable

listCallable.add(new Callable<String>() {

 

@Override

public String call() throws Exception {

// Trả về kết quả ở mỗi Callable

return "Callable " + _i;

}

});

}

 

// Dùng Future để lấy về danh sách các kết quả trả về từ mỗi Callable

List<Future<String>> futures = executorService.invokeAll(listCallable);

for(Future<String> future : futures) {

System.out.println("Result: " + future.get());

}

 

// Phương thức này đã nói ở trên đây rồi

executorService.shutdown();

}