Dừng một ExecutorService

Hãy xem xét một chương trình đơn giản:

public static void main(String[] args) throws Exception {
	// Create an ExecutorService with a fixed number of threads: three
	ExecutorService service = Executors.newFixedThreadPool(3);
 
	// Pass a simple Runnable task to the ExecutorService
	service.submit(() -> System.out.println("done"));
}

Chạy chương trình tạo ra đầu ra giao diện điều khiển mà chúng tôi mong đợi:

Nhưng điều này không theo sau kết quả mà chúng ta thường thấy trong IntelliJ IDEA:

Chúng ta thường thấy điều đó khi một chương trình kết thúc.

Tại sao điều đó lại xảy ra?

Mô tả của phương thức newFixedThreadPool() cho chúng ta biết rằng các luồng được tạo bằng ExecutorService tiếp tục tồn tại cho đến khi chúng bị dừng một cách rõ ràng. Điều đó có nghĩa là vì chúng ta đã chuyển một tác vụ cho ExecutorService , một luồng được tạo để thực thi nó và luồng đó tiếp tục tồn tại ngay cả sau khi tác vụ được thực hiện.

Dừng tại ExecutorService

Do đó, chúng ta cần "tắt" (hoặc dừng) ExecutorService . Chúng ta có thể làm điều này theo hai cách:

1. void shutdown() — sau khi phương thức này được gọi, ExecutorService ngừng nhận công việc mới. Tất cả các tác vụ đã gửi trước đó tới ExecutorService sẽ tiếp tục chạy.

   
   public static void main(String[] args) throws Exception {
   ExecutorService service = Executors.newFixedThreadPool(3);
       	service.submit(() -> System.out.println("task 1"));
       	service.submit(() -> System.out.println("task 2"));
       	service.shutdown();
       	// A RejectedExecutionException will occur here
       	service.submit(() -> System.out.println("task 3"));
   }
   

2. List<Runnable> shutdownNow() — Phương thức này cố gắng dừng các công việc hiện đang hoạt động. Các nhiệm vụ vẫn đang chờ đến lượt của chúng sẽ bị loại bỏ và được trả về dưới dạng danh sách Runnables .

   
   public static void main(String[] args) throws Exception {
       ExecutorService service = Executors.newFixedThreadPool(5);
       List.of(1, 2, 3, 4, 5, 6, 7, 8).forEach(i -> service.submit(() -> System.out.println(i)));
       List<Runnable> runnables = service.shutdownNow();
       runnables.forEach(System.out::println);
   }
   

Đầu ra:

Đầu ra sẽ khác nhau từ lần chạy này sang lần chạy khác. Có 2 loại dòng trong đầu ra:

• Một số có nghĩa là ExecutorService quản lý để xử lý tác vụ tương ứng, hiển thị số từ danh sách chúng tôi đã sử dụng để tạo tác vụ.

• Kết quả gọi phương thức toString() trên đối tượng FutureTask . Các đối tượng này là các tác vụ đã được gửi tới ExecutorService nhưng chưa được xử lý.

Đầu ra có một sắc thái thú vị khác. Trong một thế giới lý tưởng, đầu tiên chúng ta sẽ thấy tất cả các số được hiển thị, tiếp theo là các đối tượng FutureTask . Nhưng các vấn đề đồng bộ hóa làm lộn xộn các dòng trong đầu ra.

Các phương pháp khác

ExecutorService có thêm một số phương pháp liên quan đến việc ngăn chặn nó:

1. boolean awaitTermination(long timeout, TimeUnit unit) — phương thức này chặn luồng gọi nó. Việc chặn kết thúc ngay khi bất kỳ một trong ba sự kiện sau xảy ra:

   • sau khi phương thức shutdown() được gọi, tất cả các công việc đang hoạt động và tất cả các tác vụ theo lịch trình đã được thực thi;
   • thời gian chờ được xác định bởi các tham số phương pháp đã trôi qua;
   • chuỗi được gọi là phương thức awaitTermination() bị chấm dứt.

   Phương thức trả về true nếu ExecutorService bị dừng trước khi hết thời gian chờ và trả về false nếu hết thời gian chờ.

   
   public static void main(String[] args) throws Exception {
   	ExecutorService service = Executors.newFixedThreadPool(2);
   	service.submit(() -> System.out.println("task 1"));
   	service.submit(() -> System.out.println("task 2"));
   	service.submit(() -> System.out.println("task 3"));
   	service.shutdown();
   	System.out.println(service.awaitTermination(1, TimeUnit.MICROSECONDS));
   }
   

2. boolean isShutdown() — Trả về true nếu phương thức shutdown() hoặc shutdownNow() đã được gọi trên ExecutorService .

   
   public static void main(String[] args) throws Exception {
   	ExecutorService service = Executors.newFixedThreadPool(2);
   	service.submit(() -> System.out.println("task 1"));
   	service.submit(() -> System.out.println("task 2"));
   	service.submit(() -> System.out.println("task 3"));
   	System.out.println(service.isShutdown());
   	service.shutdown();
   	System.out.println(service.isShutdown());
   }
   

3. boolean isTerminated() — Trả về true nếu phương thức shutdown() hoặc shutdownNow() đã được gọi trên ExecutorService và tất cả các tác vụ đã hoàn tất.

   
   public static void main(String[] args) throws Exception {
       ExecutorService service = Executors.newFixedThreadPool(5);
       List.of(1, 2, 3, 4, 5, 6, 7, 8).forEach(i -> service.submit(() -> System.out.println(i)));
       service.shutdownNow();
       System.out.println(service.isTerminated());
   }
   

Mã ví dụ sử dụng các phương pháp này:

public static void main(String[] args) throws Exception {
   ExecutorService service = Executors.newFixedThreadPool(16);
   Callable<String> task = () -> {
       Thread.sleep(1);
       return "Done";
   };
 
   // Add 10,000 tasks to the queue
   List<Future<String>> futures = IntStream.range(0, 10_000)
           .mapToObj(i -> service.submit(task))
           .collect(Collectors.toList());
   System.out.printf("%d tasks were submitted for execution.%n", futures.size());
 
   // Attempt to shut down
   service.shutdown();
   // Wait 100 milliseconds to finish the work
   if (service.awaitTermination(100, TimeUnit.MILLISECONDS)) {
       System.out.println("All tasks completed!");
   } else {
       // Stop forcibly
       List<Runnable> notExecuted = service.shutdownNow();
       System.out.printf("%d tasks were not started.%n", notExecuted.size());
   }
 
   System.out.printf("Total tasks completed: %d.%n", futures.stream().filter(Future::isDone).count());
}

Đầu ra (khác nhau giữa các lần chạy):