자바 - Thread(4) 동기화

 

1.쓰레드의 동기화란

 

- 멀티쓰레드 프로세스의 경우 여러 쓰레드가 같은 프로세스 내의 자원을 공유해서 작업하기 때문에 서로 작업에 영향을 주게됨

- 따라서, 한 쓰레드가 특정 작업을 마무리하기 전까지 다른 쓰레드에 의해 방해받지 않도록 하는 것이 필요.

- 이러한 작업을 위해 도입된 개념이 임계영역(critical section)과 잠금(lock)이다.

- 공유 데이터를 사용하는 코드 영역을 임계 영역으로 지정 -> 쓰레드가 데이터를 사용할 때 lock을 걸고 사용 후 unlock 

- lock이 걸려있는 상태에서는 다른 쓰레드는 해당 데이터에 접근할 수 없게 함

 

정리하자면, 동기화란 한 쓰레드가 진행 중인 작업을 다른 쓰레드가 간섭하기 못하도록 막는 것. 그리고 이를 위해 임계영역 지정과 잠금이 필요

-하나의 작업을 여러 쓰레드로 처리할 때, 혹은 여러 쓰레드가 하나의 자원에 접근할때 동기화를 고려해야한다.

자료출처: https://tecoble.techcourse.co.kr/post/2021-10-23-java-synchronize/

 

9.1 synchronized를 이용한 동기화

 

synchronized 키워드를 이용하면 쉽게 동기화를 사용할 수 있다. synchronized를 이용한 동기화 방법은 아래와 같이 두가지 방법이 있다.

 

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

 public synchronized void 메서드명 () {} //메서드 전체 임계영역 

 

