CompletableFuture 2

조합하기

thenCompose()

• 두 작업이 서로 이어서 실행하도록 조합
  
• 두 작업이 연관관계가 있을 때

public class App {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {
			System.out.println("Hello " + Thread.currentThread().getName());
			return "Hello";
		});
		
		CompletableFuture<String> future = hello.thenCompose(s -> getWorld("안녕하세요"));
		
		System.out.println(future.get());
	}
	
	private static CompletableFuture<String> getWorld(String message){
		return CompletableFuture.supplyAsync(() -> {
			System.out.println(message + " " + Thread.currentThread().getName());
			return message + " World";
		}); 
	}
}

 

 

thenCombine()

• 두 작업을 독립적으로 실행하고 둘 다 종료 했을 때 콜백 실행

public class App {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {
			System.out.println("Hello " + Thread.currentThread().getName());
			return "Hello";
		});
		
		CompletableFuture<String> world = CompletableFuture.supplyAsync(() -> {
			System.out.println("World " + Thread.currentThread().getName());
			return "World";
		});
		
		CompletableFuture<String> future = hello.thenCombine(world, (h, w) -> h + " " + w);
		System.out.println(future.get());
		
	}
}

 

 

allOf()

• 여러 작업을 모두 실행하고 모든 작업 결과에 콜백 실행
• 각각의 결과가 오류가 발생 및 리턴타입이 다를 수 있으므로 List로 받아서 해야한다.

return type이 void 인경우 null 발생(리턴을 받을 수 없으므로)

public class App {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {
			System.out.println("Hello " + Thread.currentThread().getName());
			return "Hello";
		});
		
		CompletableFuture<Integer> world = CompletableFuture.supplyAsync(() -> {
			System.out.println("World " + Thread.currentThread().getName());
			return 1000;
		});
		
		CompletableFuture<Void> future = CompletableFuture.allOf(hello, world)
				.thenAccept(System.out::println);
		
		System.out.println(future.get());
	}
}

 

위의 방법을 해결하기 위해서는 각각의 결과값을 저장할 수 있게 아래와 같이 List에 담아줘야 한다.

public class App {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {
			System.out.println("Hello " + Thread.currentThread().getName());
			return "Hello";
		});
		
		CompletableFuture<String> world = CompletableFuture.supplyAsync(() -> {
			System.out.println("World " + Thread.currentThread().getName());
			return "World";
		});
		
		List<CompletableFuture<String>> futures = Arrays.asList(hello, world);
		CompletableFuture[] futuresArray = futures.toArray(new CompletableFuture[futures.size()]);
		
		CompletableFuture<List<Object>> results = CompletableFuture.allOf(futuresArray)
				.thenApply(v -> {
					return futures.stream()
							.map(f -> f.join())
							.collect(Collectors.toList());});
	
		results.get().forEach(System.out::println);		
	}
}

 

anyOf()

• 여러 작업 중에 가장 빨리 끝난 하나의 결과에 콜백 실행

public class App {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {
			System.out.println("Hello " + Thread.currentThread().getName());
			return "Hello";
		});
		
		CompletableFuture<String> world = CompletableFuture.supplyAsync(() -> {
			System.out.println("World " + Thread.currentThread().getName());
			return "World";
		});
		
		CompletableFuture.anyOf(hello, world).thenAccept((s) -> {
			System.out.println(s);
		});
	}
}

 

 

예외처리

exeptionally(Function)

public class App {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		boolean throwError = true;

		CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {

			if (throwError) {
				throw new IllegalArgumentException();
			}
			System.out.println("Hello " + Thread.currentThread().getName());
			return "Hello";
		}).exceptionally(ex -> {
			System.out.println(ex);
			return "Error!";
		});
		System.out.println(hello.get());
	}
}

 

handle(BiFunction): 

public class App {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		boolean throwError = true;

		CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {

			if (throwError) {
				throw new IllegalArgumentException();
			}
			System.out.println("Hello " + Thread.currentThread().getName());
			return "Hello";
		}).handle((result, ex) ->{
			if(ex != null) {
				System.out.println(ex);
				System.out.println("Error hadle");
			}
			return result;
		});
				
		System.out.println(hello.get());
	}
}

 

 

참고

• https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ForkJoinPool.html
• https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html