Future <Future<T>>을 미래의 <T>으로 바꾸는 쉬운 방법이 있습니까?

Question

요청을 다른 스레드에 제출할 수도 있고하지 않을 수도있는 다른 스레드에 요청을 제출하는 코드가 있습니다. 반환 유형은 Future<Future<T>>입니다. 앞으로이 전체 체인의 완성을 기다리는 Future<T>으로 즉시 전환 할 수있는 비 치명적인 방법이 있습니까?

다른 재미있는 동시성 작업을 처리하고 Google Collections 및 그 잘 작동하는 작업을 대체하기 위해 이미 Guava 라이브러리를 사용하고 있지만이 경우 뭔가를 찾지 못하는 것 같습니다.)

public class UnwrapFuture<T> implements Future<T> { 
    Future<Future<T>> wrappedFuture; 

    public UnwrapFuture(Future<Future<T>> wrappedFuture) { 
     this.wrappedFuture = wrappedFuture; 
    } 

    public boolean cancel(boolean mayInterruptIfRunning) { 
     try { 
      return wrappedFuture.get().cancel(mayInterruptIfRunning); 
     } catch (InterruptedException e) { 
      //todo: do something 
     } catch (ExecutionException e) { 
      //todo: do something 
     } 
    } 
    ... 
} 

당신은 (얻을 예외를 처리해야 올릴 수 있지만, 다른 방법은 없습니다 :

Answer 1

구아바 13.0은 이렇게하기 위해 Futures.dereference을 추가합니다. 보통 Future<Future>이 아닌 ListenableFuture<ListenableFuture>이 필요합니다. (Future으로 작동하면 makeListenable 호출이 필요합니다. 각 호출에는 작업 수명 동안 전용 스레드가 필요합니다 (메서드의 새 이름 인 JdkFutureAdapters.listenInPoolThread에 의해 명확 해짐)

Answer 2

당신은 같은 클래스를 만들 수 있습니다.

Answer 3

이것은 내 첫 찌르기 였지만 그걸로는 많은 잘못이 있다는 것을 확신합니다. Futures.compress(f)과 같은 것으로 바꾸는 것만으로도 행복 할 것입니다.

public class CompressedFuture<T> implements Future<T> { 
    private final Future<Future<T>> delegate; 

    public CompressedFuture(Future<Future<T>> delegate) { 
     this.delegate = delegate; 
    } 

    @Override 
    public boolean cancel(boolean mayInterruptIfRunning) { 
     if (delegate.isDone()) { 
      return delegate.cancel(mayInterruptIfRunning); 
     } 
     try { 
      return delegate.get().cancel(mayInterruptIfRunning); 
     } catch (InterruptedException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } catch (ExecutionException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } 
    } 

    @Override 
    public T get() throws InterruptedException, ExecutionException { 
     return delegate.get().get(); 
    } 

    @Override 
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { 
     long endTime = System.currentTimeMillis() + unit.toMillis(timeout); 
     Future<T> next = delegate.get(timeout, unit); 
     return next.get(endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS); 
    } 

    @Override 
    public boolean isCancelled() { 
     if (!delegate.isDone()) { 
      return delegate.isCancelled(); 
     } 
     try { 
      return delegate.get().isCancelled(); 
     } catch (InterruptedException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } catch (ExecutionException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } 
    } 

    @Override 
    public boolean isDone() { 
     if (!delegate.isDone()) { 
      return false; 
     } 
     try { 
      return delegate.get().isDone(); 
     } catch (InterruptedException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } catch (ExecutionException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } 
    } 
} 

Answer 4

나는 이것이 미래의 계약을 이행하기 위해 할 수 있다고 생각합니다. 나는 그것이 계약을 준수하는지 확신 할 수 있도록 가능한 한 험악한 태도를 취했다. 특히 타임 아웃을 사용하여 get을 구현하는 것은 아닙니다.

import java.util.concurrent.*; 

public class Futures { 
    public <T> Future<T> flatten(Future<Future<T>> future) { 
    return new FlattenedFuture<T>(future); 
    } 

    private static class FlattenedFuture<T> implements Future<T> { 
    private final Future<Future<T>> future; 

    public FlattenedFuture(Future<Future<T>> future) { 
     this.future = future; 
    } 

    public boolean cancel(boolean mayInterruptIfRunning) { 
     if (!future.isDone()) { 
     return future.cancel(mayInterruptIfRunning); 
     } else { 
     while (true) { 
      try { 
      return future.get().cancel(mayInterruptIfRunning); 
      } catch (CancellationException ce) { 
      return true; 
      } catch (ExecutionException ee) { 
      return false; 
      } catch (InterruptedException ie) { 
      // pass 
      } 
     } 
     } 
    } 

    public T get() throws InterruptedException, 
          CancellationException, 
          ExecutionException 
    { 
     return future.get().get(); 
    } 

    public T get(long timeout, TimeUnit unit) throws InterruptedException, 
                CancellationException, 
                ExecutionException, 
                TimeoutException 
    { 
     if (future.isDone()) { 
     return future.get().get(timeout, unit); 
     } else { 
     return future.get(timeout, unit).get(0, TimeUnit.SECONDS); 
     } 
    } 

    public boolean isCancelled() { 
     while (true) { 
     try { 
      return future.isCancelled() || future.get().isCancelled(); 
     } catch (CancellationException ce) { 
      return true; 
     } catch (ExecutionException ee) { 
      return false; 
     } catch (InterruptedException ie) { 
      // pass 
     } 
     } 
    } 

    public boolean isDone() { 
     return future.isDone() && innerIsDone(); 
    } 

    private boolean innerIsDone() { 
     while (true) { 
     try { 
      return future.get().isDone(); 
     } catch (CancellationException ce) { 
      return true; 
     } catch (ExecutionException ee) { 
      return true; 
     } catch (InterruptedException ie) { 
      // pass 
     } 
     } 
    } 
    } 
} 

Answer 5

구아바 라이브러리를 사용하는 다른 가능한 구현은 훨씬 간단합니다.

import java.util.concurrent.*; 
import com.google.common.util.concurrent.*; 
import com.google.common.base.*; 

public class FFutures { 
    public <T> Future<T> flatten(Future<Future<T>> future) { 
    return Futures.chain(Futures.makeListenable(future), new Function<Future<T>, ListenableFuture<T>>() { 
     public ListenableFuture<T> apply(Future<T> f) { 
     return Futures.makeListenable(f); 
     } 
    }); 
    } 
}