- 특정 영역을 임계 영역으로 지정 

 synchronized(객체의 참조변수) {       //임계영역}두 방법 모두 lock과 unlock이 자동으로 이루어진다. 따라서 임계영역만 적절하게 설정해 주면 된다.

 

*임계 영역은 멀티 쓰레드 프로그램의 성능을 좌우하기 때문에 가능하면 메서드 전체보단 블럭별로 지정하는 것이 낫다.  *이때 참조변수는 락을 걸고자하는 객체(데이터)를 참조하는 것이어야 한다. -> 동기화는 쓰레드가 아닌 공유 데이터 영역에서 이뤄진다.  

public class ThreadEX1  {
    public static void main(String[] args) throws Exception{
        Runnable r = new Threadex();
        new Thread(r).start();
        new Thread(r).start();
    }
}
class Account {
    private int balance = 1000; // balance는 private이어야 동기화하는 목적에 알맞다.
    public int getBalance(){
        return this.balance;
    }

    public synchronized void withdraw(int money){ //한 쓰레드가 lock을 얻고 인출시, 다른 쓰레드 접근 금지
        if(balance>=money){
            try {
                Thread.sleep(1000);
            }catch (InterruptedException e){}
            balance -= money;
        }
    }
}

class Threadex implements Runnable{
    Account acc = new Account();

    public void run(){
        while (acc.getBalance() >0){
            int money = (int)(Math.random() *3 +1)*100;
            acc.withdraw(money);
            System.out.println("balance :"+acc.getBalance());
        }

    }
}

//만약에 동기화를 하지 않았으면, 쓰레드가 sleep하는동안 다른쓰레드가 인출하면서, balance가 음수가 될 수 있다.

9.2 wait()과 notify()

 

- 공유 데이터를 통해서 데이터를 보호하는 과정에서 Lock이 걸린 한 작업이 오랜 시간이 걸려 다른 쓰레드가 모든 작업을 멈추고 기다리게 된다면,  효율적인 멀티쓰레드 프로그램을 만들지 못할 것이다. 

- 이러한 문제를 해결하고 동기화의 효율을 높이기 위해 wait()과 notify()가 있다.

- 마치 빵집에서 원하는 빵이 나올 때 까지 순서를 양보하다가, 원하는 빵이 나오면 사는 것과 같다.

 

 

- void wait() : 쓰레드가 lock을 반납(unlock)하고 waiting pool에서 통지를 기다린다.

- notify() -: notify()가 호출되면, 해당 객체의 대기실에 있던 모든 쓰레드 중에서 임의의 쓰레드만 통지를 받고, 중단했던 작업을 다시 lock을 걸고 진행한다.

- notifyAll() - 객체 대기실에 모든 쓰레드에게 통지한다. 다만, lock을 얻을 수 있는 쓰레드는 하나 뿐이고 나머지는 다시 대기 

 

위 메서드들의 특징은 다음과 같다.

- Object에 정의되어 있다.

- 동기화 블럭 내에서만 사용할 수 있다.

 

* waiting pool은 객체마다 존재한다. 따라서 notifyAll이 호출되면 notifyAll이 호출된 객체의 waiting pool에서 대기 중인 쓰레드만 통지를 받는 것이지 모든 객체에 waiting pool에 있는 쓰레드가 통지를 받는 것은 아니다.

public class ThreadEX1  {
    public static void main(String[] args) throws Exception{
      Table t = new Table();

      new Thread(new Cook(t),"Cook1").start();
      new Thread(new Customer(t,"dount"),"Cust1").start();
      new Thread(new Customer(t,"burger"),"Cust2").start();

      Thread.sleep(5000);
      System.exit(0);

    }
}

class Cook implements Runnable{
  private Table ta;
  Cook(Table ta){this.ta = ta; }

    public void run(){
       while(true){
        //임의의 요리를 하나 선택해서 테이블 위에 올려둔다.
           int idx = (int)(Math.random() * ta.dishNum());
           ta.add(ta.dishVariable[idx]);
           try{Thread.sleep(100);}catch (InterruptedException e){};
       }
  }

}

class Customer implements Runnable{
    private Table ta;
    private String food;
    Customer(Table ta, String food){
        this.ta = ta;
        this.food =food;
    }
    @Override
    public void run() {
        while(true) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
            }
            String name = Thread.currentThread().getName();

            if (eatFood())
                System.out.println(name + " ate a" + food);
            else
                System.out.println(name + " failed to eat :(");
        }
    }

    boolean eatFood(){ return  ta.remove(food); }
}

class Table{
    String[] dishVariable ={"dount","dount","burger"}; //테이블에 올라 갈 수 있는 음식
    final int MAX_FOOD = 6; //테이블 최대 음식 수
    private ArrayList<String> dishesOnTable = new ArrayList<>(); //테이블에 올라간 음식

    public synchronized void add(String dish){ //요리사 쓰레드가 접근
        //테이블에 음식을 두는 메서드 ,음식이 가득차면 테이블에 음식 추가 x
        if(dishesOnTable.size() >= MAX_FOOD){
            return;
        }
        dishesOnTable.add(dish);
        System.out.println("Dishes:"+dishesOnTable.toString());
    }

    public boolean remove(String dishName){ //손님 쓰레드가 접근 
        // 지정된 요리와 일치하는 요리를 테이블에서 제거한다.
        synchronized (this) {
            while(dishesOnTable.size() ==0){ // 0.5초마다 테이블에 음식이 있는지 확인
                String name =Thread.currentThread().getName();
                System.out.println(name +" is waiting");
                try { Thread.sleep(500);}catch (InterruptedException e){} 
            }
            for (int i = 0; i < dishesOnTable.size(); i++) {
                if (dishName.equals(dishesOnTable.get(i))) {
                    dishesOnTable.remove(i);
                    return true;
                }
            }
        } // synchronized
        return false;
    }
    public int dishNum() {return dishVariable.length; }

}

임계영역만 지정했을때, 무한대기 현상이 발생 이는 customer 쓰레드가 table객체에 lock을 계속 가지고 있어서 cook쓰레드가 table 객체에 접근할 수 없기 때문에 발생한다.

 

