세마포어

 

 

이전에 나온 synchronized 키워드나 락을 사용한 이유는 임계영역에 여러 스레드가 접근하는 것을 막고, 오직 하나의 스레드만 접근 가능하게하기 위해서였다.

 

세마포어도 이러한 이유 때문에 사용하는 기술이다.

 

하지만 여기서 더 나아가 세마포어는 오직 하나의 스레드만 허용하게 만들 수 있고, 더 늘릴 수도 있다. 즉, 임계영역에 접근하는 스레드의 수를 조절할 수 있는 것이다.

 

 

세마포어의 로직을 이해하기 가장 좋은 예시는 주차장이다.

 

  1. 주차장에 자리가 6자리 있는데, 모든 자리에 차가 주차되어있는 경우, 새롭게 들어온 차는 대기해야 한다.
  2. 만약 한자리가 빠지는 경우, 주차 가능한 공간이 한 자리 늘어나고, 대기하고 있던 차량 중, 가장 먼저온 차량이 비어있는 자리에 주차한다.
  3. 그 뒤에 온 차량은 계속해서 대기한다.

 

다시 세마포어에 맞게 용어를 정리하면 아래와 같다.

  • 주차장 자리 수 = 임계 영역에 동시에 접근 가능한 스레드 수
  • 새로운 주차 자리 생성 = 작업을 마친 스레드가 빠지고, 새로운 공간 확보
  • 대기중이던 차량 주차 = 대기열에 있던 스레드가 남는 공간이 확보된 것을 확인하고 임계 영역 진입
  • 차량 대기 = 대기열은 FIFO 형태의 큐이므로, 순서대로 대기

 

 

생산자 - 소비자 패턴

 

세마포어를 사용해 구현하는 생산자 - 소비자 패턴은 실시간으로 생산자가 자원을 생성해 버퍼에 넣고, 소비자가 버퍼에서 자원을 가져와 사용하는 과정을 반복하는 패턴이다.

  1. 소비자는 생산자가 자원을 추가할 때 까지 대기 → 자원 소비 락 대기
  2. 생산자는 자원 추가를 위한 작업 시작 → 자원 추가 락 획득
  3. 생산자는 자원을 공유 자원 저장소에 추가한 뒤, 추가 락을 반납하고, 소비 락을 release()
  4. 소비 락을 획득하기 위해 대기중이던 소비자는 락을 획득하고 자원을 소비
  5. 다시 1로 돌아가서 반복…

 

내가 생각하기에 이 패턴을 가장 잘 표현한 것은 웹 소켓을 기반으로 한 채팅 이나 Kafka, RabbitMQ 같은 메시지 큐이다.

  • 웹 소켓을 기반으로 한 채팅
    • 생산자 (메시지를 보내는 사용자) : 채팅 메시지를 보낼 때마다 사용자는 생산자로서 메시지를 서버에 전송
    • 소비자 (메시지를 받는 사용자) : 서버는 메시지를 받아 큐에 저장하고, 이를 구독중인 다른 사용자에게 전달

 

 

이러한 패턴은 여러 스레드 간 작업 순서와 역할을 제어하면서 스레드 간의 정보 전달을 가능하게 하는 데 유용하며, 이 과정에서 세마포어가 통신 도구처럼 사용된다.

 

 

그 이유는 세마포어는 공유 자원의 접근을 조절하고, 대기와 허용을 통해 스레드 간 동기화를 제공하므로 마치 스레드 간에 정보를 전달하고 통신하는 것처럼 보이기도 하기 때문이다.

 

 

Inter Thread - await(), signal()

 

 

만약 ThreadA가 세마포어를 획득할려고 한다면, ThreadA가 확인하는 것은 사용 가능한 권한(세마포어)의 수가 0보다 큰지 확인할 것이다. 만약 사용 가능한 권한이 0이라면 ThreadA는 대기 상태에 빠지게 될 것이다.

 

 

이후 ThreadB가 세마포어를 릴리즈한다면, 자연스럽게 권한을 획득한 ThreadA가 깨어나 작업을 수행할 것이다.

 

 

이 과정을 위의 이미지와 함께 본다면 ThreadB의 행위가 마치 ThreadA를 깨우려고 한 신호처럼 되는 것이다.

 

 

이게 스레드 간 통신의 기초이다.

 

 

이러한 방법 뿐만 아니라 Condition 변수를 사용한 방식 또한 존재한다.

 

 

보다시피 흐름은 위의 세마포어를 사용한 방식과 동일하다. 그 대신, 여기선 condition의 await()와 signal() 메서드가 사용된다.

 

 

await()

 

작업을 계속해서 진행하기 위한 조건을 만족하지 못한 경우 현재 가지고있는 락을 반납하고 대기상태에 들어간다.

 

 

signal()

 

await() 메서드로 인해 대기중인 스레드 하나를 깨우는데 사용된다.

