ЗавершенныйFuture, чтобы сделать webservice звонки и сохранить, когда все сделано

Question

У меня есть список сессий, которые я должен назвать веб-сервисом, чтобы установить какое-либо свойство в каждом сеансе.

Я пытаюсь вызвать webservice с помощью async-процесса и использовать для этого completeablefuture, так что, когда все будет готово, я могу сохранить их всех в db.

Как это сделать? Пока что мой код выглядит следующим образом, он не работает.

sessions.stream() 
     .forEach(s -> CompletableFuture.runAsync(() -> webServiceCall(s), executor)); 
sessionService.saveAll(sessions); 

EDIT:

Я пришел к этому решению, не уверен, что это правильный способ сделать это.

List<CompletableFuture<Void>> futures = sessions.stream() 
      .map(s -> CompletableFuture.runAsync(() -> webServiceCall(s), executor)) 
      .collect(Collectors.toList()); 
     CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) 
         .join(); 
     sessionService.saveAll(sessions); 

Я использую присоединиться, чтобы убедиться, что ждет ответа вернуться перед сохранением сессий

Answer 1

Короче - все, что вам нужно что-то вроде этого -

CompletableFuture.supplyAsync(this::supplySomething, ex).thenAccept(this::consumer); 

Вам нужен метод, который будет вызов в исполнитель (threadpool). В моем случае размер моего пула - 100. Затем вам нужно позвонить своему поставщику столько раз, сколько захотите.

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

Ваш поставщик должен вернуть какой-то объект, который содержит ответ от webservice. Затем этот объект станет параметром вашего «потребительского» метода.

Возможно, вы захотите убить бассейн после (или в середине), все будет сделано.

Обнаружили ниже пример -

package com.sanjeev.java8.thread; 

import java.io.BufferedReader; 
import java.io.DataOutputStream; 
import java.io.InputStreamReader; 
import java.io.Reader; 
import java.net.HttpURLConnection; 
import java.net.URL; 
import java.util.concurrent.CompletableFuture; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.TimeUnit; 

public class Caller { 

public static ExecutorService ex = Executors.newFixedThreadPool(100); 

public static void main(String[] args) throws InterruptedException { 
    Caller caller = new Caller(); 

    caller.start(); 

    ex.shutdown(); 
    ex.awaitTermination(10, TimeUnit.MINUTES); 
} 

private void start() { 
    for (int i = 0; i < 10000; i++) { 
     CompletableFuture.supplyAsync(this::supplySomething, ex).thenAccept(this::consumer); 
    } 
} 

private int supplySomething() { 
    try { 
     URL url = new URL("http://www.mywebservice.com"); 

     HttpURLConnection connection = (HttpURLConnection) url.openConnection(); 
     connection.setRequestMethod("POST"); 
     connection.setDoOutput(true); 
     connection.setDoInput(true); 

     connection.connect(); 

     try (DataOutputStream wr = new DataOutputStream(connection.getOutputStream())) { 
      wr.write("supply-some-data".getBytes()); 
     } 

     Reader in = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8")); 

     for (int c; (c = in.read()) >= 0;) { 
      System.out.print((char) c); 
     } 

     in.close(); 

     // return the response code. I'm return 'int', you should return some sort of object. 
     return 200; 

    } catch (Exception e) { 
     e.printStackTrace(); 
     throw new RuntimeException(e); 
    } 
} 

public void consumer(Integer i) { 
    // This parameter should be of type 'your object' that supplier returned. 
    // I got the response; add it in the list or whatever.... 
} 

}

Другой пример, который может подходит ваша потребность лучше -

public class Caller2 { 

public static ExecutorService ex = Executors.newFixedThreadPool(2); 
private static Iterator<String> addresses = Stream.of("www.google.com", "www.yahoo.com", "www.abc.com").collect(Collectors.toList()).iterator(); 
private static ArrayList<String> results = new ArrayList<>(); 

public static void main(String[] args) throws InterruptedException { 
    Caller2 caller = new Caller2(); 

    caller.start(); 

    ex.shutdown(); 
    ex.awaitTermination(1, TimeUnit.HOURS); 

    System.out.println(results); 
} 

private void start() { 
    while (addresses.hasNext()) { 
     CompletableFuture.supplyAsync(this::supplyURL, ex).thenAccept(this::consumer); 
    } 
} 

private String supplyURL() { 
    String url = addresses.next(); 
    // call this URL and return response; 
    return "Success"; 
} 

public void consumer(String result) { 
    results.add(result); 
}