[아이템 80] 스레드보다는 실행자, 태스크, 스트림을 애용하라

[kihyun-yang]

서론

• 동시성 작업을 할 때는 작업 큐를 직접 생성할 수도 있겠지만 복잡한 작업들(안전실패, 응답불가 예방)이 필요하다.
• java.util.concurrent 패키지는 실행자 프레임워크 (Executor Framework)라고 하는 인터페이스 기반의 유연한 태스크 실행 기능을 담고 있다.
• java.util.concurrent 패키지의 Executors가 제공하는 정적 팩터리 메서드를 사용하면 다양한 작업 큐를 얻을 수 있다.
  • 그러니까 java.util.concurrent 패키지의 실행자, 태스크, 스트림을 이용하는 편이 낫다.
  • 직접 작업 큐를 구현하려면 복잡한 작업들 (안전실패, 응답불가 예방, 우하한 종료 등...)을 처리하는데 큰 노력이 필요하다.

실행자 서비스 (스레드 풀)의 주요 기능들

특정 태스크가 완료되기를 기다린다.

@DisplayName("하나의 worker를 가지는 쓰레드풀")
@Test
 void single() throws ExecutionException, InterruptedException {
 ExecutorService exec = Executors.newSingleThreadExecutor();

 exec.submit(() -> {
 System.out.println("First");
 try {
 Thread.sleep(1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }).get();
 exec.execute(() -> System.out.println("Second"));
}

• submit 메서드를 사용하면 Future 인터페이스를 반환받을 수 있고 get() 메서드를 통해 해당 Runnable (Callable도 가능) 를 실행할 때까지 기다리는 작업이 가능하다.
• 따라서, 첫번째 Runnable 의 실행이 끝난 후 두번째 Runnable 도 실행이되고, 아래와 같은 출력을한다.

  First
  Second
  

태스크 모음 중 아무것 하나 혹은 모든 태스크가 완료되기를 기다린다.

@DisplayName("태스크 모음 중 아무것 하나 혹은 모든 태스크가 완료되기를 기다린다.")
@Test
void any() throws InterruptedException, ExecutionException {
 ExecutorService exec = Executors.newFixedThreadPool(3);

 List<Future<String>> returnStr = exec.invokeAll(tasks());
 System.out.println(returnStr.get(0).get()); // FIRST
 System.out.println(returnStr.get(1).get()); // SECOND
 System.out.println(returnStr.get(2).get()); // THRID
}

List<Callable<String>> tasks() {
 return Arrays.asList(() -> {
 try {
 	Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "FIRST";
 },
 () -> {
 try {
 Thread.sleep(1000);
 } catch (InterruptedException e) {
 	e.printStackTrace();
 }
 	return "SECOND";
 },
 () -> "THIRD");
}

@DisplayName("태스크 모음 중 아무것 하나 혹은 모든 태스크가 완료되기를 기다린다.")
@Test
void any() throws InterruptedException, ExecutionException {
 ExecutorService exec = Executors.newFixedThreadPool(3);

 System.out.println(exec.invokeAny(tasks())); // THRID 출력 후 
}

List<Callable<String>> tasks() {
 return Arrays.asList(() -> {
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace(); // Interrupt
 }
 return "FIRST";
 },
 () -> {
 try {
 Thread.sleep(1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "SECOND";
 },
 () -> "THIRD");
}

실행자 서비스가 종료하기를 기다린다.

@DisplayName("실행자 서비스가 종료하기를 기다린다")
@Test
void waitForTerminate() throws InterruptedException {
 ExecutorService exec = Executors.newSingleThreadExecutor();

 exec.execute(() -> {
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("TEST"); // 출력안된다
 });

 exec.awaitTermination(2000, TimeUnit.MILLISECONDS);
}

@DisplayName("실행자 서비스가 종료하기를 기다린다")
@Test
void waitForTerminate() throws InterruptedException {
 ExecutorService exec = Executors.newSingleThreadExecutor();

 exec.execute(() -> {
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("TEST"); // 출력된다
 });

 exec.awaitTermination(4000, TimeUnit.MILLISECONDS);
}

완료된 태스크들의 결과를 차례로 받는다.

@DisplayName("완료된 태스크들의 결과를 차례로 받는다")
@Test
void sequence() throws ExecutionException, InterruptedException {
 CompletionService<String> cs = new ExecutorCompletionService<>(Executors.newFixedThreadPool(3));
 List<Callable<String>> tasks = tasks();
 tasks.forEach(cs::submit);
 for (int i = tasks.size(); i > 0; i--) {
 String r = cs.take().get();
 if (r != null)
 System.out.println(r); // THRID, SECOND, FISRT 순서로 출력
 }
}
 
List<Callable<String>> tasks() {
 return Arrays.asList(() -> {
 try {
 Thread.sleep(3000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "FIRST";
 },
 () -> {
 try {
 Thread.sleep(1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return "SECOND";
 },
 () -> "THIRD");
}

태스크를 특정 시간에 혹은 주기적으로 실행하게 한다.

@DisplayName("테스크를 특정 시간에 혹은 주기적으로 실행하게 한다.")
@Test
void schedule() throws InterruptedException {
 System.out.println("싱글쓰레드 쓰케쥴executors");
 ScheduledExecutorService exc = Executors.newSingleThreadScheduledExecutor();

 //3초 대기 후 출력 (Runnable command, long delay, TimeUnit unit) <- 매개변수
 exc.schedule(() -> System.out.println("Hello"), 3000, TimeUnit.MILLISECONDS);

 //특정 시간마다 반복 - 3초기다렸다가 3초마다 반복 (command, initial delay, period, TimeUnit unit)
 //시작딜레이(initialDelay) 이후 첫번째 실행을 시작으로 지정한 시간(period)만큼 차이로 정확하게 반복 실행.
 exc.scheduleAtFixedRate(() -> System.out.println("Hello!"), 3000, 3000, TimeUnit.MILLISECONDS); /// 3번출력
 Thread.sleep(10000);
}

스레드풀의 설정을 조금 더 견고하게 하고 싶다면 ThreadPoolExecutor 클래스를 직접 사용할 수 있다.

실행자 서비스 (스레드 풀) 생성 팁

• Executors.newCachedThreadPool는 태스크를 요청받으면 바로 가용가능한 스레드에 할당하고 없으면 그 즉시 만든다. 태스크가 자주 쌓이는 시스템이라면 적합하지 않다.
  • CPU 이용률이 100%로 치닫을 우려.
• 따라서 무거운 프로덕션 서버에서는 스레드 개수를 고정한 Executors.newFixedThreadPool을 선택하거나 완전히 통제할 수 있는 ThreadPoolExecutor를 직접 사용하는 편이 훨씬 낫다.
• 작업 큐를 손수 만드는 일은 삼가야 하고, 스레드를 직접 다루는 것도 일반적으로 삼가야 한다.

포크-조인 태스크

• 자바 7이 되면서 실행자 프레임워크는 포크-조인 태스크를 지원하도록 확장됐다.
• 포크-조인 태스크는 포크-조인 풀이라는 특별한 실행자 서비스가 실행해준다.
• 포크조인 태스크에서는 먼저 일을 끝낸 스레드가 다른 스레드의 남은 태스크를 수행하는 것이 가능해진다.
  • 이렇게 하여 모든 스레드가 바쁘게 움직여 CPU를 최대한 활용하면서 높은 처리량과 낮은 지연시간을 보장한다.
• 자바 8부터 나온 병렬 스트림은 내부적으로 포크조인 프레임워크를 사용한다.
  • 포크-조인 태스크를 직접 작성하고 튜닝하기란 어려운 일이지만, 병렬 스트림을 사용하면 적은 노력으로 그 이점을 얻을 수 있다.

참고

• https://javabom.tistory.com/84
• http://kimilb412-2.blogspot.com/2016/02/java-concurrency-threads-and-executors.html
• https://blog.naver.com/PostView.nhn?isHttpsRedirect=true&blogId=2feelus&logNo=220732310413
• https://hamait.tistory.com/612