Làm thế nào để chờ đợi cho tất cả các tác vụ trong ThreadPoolExecutor để kết thúc mà không cần tắt Trình thực thi?

Question

Tôi không thể sử dụng shutdown() và awaitTermination() vì các tác vụ mới có thể sẽ được thêm vào ThreadPoolExecutor trong khi chờ đợi. Vì vậy, tôi đang tìm một cách để chờ đợi cho đến khi ThreadPoolExecutor đã làm trống hàng đợi của nó và hoàn thành tất cả các nhiệm vụ của nó mà không dừng các nhiệm vụ mới được thêm vào trước thời điểm đó.

Nếu có bất kỳ sự khác biệt nào, điều này là dành cho Android.

Cảm ơn

Cập nhật: Nhiều tuần sau đó sau khi xem xét lại này, tôi phát hiện ra rằng một CountDownLatch sửa đổi làm việc tốt hơn cho tôi trong trường hợp này. Tôi sẽ giữ câu trả lời được đánh dấu bởi vì nó áp dụng nhiều hơn cho những gì tôi hỏi.

Answer 1

Nếu bạn muốn biết khi nào một tác vụ nào đó hoàn thành hoặc một loạt tác vụ nhất định, bạn có thể sử dụng ExecutorService.submit(Runnable). Gọi phương thức này trả về một đối tượng Future có thể được đặt vào một Collection mà chủ đề chính của bạn sau đó sẽ lặp qua gọi số Future.get() cho mỗi đối tượng. Điều này sẽ khiến cho chuỗi chính của bạn ngừng thực hiện cho đến khi ExecutorService xử lý tất cả các tác vụ Runnable.

Collection<Future<?>> futures = new LinkedList<Future<?>>(); 
futures.add(executorService.submit(myRunnable)); 
for (Future<?> future:futures) { 
    future.get(); 
} 

Answer 2

Có lẽ bạn đang tìm kiếm một CompletionService để quản lý các lô nhiệm vụ, xem thêm this answer.

Answer 3

(Đây là một nỗ lực để tái tạo trước đó, câu trả lời đã xóa Thilo với điều chỉnh của riêng tôi.)

Tôi nghĩ rằng bạn có thể cần phải làm rõ câu hỏi của bạn vì có một tình trạng vô hạn tiềm ẩn ... tại một số điểm bạn có để quyết định ngừng hoạt động của bạn, và vào thời điểm đó nó sẽ không chấp nhận bất kỳ nhiệm vụ nào nữa. Câu hỏi của bạn dường như ngụ ý rằng bạn muốn đợi cho đến khi bạn biết rằng sẽ không có tác vụ nào khác được gửi, mà bạn chỉ có thể biết trong mã ứng dụng của riêng bạn.

Câu trả lời sau đây sẽ cho phép bạn chuyển đổi thuận lợi sang TPE mới (vì bất kỳ lý do gì), hoàn thành tất cả các nhiệm vụ hiện đang được gửi và không từ chối nhiệm vụ mới cho TPE mới. Nó có thể trả lời câu hỏi của bạn. @ Thilo cũng có thể.

Giả sử bạn đã xác định ở đâu đó một TPE có thể nhìn thấy được sử dụng như vậy:

AtomicReference<ThreadPoolExecutor> publiclyAvailableTPE = ...; 

Sau đó bạn có thể viết các thói quen TPE hoán đổi như vậy. Nó cũng có thể được viết bằng một phương pháp đồng bộ, nhưng tôi nghĩ rằng đây là đơn giản:

void rotateTPE() 
{ 
    ThreadPoolExecutor newTPE = createNewTPE(); 
    // atomic swap with publicly-visible TPE 
    ThreadPoolExecutor oldTPE = publiclyAvailableTPE.getAndSet(newTPE); 
    oldTPE.shutdown(); 

    // and if you want this method to block awaiting completion of old tasks in 
    // the previously visible TPE 
    oldTPE.awaitTermination(); 
} 

Ngoài ra, nếu bạn thực sự không đùa muốn giết các hồ bơi thread, sau đó bên nộp bạn sẽ cần phải đối phó với nhiệm vụ từ chối tại một lúc nào đó, và bạn có thể sử dụng null cho TPE mới:

void killTPE() 
{ 
    ThreadPoolExecutor oldTPE = publiclyAvailableTPE.getAndSet(null); 
    oldTPE.shutdown(); 

    // and if you want this method to block awaiting completion of old tasks in 
    // the previously visible TPE 
    oldTPE.awaitTermination(); 
} 

Mà có thể gây ra vấn đề thượng nguồn, người gọi sẽ cần phải biết phải làm gì với một null.

Bạn cũng có thể hoán đổi với một TPE giả mà chỉ đơn giản từ chối mọi thực thi mới, nhưng điều đó tương đương với những gì xảy ra nếu bạn gọi shutdown() trên TPE.

Answer 4

Tình huống của tôi là trình thu thập thông tin web để tìm nạp một số thông tin từ một trang web, sau đó xử lý chúng. Một ThreadPoolExecutor được sử dụng để tăng tốc quá trình bởi vì nhiều trang có thể được nạp trong thời gian. Vì vậy, nhiệm vụ mới sẽ được tạo trong tác vụ hiện tại vì trình thu thập thông tin sẽ theo các siêu liên kết trong mỗi trang. Vấn đề là như nhau: thread chính không biết khi nào tất cả các nhiệm vụ được hoàn thành và nó có thể bắt đầu xử lý kết quả. Tôi sử dụng một cách đơn giản để xác định điều này. Nó không phải là rất tao nhã nhưng làm việc trong trường hợp của tôi:

while (executor.getTaskCount()!=executor.getCompletedTaskCount()){ 
    System.err.println("count="+executor.getTaskCount()+","+executor.getCompletedTaskCount()); 
    Thread.sleep(5000); 
} 
executor.shutdown(); 
executor.awaitTermination(60, TimeUnit.SECONDS); 

Answer 5

Nếu bạn không muốn sử dụng shutdown, hãy làm theo bên dưới cách tiếp cận:

1. Duyệt qua tất cả Future nhiệm vụ từ nộp trên ExecutorService và kiểm tra tình trạng với chặn cuộc gọi get() trên Future đối tượng theo đề nghị của Tim Bender

2. Sử dụng một trong

   1. Sử dụng invokeAll trên ExecutorService
   2. Sử dụng CountDownLatch
   3. Sử dụng ForkJoinPool hoặc newWorkStealingPool của Executors (kể từ java 8)

invokeAll() về dịch vụ chấp hành viên cũng đạt được cùng một mục đích của CountDownLatch

liên quan SE câu hỏi:

How to wait for a number of threads to complete?

Answer 6

Bạn có thể gọi waitTillDone() trên Runner lớp:

Runner runner = Runner.runner(10); 

runner.runIn(2, SECONDS, runnable); 
runner.run(runnable); // each of this runnables could submit more tasks 

runner.waitTillDone(); // blocks until all tasks are finished (or failed) 

// and now reuse it 

runner.runIn(500, MILLISECONDS, callable); 

runner.waitTillDone(); 
runner.shutdown(); 

Để sử dụng nó thêm gradle phụ thuộc/maven này để dự án của bạn: 'com.github.matejtymes:javafixes:1.0'

Để biết thêm chi tiết, hãy xem tại đây: https://github.com/MatejTymes/JavaFixes hoặc tại đây: http://matejtymes.blogspot.com/2016/04/executor-that-notifies-you-when-task.html