리소스 누출 위험없이 Java 반복자 내에서 ExecutorService를 사용하는 방법

Question

원격 위치의 항목을 나열하는 Java 반복기가 있습니다. 항목 목록이 "페이지"로 나오고 "다음 페이지 가져 오기"작업이 다소 느립니다. (필자의 반복자는 S3Find이고 Amazon S3의 객체를 나열합니다.

그래서 일을 빠르게하기 위해 하나의 목록 페이지를 미리 가져오고 싶었습니다. 이렇게하려면 항목의 "페이지"를 미리 가져 오기 위해 ExecutorService 및 Callable/Future 패턴을 사용했습니다. 문제는 반복기의 호출자가 수업을 알리지 않고 언제든지 작업을 포기할 수 있다는 것입니다. 예를 들어, 다음 루프 고려 :

for (S3URL f : new S3Find(topdir).withRecurse(true)) { 
    // do something with f 
    if (some_condition) break; 
} 

결과를하는 포함하는 더 이상 언급이없는 경우에도 나는 Callable을 제출하는 데 사용하는 ExecutorService 살아 실행 남아, 자원 누수가 S3Find (그리고 다음 프리 페치가 완료 되더라도).

이 문제를 해결하는 적절한 방법은 무엇입니까? 잘못된 접근 방식을 사용하고 있습니까? 방금 ExecutorService을 포기하고 모든 프리 페치에 대해 새로운 베어 스레드를 사용해야합니까? (그리고 프리 페치가 완료되면 스레드를 죽이십시오)? 페이지를 가져올 때마다 대략 500ms가 소요되므로 매번 새로운 스레드를 만드는 것은 아마도 비교할 때 무시할 만하다. 한 가지가 있습니다 은 발신자가 S3Find에 반복적으로 수행되었음을 알리도록 요구하는 것입니다 (일부는 잊어 버릴 수 있음).

/** 
* This class holds one ObjectListing (one "page"), and also pre-fetches 
* the next page using a {@link S3Find#NextPageGetter} Callable on a 
* separate thread. 
*/ 
private static class Pager { 
    private final AmazonS3 s3; 
    private ObjectListing currentList; 
    private Future<ObjectListing> future; 
    private final ExecutorService exec; 
    public Pager(AmazonS3 s3, ListObjectsRequest request) { 
     this.s3 = s3; 
     currentList = s3.listObjects(request); 
     exec = Executors.newSingleThreadExecutor(); 
     future = submitPrefetch(); 
    } 
    public ObjectListing getCurrentPage() { 
     return currentList; 
    } 
    /** 
    * Move currentList to the next page, and returns it. 
    */ 
    public ObjectListing getNextPage() { 
     if (future == null) return null; 
     try { 
      currentList = future.get(); 
      future = submitPrefetch(); 
     } catch (InterruptedException|ExecutionException e) { 
      e.printStackTrace(); 
     } 
     return currentList; 
    } 
    private Future<ObjectListing> submitPrefetch() { 
     if (currentList == null || !currentList.isTruncated()) { 
      exec.shutdown(); 
      return null; 
     } else { 
      NextPageGetter worker = new NextPageGetter(s3, currentList); 
      return exec.submit(worker); 
     } 
    } 
} 

/** 
* This class retrieves the "next page" of a truncated ObjectListing. 
* It is meant to be called in a Callable/Future pattern. 
*/ 
private static class NextPageGetter implements Callable<ObjectListing> { 
    private final ObjectListing currentList; 
    private final AmazonS3 s3; 

    public NextPageGetter(AmazonS3 s3, ObjectListing currentList) { 
     super(); 
     this.s3 = s3; 
     this.currentList = currentList; 
     if (currentList == null || !currentList.isTruncated()) { 
      throw new IllegalArgumentException(currentList==null ? 
         "null List" : "List is not truncated"); 
     } 
    } 

    @Override 
    public ObjectListing call() throws Exception { 
     ObjectListing nextList = s3.listNextBatchOfObjects(currentList); 
     return nextList; 
    } 
} 

Answer 1

이 내가 몇 번에 실행 한 고전적인 문제가 :

여기 (S3Find 내부) 현재 프리 페치 코드입니다. 데이터베이스 연결로 나에게 일어난다.

ExecutorService를 포기하고 모든 프리 페치에 대해 새로운 베어 스레드를 사용해야합니까? (프리 페치가 끝나면 스레드를 종료해야합니까?)

나는 유일한 옵션이라고 생각합니다. 나는 실을 죽이는 것을 괴롭히지 않을 것이다. 그냥 일을 끝내고 배경에서 죽게하십시오. 다음 페이지의 새 스레드를 포크하십시오. 스레드와 결합하여 어떤 종류의 공통적 인 AtomicReference (또는 뭔가)을 사용하여 S3Find 호출자와 스레드간에 결과 목록을 공유해야합니다.

호출자가 S3Find에 iterating이 수행되었음을 명시 적으로 알리는 것이 필요하지 않은 것 (일부는 잊어 버릴 수 있음)입니다.

나는 시도/마침내 close() 방법의 일종을 호출없이 발신자이 "오른쪽"을 할 수있는 쉬운 방법을 볼 수 없습니다. 어떻게 든 Javadocs에서 명시 적으로 할 수는 없습니까? 그게 내 ORMLite database iterators에서 한 것입니다. S3Find.close()에서 다음

S3Find s3Find = new S3Find(topdir).withRecurse(true); 
try { 
    for (S3URL f : s3Find) { 
     ... 
    } 
} finally { 
    s3Find.close(); 
} 

:

가 언어 어떤 Closeable 자원을 자동 닫을 try with resources construct 추가 한 자바 7에서

public void close() { 
    exec.shutdown(); 
} 

. 그것은 큰 승리입니다.

Answer 2

나는 위에서 설명한 것처럼 맨손으로 스레드를 사용하는 동안 아주 간단하고 매우 초기 버전에 가까운 솔루션을 가지고 있다고 생각합니다. 그것은 여전히 ​​좋은 Callable 패턴을 활용하지만 Future 대신 FutureTask을 사용하고 ExecutorService을 전혀 사용하지 않습니다.

내가 놓친 핵심 사항은 FutureTask은 Runnable이며 실제로는 new Thread(task)을 통해 시작할 수 있습니다.

NextPageGetter worker = new NextPageGetter(s3, currentList); 
FutureTask<ObjectListing> future = new FutureTask<>(worker); 
new Thread(future).start(); 

하고 나중에 : 즉

currentList = future.get(); 

지금 모든 자원을 행복하게 반복자가 소진 여부, 배치된다. 실제로 FutureTask이 완료되면 스레드가 사라집니다.

는 완벽을 위해, 여기에 (만 class Pager가 변경) 수정 된 코드입니다 : 대답 회색에 대한

/** 
* This class holds one ObjectListing (one "page"), and also pre-fetches the next page 
* using a {@link S3Find#NextPageGetter} Callable on a separate thread. 
*/ 
private static class Pager { 
    private final AmazonS3 s3; 
    private ObjectListing currentList; 
    private FutureTask<ObjectListing> future; 
    public Pager(AmazonS3 s3, ListObjectsRequest request) { 
     this.s3 = s3; 
     currentList = s3.listObjects(request); 
     future = submitPrefetch(); 
    } 
    public ObjectListing getCurrentPage() { 
     return currentList; 
    } 
    /** 
    * Move currentList to the next page, and returns it. 
    */ 
    public ObjectListing getNextPage() { 
     if (future == null) return null; 
     try { 
      currentList = future.get(); 
      future = submitPrefetch(); 
     } catch (InterruptedException|ExecutionException e) { 
      e.printStackTrace(); 
     } 
     return currentList; 
    } 
    private FutureTask<ObjectListing> submitPrefetch() { 
     if (currentList == null || !currentList.isTruncated()) { 
      return null; 
     } else { 
      NextPageGetter worker = new NextPageGetter(s3, currentList); 
      FutureTask<ObjectListing> f = new FutureTask<>(worker); 
      new Thread(f).start(); 
      return f; 
     } 
    } 
}