쓰레드

쓰레드

Thread는 프로세스 내의 작업의 단위이다.

쓰레드를 단일로 사용하면 싱글 쓰레드이고 여러개의 쓰레드를 사용하면 멀티 쓰레드라고 한다.

쓰레드의 구조

쓰레드는 프로세스의 Data, Code, Files를 다른 쓰레드들과 공유한다.

쓰레드는 독립적으로 Registers, Counter, Stack 영역이 있다. 쓰레드에 Stack 영역이 독립적으로 할당되는 이유는 독립적인 실행흐름, 즉 독립적인 함수 호출을 가능하게 하기 위함이다. 또, PC 레지스터를 독립적으로 가지는 이유는 독립적인 실행흐름이 있으므로 문맥 교환(Context Switching)이 발생하기 때문에 필요하다.

동시성(Concurrency) vs 병렬성(Parallelism)

자바의 쓰레드는 Concurreny하게 동작한다.

동시성은 논리적으로 여러 작업을 동시에 처리하는 멀티 쓰레드 동작이지만, 여러 쓰레드가 물리적으로 동시에 실행되지 않는다. 작업을 매우 빠르게 번갈아가면서 진행하여 동시에 진행되는 것처럼 보인다. 번갈아가는 과정을 문맥 교환(Context Switching)이라고 한다.

병렬성도 어떤 작업을 여러 작업으로 쪼개어 동시에 수행하는 것을 말한다. 병렬성은 동시성과 다르게 물리적으로 동시에 여러 쓰레드를 병렬적으로 수행하는 것을 말한다.

병렬성은 cpu의 멀티코어를 생각하면 된다.

자바에서 쓰레드 생성 및 사용

자바에서 쓰레드를 사용하는 방법은 여러가지가 있다.

첫번째는 Thread 클래스를 상속받아 run함수를 오버라이딩 해서 사용하는 방법이다.

생성

class TestThread extends Thread{
    @Override
    public void run() {
        
    }
}
public static void main(String[] args){
	Thread t = new TestThread();
}

실행은 Thread의 start()함수로 실행을 할 수 있다.

 

두 번째는 Runnable 인터페이스를 구현하는 방법이다.

class TestThread implements Runnable{
	public void run(){
     //...
	}
}
public static void main(String[] args){
	Thread t = new Thread(new TestThread());
}

 

세번째는 람다를 이용한 방법이다.

Thread t = new Thread(()->{
	// do something
});
t.start();

사용

사용은 만들어진 쓰레드에서 start()함수를 호출하면된다.

그 외 기능

쓰레드의 이름

var tr = new Thread(()->{}, "Thread_name");
var name = Thread.currentThread().getName();

쓰레드의 이름은 생성할 때 생성자에서 이름을 넣어줄 수 있다.

우선순위

쓰레드의 우선 순위는 1 ~ 10으로 설정할 수 있다. 주의할 점은 운영체제에서 정한 우선순위가 아닌 jvm에서 정한 우선순위이다. 1이 제일 낮은 우선순위이고 10이 제일 높은 우선순위이다, 보통은 5이다.

var curPriority = Thread.currentThread().getPriority();
Thread.currentThread().setPriority(1 ~ 10);

Demon 쓰레드

기본적으로 쓰레드를 생성하면 foreground 쓰레드이다. 이를 background 쓰레드로 변경할 수 있다.

Thread t = new Thread(()->{});
t.setDaemon(true); // true: 백그라운 / false: 포어그라운드

쓰레드 실행 제어 메서드

Sleep()

현재 쓰레드를 지정된 시간동안 멈추게 하는 메서드이다.

• 예외처리를 해야 한다.(InterruptedException이 발생하면 깨어난다)
• 특정 쓰레드를 지정해서 멈추게 하는 것은 불가능하다.

sleep()에 의해 일시정지 상태가 된 쓰레드는 지정된 시간이 다 지나거나 interrupt()가 호출되면 interruptedException이 발생되고, 잠에서 깨어나 실행대기 상태가 된다.

