JVM 동시성 프로그래밍 딥다이브 3. 스레드 기본 API - Sleep, Join, Interrupt 등

JVM의 Thread 객체가 갖고 있는 기본적인 API들이 있다.

우선 Sleep, Join, Interrupt에 대해 알아보자.

다양한 메소드들을 제공하고 있고 그에 따른 동작과 상태 변화들이 있다.

Sleep()

• sleep 메소드는 지정된 시간 동안 현재 스레드의 실행을 일시 정지하고 대기 상태(Blocked)로 빠졌다가
  시간이 지나면 실행대기 상태(Runnable)로 전환된다.
• native 메소드로 연결되어 시스템 콜이 발생하며 커널모드에서 수행 후 유저모드로 전환된다. 즉, 오버헤드가 존재한다.

API

- sleep(m: 밀리초)

- sleep(m: 밀리초, n: 나노초)

- InterruptedException

• 스레드가 수면중에 인터럽트되면 InterruptedException이 발생한다.
• 다른 스레드는 잠자고 있는 스레드에 인터럽트(중단, 멈춤) 신호를 보낼 수 있다.
• 자고있다가 InterruptedException이 발생하면서 스레드는 깨어나고 실행대기 상태인 Runnable로 전환되어 실행상태인 Running 상태를 기다리게 된다.

작동 방식 정리

• 주어진 sleep 시간이 끝나서 깨어나더라도 실행대기 상태로 전환되는것이지 바로 이전 작업을 수행하는건 아니다.
  (스케줄러의 선택 필요)
• 동기화 메서드 영역에서 수면중인 스레드는 획득한 모니터나 락을 잃지않고 유지한다. (synchronized 내부)
• sleep 중인 스레드에 인터럽트가 발생하면 대기상태에서 깨어나고 실행대기 상태로 전환되어서 인터럽트를 발생시킨다. 그러고 실행대기 상태로 실행상태를 기다림.

동작 예시 코드

fun main() {
    val sleepingSonThread = Thread {
        try {
            println("${Thread.currentThread().name} - 아들 Thread가 20초 동안 잠듭니다. ${getCurrentTime()}")
            Thread.sleep(20000L)
            println("${Thread.currentThread().name} - 아들 Thread가 기분 좋게 깨어났습니다. ${getCurrentTime()}")
        } catch (e: InterruptedException) {
            println("${Thread.currentThread().name} - 아들 Thread가 갑자기 인터럽트되며 깨어났습니다. ${getCurrentTime()}")
        }
    }

    val momThread = Thread {
        for (i in 1..3) {
            Thread.sleep(1000L) // 1초 대기
            println("... ${i} 초")
        }
        println("${Thread.currentThread().name} - 엄마: 아들아, 일어나라! ${getCurrentTime()}")
        sleepingSonThread.interrupt() // sleepingThread 인터럽트
    }

    sleepingSonThread.start()
    momThread.start()
}

스레드가 sleep을 하면 기본적으로 블록킹되어 대기상태로 들어갔다가 깨어나지만 다른 스레드가 깨우면 Interrupted 예외를 발생하며 깨어나게 된다.

---

Join()

• join() 메소드는 한 스레드가 다른 스레드가 종료될 때까지 실행을 중지하고 대기상태인 WAITING 상태로 들어갔다가, 다른 스레드가 종료되면 실행대기 상태로 다시 깨어나는 동작을 지원한다.
• 스레드의 순서를 제어하거나, 다른 스레드의 작업을 기다려야하거나, 순차적인 흐름을 구성하고자 할 때 사용할 수 있다.
• Object 클래스의 native 메소드 wait()과 연결되어 시스템 콜을 일으키고 커널 모드로 수행한다.
  커널 내부적으로 wait() & notify()로 제어한다.

주의
- ThreadTaskExecutor 로 스레드풀을 만들고 여러개의 비동기 작업을 수행한 후에 각가에 join() 을 걸어서 비동기 요청을 모두 가져오는 경우가 있다. 이건 비동기 작업의 결과값을 반환하는 CompletableFuture 메소드의 join()이다.

 

즉,

Thread.join()은 "이 스레드 끝날 때까지 기다려!" 의 합류의 의미이고
CompletableFuture.join()은 "이 작업 결과 줘!"의 획득의 의미이다.

 

조인 작동 방식 흐름 (1) - join() & wait() & notify()

 

조인 작동 방식 흐름 (2) - joint() & wait() & interrupt()

 

join() 작동 방식 정리

