Konsumpcyjne reszta usługi w sposób funkcjonalny

Question

Więc byłem ciekawy, w jaki sposób poniższy kod zostałby napisany bardziej efektywnie za pomocą API strumieni danych java8.

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; 

} 

Po metoda ta odbywa się robi, co jej robi i ja otrzymać listę obiektów FLightInfo, jestem przekształcenie go w strumieniu i będzie robić różne transformacje na nim (przez grupy, etc). Teraz jest całkiem pewna, że ​​jest to długotrwała operacja. Co więcej, w rzeczywistości łączy wiele wywołań reszty z usługą sieciową, więc mam już większość danych uzyskanych do czasu wykonania ostatniego połączenia, ale nie zacznę go przetwarzać przed zwróceniem całej metody.

Czy istnieje sposób, aby zrobić to trochę bardziej reaktywnie? Czy mogę natychmiastowo odesłać strumień i operację na tym strumieniu przetwarzać dane, gdy przychodzi do potoku, czy jest to zbyt dużo do zapytania? Jak to się stanie w Javie 8. Że

Answer 1

Wszystko zależy od tego, kiedy potrzebujesz wyniku. Jeśli chcesz, aby był sekwencyjny, to poniżej jest jeszcze przyzwoity sposób, jak leniwy. Ale gotuje się w terminalu (powiedzmy podczas 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; 
    })  
} 

Albo zrobiłbym to z destinations.stream().parallel(), jeśli mogę. Jest to rozsądny wynik w większości przypadków. Ale nadal nie zacznie przetwarzać go równolegle do czasu wywołania operacji terminalowej. Co absolutnie ma sens.

Ale wydaje mi się, że chcesz mieć coś typu producent-konsument. Dla których:

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()) 
      ); 
} 

dla prostota, w line-1 można po prostu wrócić collect zamiast sequence(collect). Następnie możesz powtórzyć listę, aby pobrać każdą wartość.

Jednak z jednym obiektem CompletableFuture należy się martwić, co oznacza, że ​​można sprawdzić wartości za jednym razem, jeśli zostały zakończone.