보통 await()된 메서드의 조건을 충족할 수 있도록 만든 뒤 signal() 을 날리는 편이다.

 

 

awaitNanoes(nanosTimeout), await(long time, TimeUnit unit), ..

 

파라미터로 입력된 시간만큼 대기한다.

 

 

signalAll()

 

await() 중인 모든 스레드를 깨운다.

 

 

Condition 예제 코드

public class ProducerConsumerExample {

    private static final int BUFFER_SIZE = 5;
    private final Queue<Integer> buffer = new LinkedList<>();
    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();    // 버퍼가 가득 차지 않았음을 나타내는 condition
    private final Condition notEmpty = lock.newCondition();   // 버퍼가 비어있지 않음을 나타내는 condition

    public void produce(int value) throws InterruptedException {
        lock.lock();
        try {
            while (buffer.size() == BUFFER_SIZE) {    // 조건 : 버퍼 가득 참 유무
                System.out.println("버퍼가 가득찼습니다. 생산자 대기중...");
                notFull.await();
            }
            buffer.add(value);
            System.out.println("Produced: " + value);
            notEmpty.signal();   // 소비자에게 버퍼에 데이터가 있다고 신호
        } finally {
            lock.unlock();
        }
    }

    public void consume() throws InterruptedException {
        lock.lock();
        try {
            while (buffer.isEmpty()) {   // 조건 : 버퍼 비어있음 유무
                System.out.println("버퍼가 비어있습니다. 소비자 대기중...");
                notEmpty.await();
            }
            int value = buffer.poll();
            System.out.println("Consumed: " + value);
            notFull.signal();  // 생산자에게 버퍼에 빈 공간이 있다고 신호
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        ProducerConsumerExample example = new ProducerConsumerExample();

        Thread producer = new Thread(() -> {
            int value = 0;
            while (true) {
                try {
                    example.produce(value++);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        Thread consumer = new Thread(() -> {
            while (true) {
                try {
                    example.consume();
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        producer.start();
        consumer.start();
    }
}

 

 

 

Inter Thread - wait(), notify()

 

 

wait()과 notify() 모두 위의 await(), signal()과 비슷한 기능을 가진 메서드이다.

그 대신 wait()과 notify() 은 반드시 동기화된 블록이나 메서드 내부에서만 사용할 수 있기 때문에 synchronized 키워드를 사용하는 객체에서 스레드의 대기와 신호를 관리할 때 사용된다.

 

 

두 메서드가 동기화된 상태에서만 사용되는 이유는 스레드 간의 혼란을 방지하기 위해서이다.

 

  1. 락 관리 : wait()을 호출하면 현재 스레드는 락을 해제하고 대기 상태로 들어간다. 이때 다른 스레드가 이 락을 획득할 수 있게 된다. 만약 wait()이 동기화되지 않은 상태에서 호출되면, 스레드는 락을 가지고 있지 않기 때문에 예외가 발생한다.
  2. 상태 일관성 : notify()를 호출하는 스레드가 wait()로 대기 중인 스레드에게 신호를 줄 때, 둘 다 같은 락을 사용해야만 서로의 상태를 제대로 인식할 수 있다. 예를 들어, 어떤 스레드가 데이터가 준비됐다고 notify()를 호출했을 때, 대기 중인 스레드가 이 데이터를 읽을 수 있는지 확인하기 위해서는 같은 락을 공유해야 한다.

 

import java.util.LinkedList;
import java.util.Queue;

public class ProducerConsumerExample {
    private static final int BUFFER_SIZE = 5;
    private final Queue<Integer> buffer = new LinkedList<>();

    // synchronized 선언
    public synchronized void produce(int value) throws InterruptedException {
        while (buffer.size() == BUFFER_SIZE) {
            System.out.println("Buffer is full. Producer is waiting...");
            wait();
        }
        buffer.add(value);
        System.out.println("Produced: " + value);
        notify();
    }

    // synchronized 선언
    public synchronized void consume() throws InterruptedException { 
        while (buffer.isEmpty()) {
            System.out.println("Buffer is empty. Consumer is waiting...");
            wait(); 
        }
        int value = buffer.poll();
        System.out.println("Consumed: " + value);
        notify();
    }

    public static void main(String[] args) {
        // 위 예재 코드와 동일
        ...
    }
}

 

 

 

 

참조

https://www.udemy.com/course/java-multi-threading/?couponCode=24T6MT102824

'백엔드 > Java' 카테고리의 다른 글

[Java] 리플렉션  (0) 2024.11.07
[Java] 멀티스레딩 - 가상 스레드  (0) 2024.11.04
[Java] 멀티스레딩 - 락 심화  (0) 2024.11.04
[Java] 멀티스레딩 - 스레드간 데이터 공유  (0) 2024.10.31
[Java] 멀티스레딩 - 성능 최적화 (Feat : 스레드 풀)  (0) 2024.10.30

티스토리툴바