• join()을 실행하면 OS 스케줄러는 join()을 호출한 스레드를 대기상태(WATING)로 전환하고, 호출된 스레드에게 CPU를 사용하도록 한다.
• 호출 대상 스레드의 작업이 종료되면 join()을 호출한 스레드는 notify()를 받아 실행 대기 상태로 전환된다. 이후 스케줄러의 선택을 받아 실행.
• join()을 호출하는 스레드의 호출 대상 스레드가 여러개라면 각 스레드의 작업이 종료될 때마다 호출한 스레드는 WATING, RUNNABLE을 반복하게 된다.
• join()을 호출한 스레드가 interrupt 되면 해당 스레드는 대기에서 빠져나와서 대기 해제되고, 실행 가능 상태가 되어  InterruptedException을 발행한다.

동작 예시 코드 (1) - 메인스레드가 대상스레드에 join()을 호출하고 대기한다.

fun main() {
    val sonThread = Thread {
        try {
            println("${Thread.currentThread().name}: 아들스레드가 3초 동안 딴질을 합니다.")
            Thread.sleep(3000)
            println("${Thread.currentThread().name}: 아들스레드가 딴짓이 끝나서 다시 일하러 갑니다.")
        } catch (e: InterruptedException) {
            println("${Thread.currentThread().name}: 아들스레드가 딴짓하던 중에 방해를 받았습니다.")
        }
    }

    sonThread.start()

    println("${Thread.currentThread().name}: 메인스레드는 아들스레드가 딴짓이 끝날때까지 기다립니다.")
    /**
     * `join()` 메서드는 호출한 스레드가 대상 스레드가 종료될 때까지 기다리도록 합니다.
     * 따라서 현재 예시에서는 join을 호출한 메인 스레드가 `sonThread` 스레드가 딴짓이 끝날 때까지 기다립니다.
     */
    sonThread.join()
    println("${Thread.currentThread().name}: 기다리던 아들스레드가 딴짓을 마쳤습니다. 이제 메인스레드도 일을 계속합니다.")
}

 

동작 예시 코드 (2) - 메인스레드가 대상스레드에 join()을 호출하고 대기한다.

fun main() {
    println("${Thread.currentThread().name}: 메인스레드 시작")

    val thread1 = Thread {
        println("${Thread.currentThread().name}: Thread1 시작")
        try {
            Thread.sleep(2000L)
        } catch (e: InterruptedException) {
            println("Thread가 인터럽트되었습니다.")
        }
        println("${Thread.currentThread().name}: Thread1 종료")
    }

    val thread2 = Thread {
        println("${Thread.currentThread().name}: Thread2 시작")
        try {
            Thread.sleep(1000L)
        } catch (e: InterruptedException) {
            println("Thread2가 인터럽트되었습니다.")
        }
        println("${Thread.currentThread().name}: Thread2 종료")
    }

    thread1.start()
    thread2.start()

    println("${Thread.currentThread().name}: 메인스레드는 thread1, thread2 스레드가 작업을 완료할 때까지 기다립니다.")

    thread1.join()
    thread2.join()

    println("${Thread.currentThread().name}: 메인스레드 종료")
}

 

---

interrupt(), interrupted(), isInterrupted()

• Interrupt는 방해하다란 뜻으로 행동이나 실행흐름을 방해한다는 의미이다.
• 자바 스레드에서 interrupt()는 특정한 스레드에게 인터럽트 신호를 줘서 실행을 중단/ 작업을 취소/ 강제적 종료 등으로 사용할 수 있다.

interrupt()

• 스레드에게 인터럽트가 발생했음 신호를 보내는 매커니즘이다.
• 스레드가 현재 실행을 멈추고 인터럽트 이벤트를 먼저 처리하도록 시그널을 보내는 장치인 것.
• interrupted는 Thread가 들고있는 인터럽트 상태로 인터럽트 발생 여부를 확인할 수 있는 상태값이다.

// Thread 클래스 내부
volatile boolean interrupted; // default = false

• 한 스레드가 다른 스레드를 interrupt 할 수 도 있고, 자기 자신을 interrupt 할 수 도 있다.
• interrupt()에 횟수 제한은 없으며 인터럽트를 할 때마다 스레드의 인터럽트 상태는 true로 변경된다.

// Thread 클래스 내부