interrupt()

interrupt()는 쓰레드에게 작업을 멈추라고 요청한다. 단지 멈추라고 요청만 하는 것일 뿐 쓰레드를 강제로 종료시키지는 못한다.

isInterrupted()는 쓰레드에 대해 interrupt()가 호출되었는지 알려준다. interrupt()가 호출되지 않았다면 false, 호출되었다면 true를 반환한다.

만약, 쓰레드가 sleep(), wait(), join()에 의해 일시정지 상태에 있을 때 해당 쓰레드에 대해 interrupt를 호출하면 sleep(), wait(), join()에서 InterruptedException이 발생하고 쓰레드는 실행대기 상태로 바뀐다. 즉, 멈춰있던 쓰레드를 깨워서 실행 가능한 상태로 만든다.

suspend() & resume() & stop()

suspend()는 sleep()처럼 쓰레드를 멈추게 한다. suspend()에 의해 정지된 쓰레드는 resume()로 호출해야 다시 실행대기 상태가 된다.

stop()은 호출되는 즉시 쓰레드가 종료된다.

suspend와 stop은 교착상태를 일으키기 쉽게 작성되어 있으므로 사용 권장을 하지 않는다. 그래서 이 메서드들은 deprecated되었다.

yield()

yield()는 쓰레드가 스케쥴러에게 할당받은 시간 중 남은 시간을 포기하고 다른 쓰레드에게 양보한다.

join()

지정된 시간동안 특정 쓰레드가 작업하는 것을 기다린다.

void join() 				// 작업이 모두 끝날 때까지
void join(long ms) 			// 밀리 초 동안
void join(long ms, int ns) 	// 밀리 초 + 나노 초 동안

작업중에 다른 쓰레드의 작업이 먼저 수행돼야할 필요가 있을 때 join()을 사용한다. join()또한 sleep()처럼 interrupt()에 의해 대기상태에서 벗어날 수 있으며, join()이 호출되는 부분을 try-catch문으로 감싸야 한다.

쓰레드의 생명주기

동기화

동기화는 멀티 쓰레딩 환경에서 데이터의 일관성과 무결성을 유지하기 위한 필수 기술이다. 동기화를 통해 여러 쓰레드가 동일 자원에 접근하는 것을 제어할 수 있다. 동기화는 멀티 쓰레딩 프로그램의 안정성을 보장하는 중요한 역할을 한다. 동기화를 해주려면 간섭받지 않아야 하는 문장을 Critical Section(임계 영역)으로 설정해줘야 한다.

Lock

락이란, 특정 자원에 대한 접근을 제어하는 메커니즘이다.

공유자원에 다수의 쓰레드가 접근한다면 Data race가 발생하여 데이터가 손상될 수 있다.

 

 

데이터 레이스

경쟁 상태(Race Condition)의 대표적인 예이며, 멀티 쓰레드/프로세스 환경에서 일어나는 오류이다. 여러 쓰레드/프로세스가 공유자원에 동시에 접근하고 다른 쓰레드가 그 값을 수정하였을 때 일어난다.

 

예를 들어 5개의 쓰레드가 i라는 공유 변수를 사용하여 i++을 100000번씩 실행했다고 가정해 보자. 우리의 예상대로라면 500000이어야 정상일 것이다. 하지만 결과는 다르다. 여기서 데이터 레이스가 발생한 것이다. 우리가 사용한 i++은 언뜻 보기엔 한 번에 처리되는 것 같지만 어셈블리 코드로 봤을 때는 아래와 같이 세 단계로 나뉜다. 

어셈블리

i++연산의 순서는 다음과 같다.

1. prt[i]에서 eax로 값을 옮겨온다.

2. eax 에 add 명령어로 1을 더해준다

3. eax에서 ptr[i]로 값을 옮겨준다.

i는 0에서 시작한다.