public class ThreadEX1  {
    public static void main(String[] args) throws Exception{
      Table t = new Table();

      new Thread(new Cook(t),"Cook1").start();
      new Thread(new Customer(t,"dount"),"Cust1").start();
      new Thread(new Customer(t,"burger"),"Cust2").start();

      Thread.sleep(5000);
      System.exit(0);

    }
}

class Cook implements Runnable{
  private Table ta;
  Cook(Table ta){this.ta = ta; }

    public void run(){
       while(true){
           int idx = (int)(Math.random() * ta.dishNum());
           ta.add(ta.dishVariable[idx]);
           try{Thread.sleep(10);}catch (InterruptedException e){};
       }
  }
  
}

class Customer implements Runnable{
    private Table ta;
    private String food;
    Customer(Table ta, String food){
        this.ta = ta;
        this.food =food;
    }
    @Override
    public void run() {
        while(true) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
            }
            String name = Thread.currentThread().getName();
            ta.remove(food);
            System.out.println(name + " ate a "+ food);
        }
    }
}

class Table{
    String[] dishVariable ={"dount","dount","burger"};
    final int MAX_FOOD = 6;
    private ArrayList<String> dishesOnTable = new ArrayList<>();

    public synchronized void add(String dish){

        while(dishesOnTable.size() >= MAX_FOOD){
            String name = Thread.currentThread().getName();
            System.out.println(name +" is waiting.");
            try {
                wait(); //음식이 꽉차서 lock 돌려주고 요리사 wait
                Thread.sleep(500);
            }catch (InterruptedException e){}

        }
        //음식이 꽉찬게 아니거나, wait()상태에서 깨어나면 실행
        dishesOnTable.add(dish);
        notify(); //음식을 추가했으니 기다리고 있던 손님에게 알리기
        System.out.println("Dishes:"+dishesOnTable.toString());
    }

    public void remove(String dishName){

        synchronized (this) {
            String name =Thread.currentThread().getName();
            while(dishesOnTable.size() ==0){
                System.out.println(name +" is waiting");
                try {
                    wait(); //음식이 없으면 lock 넘기고 손님 쓰레드 대기
                    Thread.sleep(500);
                }catch (InterruptedException e){}
            }
            while (true) { //음식이 테이블에 있다면,
                for (int i = 0; i < dishesOnTable.size(); i++) {
                    if (dishName.equals(dishesOnTable.get(i))) {
                        dishesOnTable.remove(i);
                        notify(); //테이블이 꽉차서 기다리던 요리사 깨우기
                        return;
                    }
                }
                try{
                    System.out.println(name +" is waiting"); //원하는 음식 없어서 대기
                    wait();
                    Thread.sleep(500);
                }catch (InterruptedException e){}

            }//while(true)
        } // synchronized

    }
    public int dishNum() {return dishVariable.length; }

}

wait과 notify를 추가하여 무한대기가 없도록 변경하였다. 이 코드에도 문제가 있는데, waiting pool에 손님과 요리사가 동시에 대기를 하기 때문에 notify가 호출되었을 때, 손님쓰레드를 깨울 수도 있고, 요리사를 깨울 수도 있다. 즉 어떤 쓰레드를 깨울 것인지 지정할 수 없으므로, 

대기시간이 길어지는 문제가 발생할 수 있다.

 

기아 현상과 경쟁 상태 

 

기아 현상 : 쓰레드가 계속 통지를 받지 못하고 계속 기다리게 되는 현상

-> 요리사 쓰레드가 너무 오래 기다린다. 이를 해결하기 위해 notifyAll()을 써볼 수 있다.

 

경쟁 상태 : 쓰레드간 lock을 얻기 위해 경쟁하는 상태 의미 -> 경쟁 상태를 개선하기 위해서는 각 쓰레드를 작업에 따라 구분할 필요가 있다.

