Cách đợi danh sách `Tương lai` được tạo bằng cách sử dụng `ExecutorServices` khác nhau

Question

Ok, vì vậy tôi biết câu trả lời/nhận xét đầu tiên ở đây sẽ là" sử dụng một số ExecutorService và sử dụng invokeAll ". Tuy nhiên, có một lý do chính đáng (mà tôi sẽ không cho người khác) để chúng tôi giữ riêng các hồ sợi.

Vì vậy, tôi có một danh sách các hồ bơi chủ đề (ExecutorServices) và những gì tôi cần làm là gọi Callable khác nhau trên mỗi nhóm hồ sơ bằng cách sử dụng submit (không có vấn đề ở đó). Bây giờ tôi có bộ sưu tập này gồm Future trường hợp, mỗi phiên bản được tạo trên ExecutorService riêng biệt và tôi muốn đợi tất cả chúng hoàn thành (và có thể cung cấp thời gian chờ mà không được hủy bỏ).

Có lớp học hiện tại nào sẽ thực hiện việc này hay không (bao gồm danh sách các trường hợp Future và chờ đợi cho đến khi tất cả được thực hiện)? Nếu không, đề xuất về một cơ chế hiệu quả sẽ được đánh giá cao.

Nghĩ đến việc gọi get với thời gian chờ cho mỗi lần nhưng phải tính toán tổng thời gian được chuyển cho mỗi cuộc gọi.

Tôi thấy bài đăng này Wait Until Any of Future is Done nhưng điều này mở rộng Future thay vì gói một danh sách.

Answer 1

mỗi Louis' bình luận, những gì tôi đang tìm kiếm được Futures.successfulAsList

này cho phép tôi chờ đợi cho tất cả để hoàn thành và sau đó kiểm tra cho bất kỳ tương lai mà không thành công.

Nguyên tắc dung nham!

Answer 2

Tôi không nghĩ JDK cung cấp API trực tiếp cho phép bạn làm điều đó. Tuy nhiên, tôi nghĩ nó cũng không kém phần đơn giản để tạo ra một phương thức đơn giản để thực hiện điều này. Bạn có thể muốn xem xét việc thực hiện AbstractExecutorService.invokeAll() để có được một ý tưởng rằng điều này có thể được thực hiện.

Về cơ bản, bạn sẽ gọi đến future.get() trên mỗi tương lai, giảm thời gian chờ đợi để đợi kết quả mỗi lần và trước khi trở về từ phương thức hủy tất cả các tương lai chưa thanh toán.

Answer 3

Có lẽ tôi đã không thực sự hiểu. Tuy nhiên, với tôi, nó vẫn nghe có vẻ đơn giản như

public <V> List<V> get(List<Future<V>> futures, long timeout, TimeUnit unit) 
      throws InterruptedException, ExecutionException, TimeoutException { 
    List<V> result = new ArrayList<V>(); 
    long end = System.nanoTime() + unit.toNanos(timeout); 
    for (Future<V> f: futures) { 
     result.add(f.get(end - System.nanoTime(), TimeUnit.NANOSECONDS)); 
    } 
    return result; 
} 

Tôi có sai với điều đó không?

Câu hỏi mà bạn liên kết phức tạp hơn nhiều, tôi nghĩ, vì họ chỉ muốn chờ nhanh nhất, và tất nhiên không có ý tưởng nào sẽ nhanh nhất.

Answer 4

Điều này có thể sử dụng một số dọn dẹp, nhưng nó sẽ giải quyết vấn đề của bạn. (Một số đóng gói bỏ qua thời gian và không gian):

public static <T> LatchWithWrappedCallables<T> wrapCallables(Collection<Callable<T>> callablesToWrap) 
{ 
    CountDownLatch latch = new CountDownLatch(callablesToWrap.size()); 
    List<Callable<T>> wrapped = new ArrayList<Callable<T>>(callablesToWrap.size()); 
    for (Callable<T> currCallable : callablesToWrap) 
    { 
     wrapped.add(new CallableCountdownWrapper<T>(currCallable, latch)); 
    } 

    LatchWithWrappedCallables<T> returnVal = new LatchWithWrappedCallables<T>(); 
    returnVal.latch = latch; 
    returnVal.wrappedCallables = wrapped; 
    return returnVal; 
} 

public static class LatchWithWrappedCallables<T> 
{ 
    public CountDownLatch latch; 
    public Collection<Callable<T>> wrappedCallables; 
} 

public static class CallableCountdownWrapper<T> implements Callable<T> 
{ 
    private final Callable<T> wrapped; 

    private final CountDownLatch latch; 

    public CallableCountdownWrapper(Callable<T> wrapped, CountDownLatch latch) 
    { 
     this.wrapped = wrapped; 
     this.latch = latch; 
    } 

    @Override 
    public T call() throws Exception 
    { 
     try 
     { 
      return wrapped.call(); 
     } 
     finally 
     { 
      latch.countDown(); 
     } 
    } 
} 

Sau đó, mã của bạn sẽ gọi nó là như thế này:

Collection<Callable<String>> callablesToWrap = [Your callables that you need to wait for here]; 
LatchWithWrappedCallables<String> latchAndCallables = wrapCallables(callablesToWrap); 

[Submit the wrapped callables to the executors here] 

if(latchAndCallables.latch.await(timeToWaitInSec, TimeUnit.SECONDS)) 
{ 
    [Handling for timeout here] 
}