ListenableFuture와 CompletableFuture를 실습 해 봅시다

ListenableFuture

AsyncRestTemplate 수행 결과로 리턴 됨.
callback을 통해 결과에 대한 success / error 상황 구현 가능
주의 : callback 지옥
(참고) Spring5에서는 deprecated 됨

asyncRestTemplateExample

	@Test
	public void asyncRestTemplateTest() throws InterruptedException, ExecutionException {
		ListenableFuture<ResponseEntity<String>> entity = asyncRestTemplate.getForEntity("http://google.com", String.class);

		entity.addCallback(new SuccessCallback<ResponseEntity<String>>() {
			@Override
			public void onSuccess(ResponseEntity<String> result) {
				System.out.println("success callback");     //1
			}
		}, new FailureCallback() {
			@Override
			public void onFailure(Throwable ex) {
				System.out.println("fail callback");         //2
			}
		});

		System.out.println("asyncRestTemplateTest");          //3
		
		ResponseEntity<String> responseEntity = entity.get();
		System.out.println(responseEntity); //content         //4
	}

실행 결과 ?

AsyncResult

2020-01-29 18:47:21.204  INFO 86265 --- [           main] ib.ListenableFutureTest                  : asyncRestTemplateTest
2020-01-29 18:47:21.459  INFO 86265 --- [ntLoopGroup-2-1] ib.ListenableFutureTest                  : success callback
2020-01-29 18:47:21.460  INFO 86265 --- [           main] ib.ListenableFutureTest                  : result <301,<HTML><HEAD><me...
2020-01-29 18:47:21.485  INFO 86265 --- [extShutdownHook] o.s.s.c.ThreadPoolTaskScheduler          : Shutting down ExecutorService 'taskScheduler'

2. CompletableFuture

1의 callback 지옥 대신, 선언형으로 callback 구현 가능. (코드 가독성이 좋아진다. )

callback hell에 대한 이미지 검색결과

allof, anyof 등으로 N개의 응답을 모두 기다렸다가 수행, N개 중 1개 이상 응답 올 때까지 기다렸다가 수행 가능.

AllOf

CompletableAllof

		private String print(int index) {
		try {
			Thread.sleep((5+index) * 1000L);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		String str = "Completed!!" + index;
		log.info(str);
		return str;
	}

	@Test
	public void allOfTest() throws Exception {
		log.info("allOfTest START");
		
		CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> print(0)); //sleep 5
		CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> print(1)); //sleep 6
		CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> print(2)); //sleep 7

		// callback 에서 머지하지 않는 경우
		CompletableFuture.allOf(cf1, cf2, cf3).thenAcceptAsync(s -> log.info("result 1: {}", s));            

		// wait until they are all done
		List<CompletableFuture<String>> completableFutures = Arrays.asList(cf1, cf2, cf3);

		CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[3]))
				.thenApplyAsync(
						result -> completableFutures.stream().map(future -> future.join()).collect(Collectors.toList()))
				.thenAcceptAsync(messages -> messages.forEach(message -> log.info("result 2: {}", message)));   //몇 초 뒤 찍힐까요?

		log.info("allOfTest END");  // 몇 초 뒤에 찍힐까요?

		Thread.sleep(11 * 1000L);
	}

실행 결과

all of result

19:01:07.539 [main] INFO ib.AllOfTest - allOfTest START
19:01:07.552 [main] INFO ib.AllOfTest - allOfTest END
19:01:12.552 [ForkJoinPool.commonPool-worker-1] INFO ib.AllOfTest - Completed!!0
19:01:13.553 [ForkJoinPool.commonPool-worker-2] INFO ib.AllOfTest - Completed!!1
19:01:14.552 [ForkJoinPool.commonPool-worker-3] INFO ib.AllOfTest - Completed!!2
19:01:14.553 [ForkJoinPool.commonPool-worker-1] INFO ib.AllOfTest - result 1: null
19:01:14.554 [ForkJoinPool.commonPool-worker-3] INFO ib.AllOfTest - result 2: Completed!!0
19:01:14.555 [ForkJoinPool.commonPool-worker-3] INFO ib.AllOfTest - result 2: Completed!!1
19:01:14.555 [ForkJoinPool.commonPool-worker-3] INFO ib.AllOfTest - result 2: Completed!!2

AnyOf

CompletableAnyof

		private String print(int index) {
		try {
			Thread.sleep((5+index) * 1000L);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		String str = "Completed!!" + index;
		log.info(str);
		return str;
	}

	@Test
	public void anyOfTest() throws InterruptedException {
		log.info("anyOfTest START");

		CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> print(0)); // sleep 5
		CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> print(1)); // sleep 6
		CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> print(2)); // sleep 7

		// Returns a new CompletionStage that, when this stage completesnormally
		CompletableFuture.anyOf(cf1, cf2, cf3).thenAcceptAsync(result -> log.info("result :{}", result)); //몇 초 뒤에 찍힐까요? 

		log.info("anyOfTest END");  // 몇 초 뒤에 찍힐까요?

		Thread.sleep(11 * 1000L);
	}

실행 결과

any of result

19:02:30.010 [main] INFO ib.AllOfTest - anyOfTest START
19:02:30.020 [main] INFO ib.AllOfTest - anyOfTest END
19:02:35.023 [ForkJoinPool.commonPool-worker-1] INFO ib.AllOfTest - Completed!!0
19:02:35.024 [ForkJoinPool.commonPool-worker-1] INFO ib.AllOfTest - result :Completed!!0
19:02:36.022 [ForkJoinPool.commonPool-worker-2] INFO ib.AllOfTest - Completed!!1
19:02:37.022 [ForkJoinPool.commonPool-worker-3] INFO ib.AllOfTest - Completed!!2