쓰레드의 실행 제어

13.8~13.9

8. 쓰레드의 실행 제어

• 쓰레드의 상태와 관련된 메서드([]는 가변 인자)
  • static void sleep(long mills, [int nanos]): 시간 동안 쓰레드 일시 정지, 시간이 지나면 실행 대기
  • void join([long millis, int nanos]): 시간 동안 쓰레드 실행, 시간이 지나거나 작업이 종료되면 join()을 호출한 쓰레드로 돌아와 실행
  • void interrupt(): sleep()이나 join()에 의해 일시 정지 상태인 쓰레드를 깨워서 실행 대기 상태로 만듬, 해당 쓰레드에서는 InterrutedException이 발생함으로써 일시정지 해제
  • static void yield(): 실행 중에 자신에게 주어진 실행 시간을 다른 쓰레드에게 양보하고 자신은 실행 대기 상태가 됨
• 쓰레드의 상태
  • NEW: 쓰레드가 생성되고 아직 start()가 호출되지 않은 상태
  • RUNNABLE: 실행 중 또는 실행 가능한 상태
  • BLOCKED: 동기화 블럭에 의해서 일시 정지된 상태
  • WAITING, TIMED_WAITING: 쓰레드의 작업이 종료되지는 않았지만 실행 가능하지 않은 일시정지 상태, 일시 정지 시간이 지정된 경우 TIMED_WAITING
  • TERMINATED: 쓰레드의 작업이 종료된 상태

9. 쓰레드의 동기화

• 한 쓰레드가 진행 중인 작업을 다른 쓰레드가 간섭하지 못하도록 막는 것
• 나중에 실행된 쓰레드의 공유 자원 사용으로 같은 자원을 공유하는 이전에 실행된 쓰레드의 작업 결과에 영향을 줄 수 있음
• 공유 데이터의 lock를 획득한 하나의 쓰레드가 임계 영역의 코드를 수행하면 lock반납, 다른 쓰레드가 반납된 lock을 획득하여 임계 영역의 코드 수행
• 임계 영역: 쓰레드가 공유하는 공유 자원에 접근하는 코드, 즉 쓰레드의 독점을 보장해야하는 영
• 뮤텍스(Mutex), 세마포어(Semaphor), 모니터(Monitor)
  • 운영체제의 상호 배제 동기화 기법
  • 뮤텍스
    • 임계 영역을 가진 쓰레드들이 겹치지않고 단독으로 실행시키는 기법
    • Locking과 Unlocking 사용
    • 쓰레드는 lock을 얻고 공유 자원 접근, 이전 쓰레드가 lock을 반납하면 다음 쓰레드가 공유 자원 접근 가능
  • 세마포어
    • 접근 가능한 공유 자원의 개수를 나타냄
    • 세마포어의 연산
      • initialize: 양의 정수로 세마포어 초기화
      • decrement: 프로세스를 블록
      • increment: 블록된 프로세스를 깨움
    • 세마포어의 값에 따라서 운영체제는 스레드가 즉시 자원을 사용할지 결정

    임계 영역이 공유하는 자원이 5개일때, 세마포어가 5가 되며, 세마포어는 임계 영역에 5개까지 쓰레드의 공유 자원 접근을 허용. 공유 자원에 접근하는 스레드가 n개일 때 세마포어는 공유 자원의 개수 - n

    • 임계 영역의 상호 배제를 깨지만 자원 자체에서는 상호 배제를 유지함(반드시 하나의 자원을 하나의 쓰레드가 사용)

  • 모니터
    • 상호 배타 큐(Mutual Exclusion Queue)
      • 공유 자원에 하나의 프로세스만 진입하도록 하기 위한 큐. 공유 자원을 사용하는 쓰레드가 존재하면 상호 배타 큐에 존재
    • 조건 동기 큐(Conditional Synchronization Queue)
      • 공유 자원의 Lock이 해제되기를 기다리는 스레드가 대기하는 큐
    • 공유 자원을 점유중인 쓰레드가 Lock 소유
    • 공유 자원을 점유중인 쓰레드가 존재할 때, 다른 쓰레드가 공유자원에 접근하려고하면 조건 동기 큐에서 대기
• synchronized를 이용한 동기화(모니터 사용)

  • 메서드의 전체를 임계 영역으로 지정

      public synchronized void calcSum() { // 메서드 전체를 임계 영역으로 지정
      }
    

    • 쓰레드는 synchronized메서드가 호출된 시점부터 해당 메서드가 포함된 객체의 lock을 얻어 작업을 수행, 메서드가 종료되면 lock반환

  • 특정한 영역을 임계 영역으로 지정(synchronized 블럭)

      synchronized(객체의 참조 변수) {
      }
    

    • 참조변수는 lock을 걸고자하는 객체를 참조
    • 블럭의 안으로 진입하면서 쓰레드는 지정된 객체의 lock 획득, 블럭을 벗어나면 lock 반환
  • 두 방법 모두 lock의 획득과 반환이 자동으로 이루어짐
  • 모든 객체는 lock을 하나씩 가지고 있으며 해당 객체의 lock을 가지고 있는 쓰레드만 임계 영역의 코드 수행 가능, 다른 쓰레드들은 lock 획득까지 대기
  • 임계 영역은 멀티 쓰레드 프로그램의 성능에 주는 영향이 큼 → 임계 영역 최소화 필요
  • synchronized 블럭에 쓰레드가 들어갈 때와 나올 때 캐시와 메모리가 동기화됨
