[Java] Thread

스레드란 ?

컴퓨터 세계에서 동시에 여러가지 일을 병행해서 하는 멀티태스킹은 프로세스 기반과 스레드 기반 두 가지 유형으로 나뉜다.프로세스 기반의 멀티 태스킹은 미디어 프로그램 , 인터넷 브라우저, 메신저 프로그램 등을 동시에 실행하는 작업이고 프로세스 기반의 멀티태스킹에서는 병행처리 단위가 프로세스 이다.스레드 기반의 멀티 태스킹은 하나의 프로세스 내에서 여러 작업을 병행하는 것이고 워드 사용시 문서 편집과 동시에 프린트 작업을 할 수 있는데 하나의 프로그램 내에서 동시에 실행되는 작업 단위는 스레드라고 한다.

 

프로세스 기반

프로세스가 생성될 때마다 새로운 메모리 영역을 할당받고 프로세스 실행에 필요한 시스템 자원 또한 새롭게 할당 받음

 

스레드 기반

하나의 프로그램 내에서 병행되기 때문에 프로세스에서 사용하는 메모리와 자원을 공유하고 스레드를 실행하기 위한 자원만 할당

 

스레드의 목적

메인 스레드는 자바 프로그램 시작 시 자동으로 생성되며 public static void main() 메서드를 실행하고, 메서드의 실행이 완료되면 메인 스레드는 종료되고 자바 프로그램도 종료된다. 상황에 따라 멀티 스레드로 구현해야하는 경우는 다음과 같다.

 

1. 서비스하는 프로그램에서 동시에 여러 사용자가 실행 요청을 하면 싱글 스레드 환경에서는 한 사용자의 처리가 완료된 후 다른 사용자의 요청을 처리할 수 있다. 다음 사용자는 이전 사용자의 처리가 완료될 때 까지 기다려야하기 때문에 멀티 스레드를 사용해 여러 요청을 동시에 처리할 수 있다.

 

2. 외부 데이터 처리 네트워크 환경에서 데이터 전송이나 로컬 파일 시스템의 자원을 읽거나 쓰는 작업은 CPU 가 명령문을 처리하는 속도보다 느리다. 따라서 싱글 스레드 환경에서 외부 데이터를 대상으로 작업하면 CPU는 데이터 작업이 완료 될 때까지 유휴상태(idle time)가 된다. 유휴 상태가 길어지면 프로그램의 효율성이 떨어지는데 이럴 때 멀티 스레드로 외부 데이터를 처리하면 CPU의 유휴시간이 줄어든다.

 

메인 스레드만 실행되는 싱글 스레드 환경은 메인 스레드가 종료되면 자바 프로그램도 종료되지만 멀티 스레드 환경에서는 함께 실행되던 모든 스레드가 종료되어야 프로그램이 종료된다.

 

스레드 활용

run()

메인스레드가 아닌 독립적인 스레드에서 동작할 수 있도록 지원하는 java.lang 패키지의 Thread 클래스는 실행 시 새로운 스레드를 생성한 후 자신의 run() 메서드를 찾아 실행한다. 따라서 스레드에서 동작할 명령문들은 run() 메서드에 구현한다.

스레드에서 실행하는 명령문을 가지는 run() 메서드를 구현하는 방법은 두 가지이다.

 

1. Thread 상속

class OtherThread extends Thread{
	public void run(){
    	//  스레드에서 실행할 명령문
    }
}

2. Runnable 인터페이스 구현체

class OtherRun implements Runnable{
	public void run(){
    	// 스레드에서 실행할 명령문
    }
}

 

또는 람다식으로 Runnnable의 run() 메서드 구현