쓰레드1이 2번 과정까지 끝났는데 갑자기 컨텍스트 스위칭이 일어나서 스레드2가 1번부터 3번까지 하고 다시 컨텍스트 스위칭으로 쓰레드 1이 작업한다고 생각해 보자. 쓰레드 1로 돌아왔을 때 쓰레드 1의 레지스터에는 i+1인 1이 저장되어 있을 것이다. 3번 과정을 마치면 i값은 최종적으로 1이다.

쓰레드1, 2가 각각 한 번씩 i++을 했는데 결과적으로 한 번만 처리된 것이다.

 

데이터 레이스를 해결하려면 연산을 원자적으로 처리하거나, 하나의 쓰레드만 접근해서 값을 변경할 수 있게 해야 한다.

자바에서의 Lock 사용법

자바에서 주로 사용하는 Lock 방법은 다음과 같다.

• synchronized 영역
• ReentrantLock 객체 사용
• ReentrantReadWriteLock 객체 사용
• StampedLock 객체 사용

Synchronized

Synchronized는 자바에서 동기화를 구현하는 가장 기본적인 방법이며, Monitor Lock이라고도 불린다.

Synchronized 키워드가 적용된 영역은 Lock을 획득한 쓰레드만 접근이 가능하다.  synchronized 블럭으로 동기화 시 자동적으로 lock이 잠기고 풀리며 블럭 내에서 예외가 발생해도 lock이 풀린다.

WAITING 상태인 스레드는 interrupt가 불가능하다.

사용법은 아래와 같다.

public synchronized Test(){
	// ...
}

public void Test(){
	synchronized(this){
    	// ...
    }
}

 

 

ReentrantLock

재진입이 가능한 Lock 객체이다. 명시적인 lock 방식으로 Synchrozied는 해당 블록이 끝나면 자동적으로 lock이 해제되지만, Lock객체의 경우 수동으로 unlock()을 해줘야한다. 보통 암묵적인 락만으로는 해결할 수 없는 복잡한 상황에서 사용될 수 있다.

synchronized와 다르게 lockInterruptably() 메서드를 통해 WAITING 상태의 쓰레드를 interrupt할 수 있다.

ReentrantReadWriteLock

ReentrantReadWriteLock은 동시에 읽으려는 쓰레드들은 허용하지만, 읽기 vs 쓰기, 쓰기 vs 쓰기는 허용하지 않는다.

쓰레드가 허용또는 차단되는지 확인하므로 성능 저하를 가져올 수 있다.

StampedLock

StampedLock은 세 가지 잠금모드(낙관적, 읽기, 쓰기)가 있는 재진입 락이다.

낙관적 모드는 기본 모드이다. 낙관적 모드에서 쓰레드는 차단하지 않고 잠글 수 있다. 다른 쓰레드에 의해 이미 잠긴 겨우 잠금 시도가 실패한다.

읽기 모드에서는 여러 쓰레드가 읽기를 위해 락 객체를 잠글 수 있지만, 쓰기를 위해 락 객체가 잠겨으면 락이 실패한다.

쓰기모드에서는 하나의 쓰레드만 쓰기를 위해 락 객체를 잠글 수 있다. 읽기 또는 쓰기를 위해 락 객체가 이미 잠겨있으면 잠금 시도가 실패한다.

Wait & Notify

wait()

wait() 메서드는 자바에서 동기화를 위한 Object의 인스턴스 메서드이다. wait메서드는 synchronized 블록 안에서만 호출이 가능하다. 

모니터락을 쥐고 있는 쓰레드가 wait()메서드를 호출하면 lock을 release하게 되고 다른 스레드가 해당 lock을 취득하게 된다. 