• wait()과 notify()
  • 동기화된 임계 영역의 코드를 수행하다가 작업을 더 이상 진행할 상황이 아니면, 일단 wait() 호출로 lock 반납하고 대기, 진행할 상황이 되면 notify()호출로 lock 획득 및 재실행
  • waiting pool: 쓰레드가 대기하는 장소, 객체마다 존재하며 notify(), notifyall()은 해당 객체의 waiting pool에 있는 쓰레드만 깨움
  • wait(): 실행중인 쓰레드는 해당 객체의 대기실에서 통지 대기, 시간이 지정되면 지정된 시간동안만 대기
  • notify(): 해당 객체의 대기실에 있던 임의의 쓰레드에게 통지
  • notifyAll(): 대기중인 모든 쓰레드에게 통지, 단, lock은 하나의 쓰레드만 획득
• Lock과 Condition을 이용한 동기화
  • java.util.concurrent.locks 패키지에서 제공
  • ReentrantLock
    • 재진입이 가능한 lock
    • 일반적인 배타 lock
  • ReentrantReadWriteLock
    • 읽기를 위한 Lock과 쓰기를위한 Lock 제공
    • 읽기에 공유적이고 쓰기에 배타적인 Lock
    • 읽기에 lock가 걸리면 다른 쓰레드가 읽기 lock을 중복해서 걸고 읽기 수행 가능
    • 읽기 lock 중 쓰기 lock 불가능
    • 쓰기 lock 중 읽기 lock 불가능
  • StampedLock
    • lock을 걸거나 해제할 때 스탬프를 사용
    • 읽기 쓰기를 위한 lock 외에 낙관적 읽기 lock을 사용
    • 낙관적 읽기lock은 쓰기 lock에 의해 바로 해제
    • 낙관적 읽기에 실패하면, 읽기 lock을 얻어서 다시 읽어야함
    • 무조건 읽기 lock을 걸지 않고, 쓰기와 일기가 충돌할 때 만 쓰기가 끝난 후에 읽기 lock을 거는 방식
  • volatile
    • volatile 가 붙은 변수는 코어가 변수의 값을 읽을 때 캐시가 아닌 메모리에서 읽음
    • 메모리에 저장된 값이 변경되었는데 캐시에 저장된 값이 갱신되지 않아 생기는 문제를 해결(코어는 캐시에 값이 있다면 메모리보다 캐시를 우선 사용)
• 원자화
  • JVM은 데이터를 4byte 단위로 처리 → 크기가 int 이하인 타입들은 하나의 명령어(작업의 최소 단위)로 읽고 쓰기가 가능, 작업 중간에 다른 쓰레드가 끼어들 수 없음
  • 4byte를 초과하는 타입은 다른 쓰레드가 작업중 끼어들 수 있음
  • volatile 키워드로 변수를 원자화 시킬 수 있음(나눌 수 없는 작업으로 만듬)
• fork & join 프레임워크
  • 작업을 작은 단위로 나눠서 여러 쓰레드가 동시에 처리하는 것을 쉽게 만들어줌
  • 수행할 작업에 따라 클래스 상속 및 구현 필요
    • RecursiveAction: 반환값이 없는 작업을 구현할 떄 사용
    • RecursiveTask: 반환값이 있는 작업을 구현할 떄 사용
  • 상속을통해 compute() 메서드 구현 필요
    • 수행할 작업을 compute()메서드로 구현
    • compute() 구현에서 수행 작업을 반으로 나누고 생성자를 나눈 작업으로 호출, 하나는 compute()로 작업을 계속해서 반으로 나누고 다른 쓰레드는 fork()에 의해서 작업 큐에 추가된 작업을 수행하는 재귀호출구조 가능 이때 fork()호출로 작업 큐에 추가된 작업은 compute()를 통해 나눌 수 없을 때까지 나눠지고, 자신의 작업 큐가 비어있는 쓰레드는 다른 쓰레드의 작업큐에서 작업을 가져와서 수행(작업 훔쳐오기)
  • ForkJoinPool(): 쓰레드 풀 생성자

    • 쓰레드 풀 인스턴스의 invoke() 메서드의 인자에 프레임워크를 상속받은 객체의 인스턴스를 전달함으로써 작업 시작

        ForkJoinPool pool = new ForkJoinPool();
        SumTask task = new SumTask(from, to); //수행할 작업 생성
        Long result = pool.invoke(task);      //쓰레드풀에서 task 수행
      

  • fork()
    • 작업을 쓰레드 풀의 작업 큐에 넣음, 비동기
  • join()
    • 작업의 수행이 끝나면 결과 반환, 동기

  • 예제

      class SumTask extends RecursiveTask<Long> {
      	long from, to;
            
      	SumTask(long from, long to) {
      		this.from = from;
      		this.to = to;
      	}
      	public Long compute() {
      		long size = to - from + 1;
            
      		if (size <= 5)
      			return sum(); //sum()은 from부터 to까지반환, 구현생략
      		//범위를 나누어 두개의 작업 생성
      		long half = (from+to) / 2;
            	
      		SumTask leftSum = new SumTask(from, half);
      		SumTask rightSum = new SumTask(half+1, to);
            
      		leftSum.fork();
            
      		return rightSum.compute() + leftSUm.join();
      	}
      }