Потребление службы отдыха функциональным способом

Question

Так что мне было любопытно, как следующий код будет переписан более эффективно с API-интерфейсом java8 streams.

public static List<FlightInfo> getResults(String origin,List<String> destinations) { 

    final String uri = "https://api.searchflight.com/; 
    List<FlightInfo> results = new LinkedList<FlightInfo>(); 

    for(String destination:destinations) { 


      RestTemplate restTemplate = new RestTemplate(); 

      String params = getParams(origin,destination); 
      FlightInfo result = restTemplate.postForObject(uri+params,FlightInfo.class); 

      results.add(result); 
    } 

    return results; 

} 

После этого метода делается делать то, что делает его, и я получаю список объектов FLightInfo, я преобразуя его в поток и будет делать различные преобразования на ней (группе, и т.д.). Теперь довольно неплохо, что это длительная работа. Кроме того, он фактически объединяет несколько вызовов отдыха в веб-службу, поэтому у меня уже есть большинство данных, полученных к моменту моего последнего вызова, но я не стал бы его обрабатывать до того, как весь метод вернется.

Есть ли способ сделать все это более реактивно? Могу ли я сразу же возвратить поток и выполнить операции над данными процесса потока, поскольку он идет по трубе, или это слишком много, чтобы спросить? Как это будет сделано на Java 8. Это

Answer 1

Ну, все зависит от того, когда вам нужен результат. Если вы хотите, чтобы это было последовательным, это ниже по-прежнему достойно, так как его ленивый. Но он будет кипеть при терминальной операции (скажем, во время collect).

public static Stream<FlightInfo> getResults(String origin,List<String> destinations) { 
    final String uri = "https://api.searchflight.com/"; 
    return destinations.stream().map(destination -> { 
     RestTemplate restTemplate = new RestTemplate(); 
     String params = getParams(origin,destination); 
     FlightInfo result = restTemplate.postForObject(uri+params,FlightInfo.class); 
     return result; 
    })  
} 

Или я бы сделал это с destinations.stream().parallel(), если смогу. Это разумный результат в большинстве случаев. Но все же он не начнет обрабатывать его параллельно до того момента, когда вы вызовете для него операцию терминала. Что совершенно имеет смысл.

Но мне кажется, что вы желаете для производителя-потребителя типа вещи. Для которых:

public static CompletableFuture<List<FlightInfo>> getResults(String origin,List<String> destinations) { 
    final String uri = "https://api.searchflight.com/"; 
    List<CompletableFuture<FlightInfo>> collect = destinations 
      .stream() 
      .map(destination -> CompletableFuture.supplyAsync(() -> { 
       RestTemplate restTemplate = new RestTemplate(); 
       String params = getParams(origin,destination); 
       FlightInfo result = restTemplate.postForObject(uri+params,FlightInfo.class); 
       return result;    
      })).collect(Collectors.toList()); 
    return sequence(collect);  //line-1 
} 

public static <T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com) { 
    return CompletableFuture.allOf(com.toArray(new CompletableFuture[com.size()])) 
      .thenApply(v -> com.stream() 
          .map(CompletableFuture::join) 
          .collect(Collectors.toList()) 
      ); 
} 

Для простота, в line-1 вы можете просто вернуть collect вместо sequence(collect). Затем вы можете перебирать список для получения каждого значения.

Но с sequence у вас есть один объект CompletableFuture, о котором вы можете беспокоиться, и затем вы можете проверить значения сразу, если это будет завершено.