Наблюдение за будущим <T> объектов для заполнения с использованием ExecutorService

Question

У меня есть программа, в которой я загружаю файлы с S3 с использованием объекта Callable, переданного в ExecutorService. Файлы большие, и для полной загрузки требуется несколько минут. Имеет ли смысл создавать еще один класс Callable, который берет будущее от загрузчика и наблюдает за его завершением? Моя конечная цель - добавить все полные загрузки в расположенный в центре список в кеше.

Например:

public void add(final String s3urlToDownload){ 

    Future<S3Object> futureS3obj = cachedPoolExecutor.submit(new S3Downloader(s3urlToDownload)); 

    // Instead of calling futureS3obj.get() and waiting, submit the Future to the "Watcher" service. 
    // Within FutureWatcher, the S3Object will be added to the List once the download is complete. 
    cachedPoolExecutor.submit(new FutureWatcher(downloadedList, futureS3obj)) 

} 

Answer 1

Вот некоторые поддельные объекты, используемые для иллюстрации. «Загрузка» просто случайные сны:

// for illustration only 
class S3Object { 
    String id; 
} 

// for illustration only 
class S3Downloader { 

    public S3Object download(String url) { 
     int min = 2; 
     int max = 5; 
     Random rand = new Random(); 
     int random = rand.nextInt((max - min) + 1) + min; 

     try { Thread.sleep(1000 * random); } catch (Exception ex) {} 
     S3Object result = new S3Object(); 
     result.id = url; 
     return result; 
    } 
} 

Мы можем определить задачу, которая загружает файл, обновляет список (потокобезопасный), и декрементирует CountDownLatch:

class MyTask implements Runnable { 
    private final List<S3Object> list; 
    private final CountDownLatch latch; 
    private final String url; 

    public MyTask(List<S3Object> list, CountDownLatch latch, String url) { 
     this.list = list; 
     this.latch = latch; 
     this.url = url; 
    }  

    public void run() { 
     S3Downloader downloader = new S3Downloader(); 
     S3Object result = downloader.download(url); 
     list.add(result); 
     latch.countDown(); 
    } 
} 

Пример Runner иллюстрирует клиент". Метод go является драйвером, и использует метод add (который не блокирует):

public class Runner { 
    private ExecutorService pool = Executors.newCachedThreadPool(); 
    private int numUrls = 20; 
    private CountDownLatch latch = new CountDownLatch(numUrls); 
    private List<S3Object> results = Collections.synchronizedList(new ArrayList<S3Object>()); 

    public void add(String url) { 
     pool.submit(new MyTask(results, latch, url)); 
    } 

    public void go() throws Exception { 

     for(int i = 0; i < numUrls; i++) { 
      String url = "http://example" + i; 
      add(url); 
     } 

     // wait for all downloads 
     latch.await(); 

     for (S3Object result : results) { 
      System.out.println("result id: " + result.id); 
     } 
    } 
} 

Производственный код должен обрабатывать ошибки и, возможно, реорганизовать клиента в зависимости от обстоятельств.

Answer 2

Вместо того, чтобы «наблюдатель», который потребляет ресурсы, сделать текущий загрузки уведомить хозяина о завершении.