자바 동기화 클래스

멀티 스레드 환경에서 하나의 프로세스는 다수의 스레드에 의해 실행된다.  스레드가 사용하는 독립적인 자원을 제외하고 스레드끼리 공유하는 자원 [ 스택 영역 제외 ] 을 사용하려면 동기화 작업이 반드시 필요하다. 

 

동기화

- 스레드들에게 하나의 자원에 대한 처리권한을 주거나 순서를 조정해주는 기법.

 

자바는 스레드의 동기화를 위해 concurrent package에 3개의 동기화 클래스를 제공한다.

 

public class ConCurrency {

    public static void main(String[] args) {

        Thread thread = new Thread(() -> {

            System.out.println("Thread Start");

            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        });

        thread.start();

        try {
            thread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("Thread Finish");
    }
}

join()을 사용해서 메인스레드를 대기시킨다. 메인스레드와 동기화 시킬 수 있지만 스레드 여러개를 동시에 시작하게 할 순 없다. 

 

1. java.util.concurrent.Semaphore

세마포어는 permits size를 제공하여 acquire() 메서드로 permits 허용시키고 release()로 permits 반환한다.

public class ConCurrency {

    public static void main(String[] args) throws InterruptedException {

        AtomicInteger atomicInteger = new AtomicInteger(0); //초기값 0
        Semaphore semaphore = new Semaphore(10);

        for (int i=0; i<10; i++) {
            semaphore.acquire();
            atomicInteger.getAndIncrement();
        }

        semaphore.acquire();
        System.out.println(atomicInteger.get());
    }
}

11번째 acquire() 를 호출할 경우 대기 상태에 빠지므로 11이 출력되지 않는다.

 

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class ConCurrency {

    public static void main(String[] args) {

        Semaphore semaphore = new Semaphore(10);
        ExecutorService service = Executors.newFixedThreadPool(5);

        for (int i=0; i<20; i++) {
            int n = i;
            service.execute(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(n);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        System.out.println("finish");

//        finish
//        4
//        3
//        0
//        2
//        1
//        8
//        7
//        6
//        5
//        9
    }
}

메인스레드 제어도 못하고 세마포어 크기가 10이므로 20번 acquire()할 경우 대기상태에 빠진다. 

 

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class ConCurrency {

    public static void main(String[] args) {

        Semaphore semaphore = new Semaphore(10);
        ExecutorService service = Executors.newFixedThreadPool(5);

        for (int i=0; i<20; i++) {
            int n = i;
            service.execute(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(n);
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        System.out.println("finish");

//        0
//        1
//        finish
//        3
//        2
//        6
//        5
//        9
//        11
//        12
//        8
//        7
//        15
//        16
//        17
//        18
//        4
//        19
//        14
//        13
//        10

    }
}

release()로 permits()을 줄여서 20번 실행할 수 있지만 메인스레드 제어가 안된다.

 

2. java.util.concurrent.CountDownLatch

하나 이상의 스레드가 다른 스레드에서 수행중인 일련의 작업이 완료될때까지 대기 할 수 있다.

 

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ConCurrency {

    public static void main(String[] args) throws InterruptedException {

        CountDownLatch latch = new CountDownLatch(10);
        ExecutorService service = Executors.newFixedThreadPool(5);

        for (int i=0; i<5; i++) {
            int n = i;
            service.execute(() -> {
                latch.countDown();
                System.out.println(n);
            });
        }

        latch.await();
        System.out.println("finish");
        
//        결과
//        0
//        4
//        3
//        1
//        2
    }
}

countDown()이 10번 실행되지 않으면 메인쓰레드에서 await() 해도 finish가 찍히지 않는다.  

 

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ConCurrency {

    public static void main(String[] args) throws InterruptedException {

        CountDownLatch latch = new CountDownLatch(10);
        ExecutorService service = Executors.newFixedThreadPool(5);

        for (int i=0; i<10; i++) {
            int n = i;
            service.execute(() -> {
                latch.countDown();
                System.out.println(n);
            });
        }

        latch.await();
        System.out.println("finish");

//        결과
//        1
//        3
//        0
//        7
//        8
//        2
//        9
//        6
//        5
//        4
//        finish


    }
}

10번 countDown() 실행되면 await()으로 중단됐던 코드가 다시 진행되므로 finish 출력된다.  

 

3. java.util.concurrent.CyclicBarrier

public class ConCurrency {

    public static void main(String[] args) throws InterruptedException {

        CyclicBarrier cyclicBarrier = new CyclicBarrier(10);
        ExecutorService service = Executors.newFixedThreadPool(5);

        for (int i=0; i<5; i++) {
            int n = i;
            service.execute(() -> {
                try {
                    cyclicBarrier.await();
                    System.out.println(n);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }
        
        System.out.println("finish");
        
//        결과
//        finish
        
    }
}

CountDownLatch는 카운트가 소모되기전에 await()을 호출할 경우 해당 스레드를 제외한 나머지 스레드들은 여전히 실행되지만 CyclicBarrier는 어느 스레드가 await()을 만나도 이후에 실행될 모든 스레드들도 멈춘다.