->notifyAll()로 인해 각 쓰레드들이 lock을 얻기 위해 경쟁을 한다.

 

위 문제를 개선하기 위해서는 요리사 쓰레드와 손님 쓰레드를 구분할 필요가 있다. 다음 Lock와 Condition을 이용하면 이를 해결할 수 있다.

2. Lock와 Condition을 이용한 동기화 

 

- Synchronized블럭 외에 java.util.concurrent.locks 패키지를 추가하면 lock클래스를 이용해서 동기화를 구현할 수 있다.

- lock클래스를 이용하면 wait()과 notify()에서 하지 못했던 선별적 통지가 가능하여 두 메서드의 단점을 보완할 수 있다.

 

lock클래스의 종류는 다음 3가지가 있다.

 

-ReentrantLock : 재진입이 가능한 lock 가장 일반적인 배타 lock

-ReentrantReadWriteLock : 읽기에는 공유적이고, 쓰기에는 배타적인 lock

-StampedLock : ReentrantReadWriteLock에 낙관적인 lock의 가능을 추가 했다.

 

2.1 ReentrantLock 생성자 및 메서드 

 

 생성자

 

  -ReentrantLock()

  -ReentrantLock(boolean fair)

 

생성자의 매개변수를 true로 주면, lock이 풀렸을 때 가장 오래 기다린 쓰레드가 lock을 획득할 수 있게 공정하게 처리한다.

->단 성능이 떨어진다. 대부분 공정함 보단 성능을 택한다.

 

lock메서드

 

-void lock() 

-void unlock()

-boolean isLocked() -lock가 잠겼는지 확인한다.

-boolean tryLock() 

-boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException < - interrupt() 

: 다른쓰레드에 락이 걸려있으면, 락을 얻으려고 기다리지 않거나, 지정된 시간만큼만 기다린다. lock를 얻으면 true 못얻으면 false

->락 못얻으면 다음 작업을 시도할 것인지 포기할 것인지 지정 가능 

 

synchronized 블럭과 달리 lock클래스들은 수동으로 lock을 잠그고 해제해야한다. 

-lock.lock() ---- lock.unlock() 까지가 임계영역이 된다.

-보통 임계영역 내에서 예외가 발생하거나 return문으로 빠져나가게 되면, lock이 풀리지 않을 수 있음으로 try-finally문으로 감싸는 것이 일반적이다.

 

 

 

2.2 ReentrantLock과 Condition

 

-Condition은 이미 생성된 lock으로부터 newCondition()을 호출해서 생성한다.

-Condition은 wait()과 notify()대신에  await()과 signal()을 사용한다.

 

public class ThreadEX1  {
    public static void main(String[] args) throws Exception{
      Table t = new Table();

      new Thread(new Cook(t),"Cook1").start();
      new Thread(new Customer(t,"dount"),"Cust1").start();
      new Thread(new Customer(t,"burger"),"Cust2").start();

      Thread.sleep(2000);
      System.exit(0);

    }
}

class Cook implements Runnable{
  private Table ta;
  Cook(Table ta){this.ta = ta; }

    public void run(){
       while(true){
           int idx = (int)(Math.random() * ta.dishNum());
           ta.add(ta.dishVariable[idx]);
           try{Thread.sleep(10);}catch (InterruptedException e){};
       }
  }

}

class Customer implements Runnable{
    private Table ta;
    private String food;
    Customer(Table ta, String food){
        this.ta = ta;
        this.food =food;
    }
    @Override
    public void run() {
        while(true) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
            }
            String name = Thread.currentThread().getName();
            ta.remove(food);
            System.out.println(name + " ate a "+ food);
        }
    }
}

class Table{
    String[] dishVariable ={"dount","dount","burger"};
    final int MAX_FOOD = 6;
    private ArrayList<String> dishesOnTable = new ArrayList<>();

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