release했을 때의 쓰레드를 waiting pool에 넣는다. 해당 쓰레드는 대기 상태(WAITING | TIMED_WATING)로 바뀐다. TIMED_WAITING 상태의 쓰레드는 (wait(1000)과 같이 생성자에서 특정 시간을 설정해주었다면, 시간이 지난 뒤 자동으로 RUNNABLE 상태로 변경한다.

notify()

waiting pool에 있는 쓰레드에게 알림을 보내서 깨운다. 단, 어느 쓰레드가 일어나는지는 알 수 없다. 즉, 기아 문제가 발생할 수 있다.

notifyAll()

위 notify()는 어느 쓰레드가 일어나는지 알 수없기에 제어가 어렵다. 그래서 보통은 notifyAll()을 사용한다. waiting 된 코드가 synchronized 블록 안에 있는 쓰레드들을 모두 깨운다. 주의할 점은, waiting pool은 객체마다 있기에 notifyAll()이 호출된 객체의 waiting pool에 대기 중인 쓰레드들 깨운다.

Condition

컨디션은 Lock 인터페이스와 함께 사용되며 Lock 객체에서 newCondition() 메서드를 통해 생성된다.

Wait&Notify의 경우 synchronized 블록과 결합이 되는 반면, Condition 객체는 Lock과 결합되어 명시적 락을 사용할 수 있다. 또한 타임아웃 및 인터럽트 처리가 더 유연하다.

await()

현재 쓰레드를 기다리게 한다.

signal()

대기 중인 쓰레드 중 하나를 깨운다

signalAll()

대기 중인 모든 쓰레드를 깨운다.

Condition 예제

import java.util.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

class Main {
    public static void main(String[] args) throws InterruptedException {
        Table table = new Table();
        new Thread(new Cook(table), "COOK1").start();
        new Thread(new Customer(table, "donut"), "CUSTOMER_1").start();
        new Thread(new Customer(table, "burger"), "CUSTOMER_2").start();
    }
}

class Table {
    private final int MAX_FOOD = 5;
    public String[] dishNames = {"donut", "donut", "burger"};
    public ArrayList<String> dishes = new ArrayList<String>();

    private ReentrantLock lock = new ReentrantLock();
    private Condition forCook = lock.newCondition();
    private Condition forCustomer = lock.newCondition();

    public int dishNum(){return dishNames.length;}
    public void add(String food) {
        lock.lock();
        try {
            // 꽉 차있을 때
            while (dishes.size() >= MAX_FOOD) {
                String name = Thread.currentThread().getName();
                System.out.println(name + "is waiting...");
                try {
                    forCook.await();
                    Thread.sleep(500);
                }
                catch (InterruptedException e) { }
            }

            dishes.add(food);
            forCustomer.signal(); // 음식이 추가 될 때 기다리는 고객에게 알려준다.
            System.out.println("Dishes:" + dishes.toString());
        } finally {
            lock.unlock();
        }
    }

    public void remove(String food){
        lock.lock();
        try{
            while(dishes.size() == 0){
                System.out.println(Thread.currentThread().getName() + "is waiting...");
                try{
                    forCustomer.await();
                    Thread.sleep(500);
                }
                catch (InterruptedException e) { }
            }
            while(true) {
                for (int i = 0; i < dishes.size(); i++) {
                    if (dishes.get(i).equals(food)) {
                        dishes.remove(i);
                        forCook.signal();
                        return;
                    }
                }

                try {
                    System.out.println(Thread.currentThread().getName() + "is waiting...");
                    forCustomer.await();
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                }
            }
        }
        finally {
            lock.unlock();
        }
    }
}
class Cook implements Runnable{
    public Cook(Table t) {
        table = t;
    }
    Table table;
    @Override
    public void run() {
        while(true){
            int randIdx = (int)(Math.random() * table.dishNum());
            table.add(table.dishNames[randIdx]);
            try{
                Thread.sleep(10);
            } catch (InterruptedException e) {}
        }
    }
}
class Customer implements Runnable {
    public Customer(Table t, String food) {
        table = t;
        this.food = food;
    }
    Table table;
    String food;

    @Override
    public void run() {
        while(true){
            try{
                Thread.sleep(100);
            } catch (InterruptedException e) {}
            table.remove(food);
            System.out.println(Thread.currentThread().getName() + " ate a " + food);
        }
    }

    String name;
}