[Java 8] Concurrent Programming

더 자바, Java 8

이 포스팅에서 작성하는 내용은 더 자바, Java 8 에서 발췌하였습니다.

 

https://www.inflearn.com/course/the-java-java8

더 자바, Java 8 - 인프런 | 강의

 

---

Concurrent Programming

• 동시에 여러 작업을 할 수 있는 소프트웨어 프로그램
  ex) chrome, Intellij, youtube, word 등 여러 프로그램을 동시에 실행
• Java의 Concurrent Programming
  • Multi Processing
  • Multi Thread

 

Multi Thread

• Thread 상속
• Runnable 상속

 

Thread 주요 기능

• start() :  Thread 시작
• sleep(1000L) : 1초만큼 Thread 일시 정지
• join() : 다른 Thread 기다리기

 

public class Example10 {

    static class Thread1 extends Thread {
        @Override
        public void run() {
            System.out.println("Thread is Running : " + Thread.currentThread().getName());
        }
    }

    public static void main(String[] args) {
        /*** Thread를 사용하는 방법 ***/
        // 1. Thread 상속
        Thread1 thread1 = new Thread1();
        thread1.start();

        // 2. Runnable 을 상속한 익명 함수 (람다 표현식 가능)
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                try{
                    // sleep 하는 동안, 다른 Thread 는 작업을 진행.
                    Thread.sleep(2000L);
                }catch (InterruptedException ie) {
                    ie.printStackTrace();
                }
                System.out.println("Thread is Running : " + Thread.currentThread().getName());
            }
        });
        thread2.start();

        System.out.println("Main is Running");
    }
}

 

 

 

 

Executors

• Thread를 만들고 관리하는 작업을 어플리케이션에서 분리
• 이러한 기능을 처리하는 곳.
• Thread Pool 생성 / Thread 관리 / 작업 처리 등 진행
• ExecutorService
  • Executor를 상속받은 인터페이스
  • Callable도 실행할 수 있으며, Executor를 종료시키거나, 여러 Callable 을 동시에 실행 가능
• ScheduledExecutorService : ExecutorService를 상속받은 인터페이스로 특정시간 이후 또는 주기적으로 작업 실행 가능

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Example11 {
    public static void main(String[] args) {
        // ExecutorService 사용해보기
        ExecutorService executorService = Executors.newSingleThreadExecutor();

        // ExecutorService 는 작업을 실행시키면, 계속 대기하기 때문에 명시적으로 종료해줘야 함.
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println("Thread Name : " + Thread.currentThread().getName());
            }
        });

        executorService.shutdown();         // 현재 진행중인 작업을 완료한 후에 종료
    }
}

 

 

Callable

• Runnable과 비슷하지만, Callable은 리턴값이 있음
• Callable 작업을 진행하는 경우, 해당 작업이 완료될 때까지 프로그램은 기다림
• ExecutorService에서 invokeAll, invokeAny 사용 가능
  • invokeAll : 주어진 모든 Callable 작업 완료 후 반환
  • invokeAny : 주어진 Callable 작업 중 빨리 끝나는 작업 하나 반환

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

public class Example12 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newSingleThreadExecutor();

        Callable<String> callable1 = () -> {
            Thread.sleep(1000L);
            return "callable1";
        };

        System.out.println("START");
        Future<String> result = executorService.submit(callable1);

        String s = result.get();            // get() : ExecutorService 의 callable 작업이 끝날 때까지 프로그램 대기.
        System.out.println("END");
        System.out.println("s : " + s);


        Callable<String> callable2 = () -> {
            Thread.sleep(2000L);
            return "callable2";
        };

        Callable<String> callable3 = () -> {
            Thread.sleep(3000L);
            return "callable3";
        };

        // invokeAll : 해당 callable 작업이 모두 끝나야 결과값 반환.
        // invokeAny : 하나라도 빨리 끝나면, 그 하나의 결과값 반환
        List<Future<String>> resultList = executorService.invokeAll(Arrays.asList(callable1, callable2, callable3));

        for (Future<String> f : resultList) {
            System.out.println(f.get());
        }

        executorService.shutdown();
    }
}

 

 

 

CompletableFuture

• Future의 단점을 개선
  • get()을 하면, 결과를 받을 때까지  Blocking 되는 현상
  • 여러 Future를 조합할 수 없음
  • 예외 처리용 API를 제공하지 않음
• 비동기 작업 방법
  • runAsync() : 리턴값이 없을 때 사용
  • supplyAsync() : 리턴값이 있을 때 사용
• Callback 
  • thenApply(Function) : 리턴값을 받아서, 처리 후 반환
  • thenAccept(Consumer) : 리턴값을 처리(반환 X)
  • thenRun(Runnable) : 리턴값을 받고, 다른 작업 처리

import java.util.concurrent.*;

public class Example13 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // CompletableFuture 사용해보기 1
        CompletableFuture<String> future = new CompletableFuture<>();
        future.complete("Hello");

        System.out.println(future.get());

        // CompletableFuture.runAsync() : 리턴값이 없을 때 사용
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            System.out.println("Thread is Running : " + Thread.currentThread().getName());
        });
        future1.get();

        // CompletableFuture.supplyAsync() : 리턴값이 있을 때 사용
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            return "Thread is Running!! : " + Thread.currentThread().getName();
        });
        String s1 = future2.get();
        System.out.println(s1);

        // Callback 사용해보기 : thenApply(Function)
        // 1) thenApply(Function)
        // 2) thenAccept(Consumer)
        // 3) thenRun(Runnable)
        CompletableFuture<String> futureCallback1 = CompletableFuture.supplyAsync(() -> {
            return Thread.currentThread().getName();
        }).thenApply(s -> {
            return "Thread Name : " + s;
        });
        String s2 = futureCallback1.get();
        System.out.println(s2);
    }
}

 

 

CompletableFuture 조합

• thenCompose() : 두 작업을 실행
• thenCombine() : 두 작업을 각각 독립적으로 실행 후 콜백 실행
• allOf() : 여러 작업을 모두 실행 후 모든 작업 결과에 콜백 실행
• anyOf() : 여러 작업 중에 가장 빨리 끝난 하나의 결과에 콜백 실행

 

예외 처리

• exeptionally(Function) 
• handle(Bifunction)

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class Example14 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        CompletableFuture<String> nameFuture1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("nameFuture1 is Running : " + Thread.currentThread().getName());
            return "JAVA";
        });

        CompletableFuture<String> nameFuture2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("nameFuture2 is Running : " + Thread.currentThread().getName());
            return "C++";
        });

        // 1. thenCompose()
        CompletableFuture<String> future = nameFuture1.thenCompose((s) -> combineString(s));
        System.out.println(future.get());

        // 2. thenCombine()
        CompletableFuture<String> futureCombine = nameFuture1.thenCombine(nameFuture2, (result1, result2) -> result1 + ", " + result2);
        System.out.println(futureCombine.get());
    }

    static CompletableFuture<String> combineString(String s) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("combineString is Running : " + Thread.currentThread().getName());
            return s + " KING";
        });
    }
}