public void interrupt() {
    // 자기 자신이 아닌 경우에는 접근 가능 여부 확인. 즉 자기 자신 인터럽트도 가능.
    if (this != Thread.currentThread()) {
        checkAccess();

        // thread may be blocked in an I/O operation
        synchronized (interruptLock) {
            Interruptible b = nioBlocker;
            if (b != null) {
                interrupted = true;
                interrupt0();  // inform VM of interrupt
                b.interrupt(this);
                return;
            }
        }
    }
    interrupted = true;
    interrupt0();  // inform VM of interrupt
}

private native void interrupt0();

인터럽트 상태 확인

interrupted()

• static boolean interrupted() 함수를 통해 제공된다. 정적 메소드.

// Thread 클래스 내부

public static boolean interrupted() {
    return currentThread().getAndClearInterrupt();
}

boolean getAndClearInterrupt() {
    boolean oldValue = interrupted; // 만약 현재 interrupted가 true 라면

    if (oldValue) { // 과거에 인터럽트 된 적이 있는거고,
        interrupted = false; // 인터럽트 내역은 우선 제거한 다음.
        clearInterruptEvent(); // 인터럽트 이벤트도 제거한 다음.
    }
    
    return oldValue; // 과거에 인터럽트 된 적이 있음을 반환해준다.
}

• 인터럽트 상태가 true라면 인터럽트 상태를 false로 초기화하고, 인터럽 이벤트도 초기화한 다음에 인터럽트 상태가 true였음을 반환해준다.
• interrupt() -> interrupted 상태 true
• iinterrupted() -> 반환은 true / interrupted 상태 false
• interrupt() -> interrupted 상태 true ... 반복

isInterrupted()

• interrupted 필드를 그대로 반환해주는 인스턴스 메소드이다.
• interrupted의 상태를 초기화하지 않으므로 상태 확인만 하는 용도라면 해당 함수를 사용해야한다.

public boolean isInterrupted() {
    return interrupted;
}

 

InterruptedException

• 인터럽트 익셉션은 interrupt() 매커니즘의 일부이다. 대기나 차단 등 블로킹 상태에 있거나 블로킹 상태를 만나는 시점의 스레드에
  인터럽트를 하면 발생한다.
• 해당 예외가 발생하면 인터럽트 상태는 초기화 된다. (interrupted()를 한것과 동일)
• Thread.sleep, Thread.join, Object.wait(), Future.get(), BlockingQueue.take() 중에 인터럽트를 받으면 인터럽티드 익셉션이 발생한다.

Thread 코드에서는 native로 감춰져있어서 확인이 불가하지만 예외를 던지도록 되어있다.

// Thread 클래스 내부
private static native void sleep0(long nanos) throws InterruptedException;

 

아래는 virtual Thread 의 코드이긴하지만 참고해볼 수 있겠다.

// VirtualThread 클래스
void sleepNanos(long nanos) throws InterruptedException {
    assert Thread.currentThread() == this && nanos >= 0;
    
    // 위에서 봤던 getAndClearInterrupt를 통해 인터럽트 여부를 확인하고 초기화한 후 반환하는 그 메소드이다.
    // sleep 중에 인터럽트 받은게 확인되면 예외를 던진다.
    if (getAndClearInterrupt()) throw new InterruptedException();
     // 생략
}

 

인터럽트 상태 다이어그램

 

예시 코드

fun main() {
    val thread1 = Thread {
        try {
            for (i in 1..10) {
                printWithThread("$i - 인터럽티드? ${Thread.currentThread().isInterrupted}")
                Thread.sleep(500)
            }
        } catch (e: Exception) {
            if (e is InterruptedException) {
                printWithThread("인터럽트 예외 발생!")
                printWithThread("현재 인터럽트 상태: "+Thread.currentThread().isInterrupted)
            } else {

            }
        }
    }
    val thread2 = Thread({
        Thread.sleep(2000)
        printWithThread("thread1 인터럽트 시도")
        thread1.interrupt()
    })

    thread1.start()
    thread2.start()

    thread1.join()
}

• 디버깅을 해보면서 스레드의 인터럽트 필드 값의 상태를 확인해 볼 수 있다.
• 예외가 발생되어서 왔을때는 이미 getAndClearInterrupt()이 호출되었으므로 인터럽트 상태는 초기화 되어있다.

---

그 외 메소드들을 알아보자.

Thread Name

• 스레드의 이름을 확인할 수 있다.
• 기본적으로 스레드를 생성하면 스레드 이름은 자동으로 생성된다.
• 가장 먼저 생성되는 메인 스레드의 이름은 main 이다.
• setName 등으로 직접 설정도 가능하다.

currentThread

• Thread 클래스의 정적 메서드로서 현재 실행 중인 스레드 개체에 대한 참조를 반환한다.
• 예시: `Thread.currentThread().name`