Runnable task = () ->{// 스레드 실행 명령문};
Thread p3 = new Thread(() -> {스레드 실행 명령문});

 

start()

스레드에서 처리할 로직을 구현한 run() 메서드는 Thread 클래스의 start() 메서드로 실행한다.

 

Thread 클래스 상속

run() 메서드를 구현한 경우는 바로 start() 메서드를 호출 가능

 

Runnable 인터페이스를 구현

Thread 객체를 생성해야 start() 메서드를 호출 가능

OtherThread p1 = new OtherThread();
p1.start();

OtherRun p2 = new OtherRun();
new Thread(p2).start();

 

위와 같이 start() 메서드를 호출해 p1, p2, p3 스레드를 실행하면 각 스레드는 실행 대기 상태로 변환

자바 프로그램은 JVM에 의해 실행되는 스레드가 결정되며 이것을 스케줄링이라 한다.

실행 대기 상태는 JVM의 스케줄링에 의해 실행할 수 있는 상태를 의미한다.

 

스레드 설정

이름 지정

Thread의 start() 메서드는 run() 메서드에 구현한 명령문을 실행하는 스레드를 실행 대기 상태로 만든다. 

실행 대기 상태에 있는 스레드는 기본적으로 메인 스레드는 main, 그 외의 스레드는 'Thread-번호' 형태의 이름이 지정된다.

스레드에 대한 디버깅, 제어를 위해 다른 이름으로 변경하고 싶다면 Thread의 setName() 메서드를 사용해 변경한다.

 

우선순위 지정

실행 대기 상태에 있는 스레드들은 JVM의 스케줄링에 의해 선택되어 실행된다. 그런데 스레드에 우선순위를 지정하여 실행을 제어할 수 있다. 우선순위는 정수로 표현하며 가장 높은 우선순위는 10, 가장 낮은 우선순위는 1, 기본값은 5이다.

우선순위가 높을 수록 다른 스레드보다 실행 시간을 더 많이 확보할 수 있다.

 

동기화

하나의 공유 자원을 여러 스레드가 동시에 접근하여 사용할 때 자원의 일관성을 위해 동시접근을 제어하는 것

 

동기화 처리

프로그램 구현 시 동기화는 블록이나 메서드 단위 선언 시에 synchronized 키워드 선언으로 동기화처리를 할 수 있다.

 

블록 동기화 

class OtherThread extends Thread{
	public void run(){
    	synchronized(공유 자원 객체){
        	// 스레드에서 실행할 명령문
        }
    }
}

객체 명에 공유 객체명을 지정하고, 만약 객체 자신이 공유된다면 this를 지정

 

메서드 동기화

class OtherThread extends Thread{
	public synchronized void hello(){
    	System.out.println("hello");
    }
	public void run(){
    	hello();// synchronized 키워드 메서드
    }
}

synchonized가 선언된 메서드는 한 스레드에 의해 호출되었다면 실행이 완료될 때 까지 다른 스레드는 실행할 수 없다.

 

스레드 제어

스레드 상태

여러개의 스레드가 동작하는 멀티태스킹은 동시에 여러작업이 진행된다고 하지만 엄밀히 따지면 CPU가 몇 개 인지에 따라 동시에 실행되는 스레드 수가 결정된다. 만약 CPU가 한개라면 특정 순간에 실행되는 스레드는 하나뿐이다. 왜냐하면 명령문을 처리하는 CPU가 하나뿐이기 때문이다. CPU가 하나뿐임에도 여러 스레드가 동시에 실행되는 것처럼 보이는 이유는 스레드를 번갈아가며 실행하기 때문이다.

 

JVM의 스케줄링에 의해 실행되는 스레드는 여러 상태를 거치면서 실행이 완료된다.

스레드의 각 상태는 Thread 클래스 내부에 enum 내부 클래스로 정의 되어 있다.

상태값	상태
NEW	스레드 객체는 생성되었지만 start() 메서드 호출 전
RUNNABLE	start() 메서드가 호출되어 실행할 수 있는 상태, 이 상태에서 JVM에 의해 선택되어 실행
BLOCKED	실행 대기 상태. JVM에 의해 RUNNABLE 상태로 변경
WAITING	실행 대기 상태. 다른 스레드에 의해 RUNNABLE 상태로 변경
TIME_WAITING	실행 대기 상태. 일정 시간이 지나면 RUNNABLE 상태로 변경
TERMINATED	스레드 실행 종료 상태

 

스레드 제어

wait(), notify(), notifyAll() 메서드

동기화는 자원의 성격에 따라 스레드를 좀 더 자세하게 제어 가능하다.

예를 들어 하나의 자원을 대상으로 소비와 생산 작업이 동시에 실행되는 경우이다.

자원을 소비하는 스레드는 생산된 자원이 있을 때만 실행할 수 있으므로 소비할 자원이 없을 때는 자원을 생산하는 스레드가 실행될 때까지 기다려야 한다. 그리고 자원이 생산되었으면 대기 중인 스레드에 자원이 생산되었음을 알려줘야 한다.

 

java.lang.Object의 wait()

자원을 소비하는 스레드는 자원이 없을 경우 자원이 생산될 때까지 대기해야 한다. 스레드가 wait() 메서드를 호출하면 해당 스레드는 RAUNNABLE 사앹에서 대기 상태인 WAITING 상태로 변경된다. WAITING 상태에서 RUNNABLE 상태로 변경되는 시점은 생산 스레드에서 notify() 또는 notifyAll() 메서드를 실행해줄 때이다.

 

java.lang.Object의 notify(), notifyAll()

자원을 생산하는 스레드에서 notify(), notifyAll() 메서드를 실행하면 WAITING 대기 상태에 있는 소비 스레드가 RUNNABLE 상태로 전환된다. notify()는 대기 상태인 한 개의 스레드만, notifyAll()은 모든 스레드를 RUNNABLE로 전환하는 차이점이 있다.

 

join() 메서드

여러 스레드가 동시 작업을 할 때 스레드간의 종속관계가 맺어지는 경우가 있다. 예를 들어 A라는 스레드 작업이 완료되어야 B라는 스레드 작업을 진행할 수 있는 경우이다. 이럴 때 B 스레드는 A 스레드 작업이 완료 될때까지 기다렸다가 실행되어야 하는데,  이럴 때 Thread의 join() 메서드를 사용한다.

public class Main{
	public static void main(Stirng[] args){
    	Phone calling = new Phone();
        calling.start();
        
        try{
        	calling.join(); // 전화가 끝난 후 걷기
        }catch(InterruptedException e){
        	e.printStackTrace();
        }
        
        for(int i=1;i<1000;i++){
        	Systme.out.println("걷기 : " + i);
        }
    }
}

class Phone extends Thread{
	public void run(){
    	for(int i=1;i<1000;i++){
        	Systme.out.println("전화 : " + i);
        }
    }
}

 

sleep() 메서드

지정된 시간동안 스레드를 TIME_WAITING 대기 상태로 전환하는 메서드

java.lang.Thread 클래스와 java.util.concurrent.TimeUnit 열거형 클래스에서 제공

 

interrupt() 메서드

slee(), wiat(), join() 메서드가 실행되어 실행 대기 상태에 있는 스레드들의 실행을 중지

TIME_WAITING 상태에서 interrupt() 메서드가 실행되면 InterruptedException이 발생

 

스레드풀

모든 스레드는 실행을 위해 다음의 과정을 거친다

 

1. Thread  또는 Runnable 상속하여 run() 메서드 구현

2. 스레드 객체 생성

3. start() 호출

4. 스레드 실행

5. 스레드 종료

 

스레드마다 Thread 객체 생성, 실행준비 상태에서 스케줄링에 의해 실행 또는 대기 상태로 전환되는 모든 과정이 모든 스레드에 적용되는 공통 사항이다. 즉, 스레드를 많이 생성할수록 Thread 객체는 많아지고, JVM이 스케줄링해야하는 스레드 또한 많아지는 구조이다. 따라서 스레드가 늘수록 메모리 사용량도 함께 늘며 스케줄링 작업 또한 복잡해지고 그만큼 개발자의 스레드 제어가 힘들어진다.

 

이런 단점을 보완하기 위한 것이 스레드 풀이다. 스레드풀은 처리로직과 스레드를 일대일로 매핑하는 거싱 아니라, 미리 스레드를 생성해놓고 이 스레드를 재사용하는 방식이다. 장점은 다음과 같다.

 

1. 스레드를 재사용함으로써 스레드의 생성, 삭제 비용을 절감할 수 있다.
2. 스레드를 제한된 개수로 사용하므로 스케줄링에 많은 오버헤드가 발생하지 않게 한다
3. 스레드 풀에서 스레드 제어를 지원하므로 간단한게 스레드 제어도 할 수 있다.

 

ExecutorService 인터페이스

스레드 풀 생성

 

스레드 풀을 사용하려면 ExecutorService 객체를 생성해야 한다.

ExecutorService는 java.util.concurrent.Executors에서 제공하는 메서드를 사용한다.

 

public static ExecutorService newFixedThreadPool(int nThreads)

위 메서드는 매개변수로 전달받은 개수의 스레드를 생성하여 관리하는 스레드 풀이다.

 

public static ExecutorService newCachedThreadPool()

위 메서드는 스레드 풀에 재사용할 수 있는 스레드가 있다면 재사용하고 없으면 새로운 스레드를 생성한다. 스레드 풀에서 60초 동안 사용되지 않는 스레드는 삭제한다.

 

작업 실행

 

스레드 풀 기반으로 스레드 작업을 할 때는 스레드 풀에 미리 준비된 Thread 객체를 사용하므로 별도의 Thread 객체를 생성할 필요는 없고 run() 메서드만 구현한다. 그리고 run() 메서드를 스레드 풀로 실행하려면 ExecutorService에 선언된 execute() 메서드를 사용한다. execute() 메서드의 인자로 run()을 구현한 Runnable 객체를 전달하면 스레드가 실행된다.

 

 

 

스레드 풀 종료

 

실행 대기 상태의 모든 스레드 작업이 완료되면 스레드 풀을 종료해야 한다. 스레드 풀을 종료하지 않으면 프로그램이 종료되지 않기 때문이다. 스레드풀을 종료하는 메서드는 ExecutorService에 선언된 shutdown() 또는 shutdownNow() 메서드를 사용한다.

 

ExecutorService 예제

class Task implements Runnable{
	@Override
    public void run(){
    	for(int i=1;i<1000;i++){
        	System.out.println("스레드 작업 : " + i);
        }
    }
}

public class Main{
	public static void main(String[] args){
    	ExecutorService thredPool = Executors.newFixedThreadPool(10); // 스레드풀 생성
        threadPool.execute(new Task); // 스레드 실행
        threadPool.shutdown(); // 스레드풀 종료
    }
}

 

Future 인터페이스

Callable

스레드로 동작할 처리 로직은 Runnable의 run() 메서드에 구현한다. 그런데 스레드 풀을 사용할 때 또 다른 구현 방법이 있다. java.util.concurrent에 정의된 Callable 함수형 인터페이스의 call() 메서드 이다. 

call() 과 run()의 차이점은 run은 반환값이 없지만, call은 반환값이 있다.

 

submit()

ExcutorService에서 run()을 실행하던 execute() 메서드 대신 call()을 실행할 때는 submit() 메서드를 사용한다.

 

Future<V>

Futuer는 java.util.concurrent에 정의된 인터페이스로서 ExcutorService의 submit() 메서드 실행 후 반환하는 객체이다. 이 객체는 Callable의 call() 메서드에서 반환하는 값을 가진다. Future 인터페이스에 선언된 메서드는 다음과 같다.

제어자 및 타입	메서드	설명
boolean	cancel(boolean flag)	매개변수가 true이면 이미 실행중인 작업도 작업취소, false이면 실행중인 작업은 취소하지 않음
V	get()	결과값 추출
V	get(loing timeout, TimeUnit unit)	지정된 시간 동안 대기 후 결과값 추출
boolean	isCacelled()	작업 취소 여부 판단
boolean	isDone()	작업 취소 여부 판단

 

Runnable  > Collable, Future 예제

public class Main{
	public static void main(String[] args){
    	ExecutorService threadPool = Executors.newCachedThreadPool(); // 스레드 풀 생성
        Future<Date> future = threadPool.submit(new Callable<Date>(){ // submit() 인자로 전달한 Runnable 또는 Callable 구현 객체를 스레드로 실행, Callable의 call() 메서드의 반환값을 Future 객체로 반환
        	@Override
            public Date call() throws Exception{
            	Thread.sleep(1000);
                return new Date();
            }
        });
        
        Date date = null;
        try{
        	date = future.get(); // Future 타입의 future는 스레드가 실행된 후 반환된 값을 가지고 있다. get() 메서드는 submit() 메서드로 실행된 스레드의 처리 로직이 완료될 때 까지 기다린 후 결과값을 추출
            System.out.println(date);
        }catch(Exception e){
        	e.printStackTrace();
        }
    }
}

 

CompletableFuture

자바 5에 Future가 추가되면서 비동기 작업에 대한 결과값을 반환 받을 수 있지만 외부에서 작업을 완료시킬 수 없고 블로킹 호출인 get을 이용하고, 예외처리가 어렵고, 여러 Future들을 조합할 수도 없었기에 이러한 문제를 해결하기 위해 자바 8에 CompletableFuture가 추가됐다. 

 

Future를 기반으로 외부에서 완료시킬 수 있고, 완료후 콜백등이 추가되어 작업들을 중첩시킬 수 있게한다.

 

비동기 작업실행

기본적으로 자바7에 추가된 ForkJoinPool의 commonPool을 사용해 스레드를 동작한다.

원하는 쓰레드 풀을 사용하려면 쓰레드 풀을 파라미터로 넘겨주면 된다.

public class CompletableFutureTest {

  private final Logger log = LoggerFactory.getLogger(getClass());
  
  private void sleep(long millis) {
    try {
      Thread.sleep(millis);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  public int 메소드_완료_1초_필요() {
    sleep(1000L);
    return 1;
  }

  public int 메소드_완료_2초_필요() {
    sleep(2000L);
    return 2;
  }
  
  @Test
  void 동기처리() {
    long start = System.currentTimeMillis();
    int a = 메소드_완료_1초_필요();
    int b = 메소드_완료_2초_필요();
    long elapsed = System.currentTimeMillis() - start;
    log.info("[전체시간={}s] 결과는 a + b = {}", MILLISECONDS.toSeconds(elapsed), (a + b));
  }
  
  @Test
  void 병렬처리() {
    long start = System.currentTimeMillis();

    CompletableFuture<Integer> fut1 = supplyAsync(this::메소드_완료_1초_필요);
    CompletableFuture<Integer> fut2 = supplyAsync(this::메소드_완료_2초_필요);

    int b = fut2.join();
    log.info("fut2 걸린시간 {}s", MILLISECONDS.toSeconds((System.currentTimeMillis() - start)));

    long start2 = System.currentTimeMillis();
    int a = fut1.join();
    log.info("fut1 걸린시간 {}s", MILLISECONDS.toSeconds((System.currentTimeMillis() - start2)));

    long elapsed = System.currentTimeMillis() - start;
    log.info("[전체시간={}s] 결과는 a + b = {}", MILLISECONDS.toSeconds(elapsed), (a + b));
  }
  
  }

static CompletableFuture<Void> runAync(Runnable runnable)

반환값이 없는 경우, 비동기로 작업 실행 콜, Runnable의 역할 수행

 

static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)

반환값이 있는 경우, 비동기로 작업 실행 콜, Callable의 역할 수행

 

비동기 작업 완료 콜백

CompletableFuture<Void> thenAccept(Consumer<? super T> action)

비동기 작업이 완료됏을 때 완료 결과를 소비(Consume) 처리함. 작업의 끝

 

<U> CompletableFuture<U> thenApply(Function<? super T, ? extends U> fn)

비동기 작업이 완료됐을 때, 결과 T를 새로운 U로 변환하는 함수를 실행

결과값은 다시 CompletableFuture

 

CompletableFuture<Void> thenRun(Runnable action)

비동기 작업이 완료 됐을 때 반환 값을 받지 않고 다른 작업을 실행

 

CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)

비동기 작업에서 예외가 발생했을 때, 예외를 Throwable로 받고, 결과값 T를 생성

 

작업 조합

allOf

여러 작업들을 동시에 실행하고, 모든 작업 결과에 콜백을 실행

 

anyOf

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

 

thenCompose

두 작업이 이어서 실행하도록 조합, 앞선 작업의 결과를 받아서 사용가능

 

thenCombine

두 작업을 독립적으로 실행하고, 둘 다 완료되었을 때 콜백을 실행

 

사용 예시

멀티 스레딩을 사용해야하는 네트워크와 연결같은 비교적 처리속도가 느린 컨트롤러 위의 코드 예시이다.

Controller code
@PostMapping(path = "user/join", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
  public ApiResult<JoinResult> join(
    @ModelAttribute JoinRequest joinRequest,
    @RequestPart(required = false) MultipartFile file
  ) {
    User user = userService.join(// ..); 회원 가입 메소드
    
    toAttachedFile(file).ifPresent(attachedFile -> 
      supplyAsync(()-> // 컨트롤러로 받은 이미지를 S3에 업로드 하는 부분은 다른 쓰레드 이용, suppliAsync에 묶인 메소드는 요청, 응답하는 컨트롤러와 별개로 작동
              uploadProfileImage(attachedFile))  
              .thenAccept(opt->
                      opt.ifPresent(profileImageUrl ->
                             // userService profileImageUrl 업데이트
                      )
              )
    ).exceptionally(throwable -> {}); // 다른 쓰레드에서 작동하기에 ExceptionHandler에 걸리지 않아 예외처리 필요

    return OK(
      new JoinResult(apiToken, new UserDto(user))
    );
  }

 

참고

처음해보는 자바 프로그래밍 - JVM 메모리 구조로 이해하는 객체지향(오정임)

https://mangkyu.tistory.com/263

[Java] CompletableFuture에 대한 이해 및 사용법