    public void add(String dish){
        lock.lock(); //lock시작
       try {//unlock을 위한 try-finally문
           while (dishesOnTable.size() >= MAX_FOOD) {
               String name = Thread.currentThread().getName();
               System.out.println(name + " is waiting.");
               try {
                   forCook.await(); //음식이 꽉차서 lock 돌려주고 요리사 await
                   Thread.sleep(500);
               } catch (InterruptedException e) {
               }

           }

           //음식이 꽉찬게 아니거나, await()상태에서 깨어나면 실행
           dishesOnTable.add(dish);
           forCust.signal(); //음식을 추가했으니 기다리고 있던 손님에게 알리기
           System.out.println("Dishes:" + dishesOnTable.toString());
       }finally {
           lock.unlock();
       }
    }

    public void remove(String dishName){

       lock.lock();
       String name =Thread.currentThread().getName();
        try {
            while (dishesOnTable.size() == 0) {
                System.out.println(name + " is waiting");
                try {
                    forCust.await(); //음식이 없으면 lock 넘기고 손님 쓰레드 대기
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                }
            }
            while (true) { //음식이 테이블에 있다면,
                for (int i = 0; i < dishesOnTable.size(); i++) {
                    if (dishName.equals(dishesOnTable.get(i))) {
                        dishesOnTable.remove(i);
                        forCook.signal(); //테이블이 꽉차서 기다리던 요리사 깨우기
                        return;
                    }
                }
                try {
                    System.out.println(name + " is waiting"); //원하는 음식 없어서 손님 대기
                    forCust.await();
                    Thread.sleep(500);
                } catch (InterruptedException e) {}

            }//while(true)
        }finally {
            lock.unlock();
        }


    }
    public int dishNum() {return dishVariable.length; }

 

3. volatile

멀티코어에서 cpu가 메모리에서 값을 읽어와서 작업하는 과정 

 

출처:https://incheol-jung.gitbook.io/docs/q-and-a/java/thread

 

- 멀티코어 프로세서에서는 코어별로 캐시메모리를 가지고 있다.

- 코어는 메모리에서 읽어온 값을 캐시에 저장하고, 캐시에서 값을 읽어서 작업한다.

- 코어가 다시 같은 값을 읽어올 때, 먼저 같은 값이 캐시에 있는지 확인하고, 없다면 메모리에서 읽어온다.

- 이러한 이유로 메모리에 저장된 값을 변경시켰는데, 그 사이에 캐시에 저장된 값이 갱신되지 않아서 저장된 값과 캐시의 값이 다른 경우가 발생한다. 

 

이러한 문제를 해결하기 위해서 변수앞에 volatile을 붙이면 해결할 수 있다. volatile이 붙은 변수는 코어가 변수의 값을 읽어올 때, 캐시가 아닌 메모리에서 읽어오기 때문에 캐시와 메모리간의 값의 불일치가 해소된다.

 

출처:https://incheol-jung.gitbook.io/docs/q-and-a/java/thread

 

3.2 volatile로 long과 double 원자화

 

-JVM은 데이터를 4바이트 단위로 처리한다. int까지의 타입들의 변수는 한 번에 읽거나 쓰는 것이 가능하다.

-이는 더 이상 나눌 수 없는 최소의 작업단위이므로, 작업 중간에 다른 쓰레드가 끼어들 틈이 없다.

- 이보다 더 큰 자료형인 long과 double타입의 변수는 4바이트 이상의 자료형이므로, 한 번에 읽거나 쓸 수 없으므로, 도중에 

다른 쓰레드가 끼어들 틈이 있다. 

-이를 방지하기 위해 변수를 읽고 쓰는 문장에 synchronized블럭을 이용할 수 도 있지만, 변수 선언시에 volatile을 붙이면 더 편하게 

방지할 수 있다.

 

*상수에는 volatile을 붙일 수 없다. 상수는 변하지 않는 값이므로 멀티쓰레드에 안전하다.

 

- volatile은 해당 변수에 대한 읽거나 쓰기를 원자화 한다. 

- 원자화라는 것은 작업을 더 이상 나눌 수 없게 한다는 의미이다 -> synchronized블럭도 일종의 원자화라고 할 수 있다.

- 단, volatile은 원자화일 뿐 동기화 시키는 것은 아니다

 

4. fork & join 프레임 워크

 

하나의 작업을 작은 단위로 나눠서 여러 쓰레드가 동시에 처리하는 것을 쉽게 만들어주는 프레임 워크이다. JDK 1.7부터 추가됨

 

-생성

 

수행할 작업 클래스는 다음 아래 두 클래스를 상속받아서 구현해야한다.

 

RecursiveAction : 반환값이 없는 작업을 구현할 때 사용

RecursiveTask : 반환값이 있는 작업을 구현할 때 사용

 

상속 받은 뒤 추상메서드 compute()메서드를 구현하면 된다.

 

-사용 

 

쓰레드풀과 수행할 작업 클래스 인스턴스를 생성하고 invoke()로 작업을 시작한다.  (마치 run()과 start()관계)

ForkJoinPool pool = new ForkJoinPool() // 쓰레드 풀 생성

SumTask task = new SumTask(from,to) //수행할 작업 생성

Long result = pool.invoke(task) // 생성한 쓰레드 풀이 작업을 수행하도록 지정 

 

-프레임 웍에서 제공하는 지정된 수의 쓰레드가 미리 생성되어 반복해서 재사용할 수 있다.

-쓰레드 풀은 쓰레드가 수행해야하는 작업이 담긴 큐를 제공하며, 각 쓰레드는 자신의 작업 큐에 담긴 작업을 순서대로 처리한다.

 

4.1 compute()구현

 

- compute()는 수행할 작업과, 작업을 어떻게 나눌 것인가에 대해 명시해야한다. - 작업은 더 이상 나누어질 수 없는 단위가 존재하고, 이에 따른 작업 중단점이 있다.- 때문에 compute()으 구조는  재귀호출과 비슷하다고 볼 수 있다.

 

4.2 fork()와 join()

 

- fork() 해당 작업을 쓰레드 풀의 작업 큐에 넣는다. 작업 풀에 들어가면, 각 쓰레드가 작업을 할당받고 다시 compute() (비동기메서드)

- join() 해당 작업의 수행이 끝날 때까지 기다렸다가, 수행이 끝나면 결과를 반환한다 (동기메서드)

 

출처 :https://jenkov.com/tutorials/java-util-concurrent/java-fork-and-join-forkjoinpool.html

출처 :https://jenkov.com/tutorials/java-util-concurrent/java-fork-and-join-forkjoinpool.html

-작업은 의미가 있을 정도로 큰 경우에만 fork()되도록 하자 왜냐하면 fork()시 작업을 하위작업으로 나눌때에는 작업비용이 있으므로,

때에 따라 의미 없을 정도로 작은 단위로 나누면, 동시에 처리하는 것 보다 성능이 떨어질 수 있다.

 

- 작업들이 하위 작업으로 모두 나뉘고(compute()), 하위 작업이 작업을 종료하는 동안 최상위 작업은 이를 기다리고 있는다. 후에 하위 작업들이 모두 완료되고 난 뒤에 join()을 통해 하위 작업들의 결과가 최상위 작업으로 return된다. --재귀호출과 유사하다. 

 

 

참고자료

 

자바의정석 -남궁성 저

 

https://jenkov.com/tutorials/java-util-concurrent/java-fork-and-join-forkjoinpool.html

Java Fork and Join using ForkJoinPool

https://incheol-jung.gitbook.io/docs/q-and-a/java/thread

Thread(쓰레드) - Incheol's TECH BLOG

https://junghyungil.tistory.com/103

[Java] Fork Join Pool