isAlive

• Thread 내부 인스턴스 메소드로 스레드가 살아있는지 확인할 수 있다.
• 내부적으로 eetop 변수를 통해 확인한다. (eetop != 0)

private volatile long eetop;

Priority (스레드 우선순위)

• 단일 CPU에서 여러 스레드를 실행하는 것을 스케줄링이라고 하며, 스레드는 스케줄링에 의해 선점되어 CPU를 할당받는다.
• 자바 런타임은 고정 우선순위 선점형 스케줄링 (fixed-priority primitive scheduling)으로 알려진 매우 단순하고 결정적인 스케줄링 알고리즘을 지원한다.
• Java에서 스레드 우선순위는 1~10 사이의 정수값을 지원한다. 높을수록 우선순위가 높다!
• 다만, 스케줄러가 무조건적으로 우선순위가 높은 스레드를 실행한다고 보장 할 수는 없다.
  운영체제마다 다른 정책들이 있을 수 있으며 기아상태를 피하기 위해 스케줄러는 우선순위가 낮은 스레드를 선택할 수도 있다.
• setPriority(값)을 통해 설정할 수 있다.
  • 최소 1 / 기본 5 / 최대 10

// Thread 클래스 내부

/**
 * The minimum priority that a thread can have.
 */
public static final int MIN_PRIORITY = 1;

/**
 * The default priority that is assigned to a thread.
 */
public static final int NORM_PRIORITY = 5;

/**
 * The maximum priority that a thread can have.
 */
public static final int MAX_PRIORITY = 10;

 

코드 예시

fun main() {
    val one = 1L

    val priorityHigh = Thread({
        var count = 0L
        for (i in 1..10000000000) {
            if (Thread.currentThread().isInterrupted) {
                printWithThread("High Priority Thread가 인터럽트 상태입니다. count: $count")
                return@Thread
            }
            val toBigDecimal = one.toBigDecimal()
            count += toBigDecimal.toLong()
        }
        printWithThread("High Priority Thread 완료: $count")

    })

    val priorityMiddle = Thread({
        var count = 0L
        for (i in 1..10000000000) {
            if (Thread.currentThread().isInterrupted) {
                printWithThread("Middle Priority Thread가 인터럽트 상태입니다. count: $count")
                return@Thread
            }
            val toBigDecimal = one.toBigDecimal()
            count += toBigDecimal.toLong()
        }
        printWithThread("Middle Priority Thread 완료: $count")

    })

    val priorityLow = Thread({
        var count = 0L
        for (i in 1..10000000000) {
            if (Thread.currentThread().isInterrupted) {
                printWithThread("Low Priority Thread가 인터럽트 상태입니다. count: $count")
                return@Thread
            }
            val toBigDecimal = one.toBigDecimal()
            count += toBigDecimal.toLong()
        }
        printWithThread("Low Priority Thread 완료: $count")

    })

    priorityHigh.priority = Thread.MAX_PRIORITY      // 10
    priorityMiddle.priority = Thread.NORM_PRIORITY   // 5
    priorityLow.priority = Thread.MIN_PRIORITY       // 1

    priorityLow.start()
    priorityMiddle.start()
    priorityHigh.start()

    priorityHigh.join()
    priorityMiddle.interrupt()
    priorityLow.interrupt()
}

• PRIORITY를 MAX, NORMAL, MIN 각각 부여한 스레드에게10_000_000_000 (100억)을 카운팅하도록 했다.
• low, middle, high 순으로 start 후 high에 join()을 걸고 완료 후 middle, low 스레드에 인터럽트를 걸도록 했다.
• 모든 스레드에는 카운트할 때 인터럽트 여부를 확인하고 인터럽트 당하면 현재까지 센 카운트를 반환하고 리턴하도록 했다.
• 실제로 HIGH 스레드는 100억을 모두 세었지만 NORMAL, MIN 스레드는 99억8천만, 99억9천만 정도까지 센것을 확인할 수 있다.

 

TMI - voltile 키워드는 무엇일까?

일반적으로 멀티스레드 환경에서는 스레드는 최적의 성능을 내기 위해 변수의 값을 CPU 캐시에 저장해놓고 사용하게 된다.
하지만 voltile 키워드가 붙은 변수는 캐시를 바라보지 않고 무조건 실제 적재되어있는 메인메모리에 직접 접근해서 값을 읽고 쓰도록 한다. 이를 통해 스레드들이 항상 최신화된 값만 바라볼 수 있도록 한다.