Thread in Java with Concurrency

허진혁·2023년 7월 18일

Java

기본기를 다지자

목록 보기

5/10

🤔 궁금사항

스레드는 왜 만들까?

자바는 어떻게 스레드를 활용할 수 있게 해주었을까?

동기화는 어떻게 처리할까?

Thread

Java에서 Thread는 하나의 프로세스 내에서 실행되는 하나의 실행 흐름을 나타내며, 동시성을 제공하기 위해 사용해요. 새로운 스레드를 만들기 위해서는 Thread 클래스를 상속하는 방식, Runnable 인터페이스를 구현하고 Thread 클래스의 생성자에 전달하는 방식으로 두 가지 방식이 있어요.

class ThreadA extends Thread {
	// ...
}

class RunnableB implements Runnable {
	@override
	public void run() {
		// ...
	}
}

class Main {
	public static void main(String[] args) {
		ThreadA threadA = new ThreadA();
		RunnableB threadB = new RunnableB();
		
		threadA.strat();
		new Thread(threadB).start();
		
	}
}

왜 두가지 방식을 지원했을까?

우선 두 방식 모드 스레드가 start() 메서드를 실행하는 방식임을 확인할 수 있어요. 그러면 왜 생성하는 방식으 ㄹ두가지로 분류했을까요?

이 부분을 고민하다보니 상속과 인터페이스의 차이점을 생각해보니 알 수 있었어요.

자바에서 다중 상속은 불가능하다.
인터페이스는 여러 개의 인터페이스를 상속이 가능하다.

스레드 클래스가 다른 클래스로 확장할 필요가 있을 경우 Runnable 인터페이스를 implements 하면 되고, 그렇지 않은 경우에 Thread를 상속받아 사용해요.

한 가지 의문이 들었어요. Runnable 인터페이스를 보면

다음과 같이 run() 메서드를 구현하도록 되어 있어요. 하지만, 스레드를 실행시킨 것은 start()에요.

이것 뿐만 아니라 스레드에서 활용하는 메서드들을 알아볼게요.

Method In Thread

🏃‍♀️ run() / strat()

start() 메서드를 호출하면 JVM은 새로운 스레드를 생성하고 해당 스레드에게 작업을 위임합니다. 단, 실행 순서를 보장하지는 않습니다.

run() 메서드는 스레드가 실행될 때 내부적으로 호출하는 메서드로, 직접 호출하더라도 스레드가 생성되지는 않습니다.

🥸 스레드는 왜 실행 순서를 보장하지 않을까요?

😪 sleep()

Thread 클래스에 static 메소드가 많이 있어요. 이는 해당 스레드를 위해 존재하는 것이 아니라, JVM에 있는 스레드를 관리하기 위한 용도가 많아요. 그리고 그 중 하나가 sleep() 메서드에요.

JVM은 주어진 스레드가 끝날 때 까지 기다리는 특성을 갖고 있어요. 일반적으로 메인 스레드가 작업을 끝낼 때 까지 JVM은 종료되지 않아요.(demon 스레드 때문에 항상은 아니에요.) 스레드가 종료되지 않으면 JVM이 끝나지 않게 되고 프로그램이 완전히 종료 되지 않아요.

만약 스레드를 기다리지 않고 JVM이 종료된다면, 실행 중인 스레드는 강제로 중단되고, 실행 중인 작업이 완료되지 않을 수 있어요. 이는 예기치 않은 동작이 발생할 수 있으며, 데이터의 일관성이 깨질 수도 있습니다. 따라서, 스레드를 사용하는 경우에는 스레드가 정상적으로 종료될 수 있도록 설계해야 하는 것이 중요해요.

sleep() 메서드는 주어진 시간 동안 스레드를 일시적으로 중지시키는 역할을 하며, 다른 스레드들의 실행에 영향을 주지 않아요.

Thread.sleep() 메서드를 사용할 때는 try-catch 구문을 사용해야 해요. sleep() 메서드는 InterruptedException 메서드를 던질 수 있기 때문이에요.

🥸 위에서 언급한 demon 스레드는 무엇이고, 왜 사용할까요?

데몬 스레드는 백그라운드 작업이나 서비스를 제공하는 역할을 하는 스레드로 일반 스레드에 비해 우선수위가 낮아요. 그리고 주 스레드가 종료되면 함께 종료되는 특성을 갖고 있어요.

데몬 스레드를 통해 메인 애플리케이션 종료시 자원 정리나 부가적인 작업을 자동으로 처리할 수 있어요.

예를 들어, 모니터링 하는 스레드를 데몬 스레드로 지정한다고 가정해보아요. 주 스레드가 끝난 후에 모니터링 스레드가 종료되요. 만약 주 스레드 끝나기 전에 모니터링 스레드가 종료된다고 하면 이는 제 역할을 다하지 못하기에 데몬으로 지정해 두는 거에요.

데몬스레드의 특징은 JVM과 별개로 종료된 다는 점이에요. 데몬스레드가 실행 유무와 상관없이 JVM은 끝날 수 있어요.

❗️ 주의할점

해당 스레드가 시작 되기 전에 데몬 스레드로 지정해야 해요. 실행 도중에 데몬 스레드로 지정할 수 없어요.

🫂 join()

join() 메서드는 현재 실행 중인 스레드가 다른 스레드가 종료될 때까지 기다리도록 하는 역할을 해요.

스레드가 종료될 때 까지 기다리는 메서드로, (long) mills 파라미터를 통해 특정 시간만큼 기다리게 할 수 있어요.

👀 interupt()

현재 수행중인 스레드를 중단시키는 메서드에요. 다만, 스레드를 강제로 중지시키는 것이 아니라, sleep() 메서드나 join() 메서드 등이 호출되어 대기 상태에 있는 스레드를 깨우는 역할을 해요. 이를 통해 스레드가 대기 상태에서 빠져나와 실행을 계속할 수 있도록 도와줘요.

interrupt() 메서드를 호출하면, 해당 스레드에게 InterruptedExcetpion 예외를 발생시켜요. 스레드는 이 예외를 처리하거나 전파하는 방식으로 중단 상태를 처리해요.

보통 대기상태를 만드는 메서드가 호출될 때 interrupt() 메서드가 사용 되고, 스레드 시작 전이나 종료된 상태에서는 예외나 에러 없이 다음 코드로 넘어가요.

Object 클래스에 선언된 스레드 관련 메소드

Thread에 구현되어 있지 않지만 Object 클래스에서 Thread를 다루기 위해 내장하고 있는 메서드에요.

wait(long timeout) → 다른 스레드가 Object 객체에 대해 notify() or notifyAll() 할 때 까지 스레드를 대기하거나 파라미터에 지정한 시간만큼 대기한다.
notify() → Obejct 객체의 모니터에 대기하고 있는 단일 스레드를 깨운다.
notifyAll() → Object 객체의 모니터에 대기하고 있는 모든 스레드를 깨운다.

스레드의 특성

스레드 간의 협력 방식은 스핀락이 아닌 wait 방식이다.
스레드는 실행 순서에 의존하지 않는다.
스레드가 공유하는 리소스는 heap 메모리이며, 스레드마다 stack 메모리는 고유하다.

스레드를 사용할 때 주의할 점 - Synchronization

스레드의 특징은 순서를 보장하지 않아요. 그렇기에 우리가 예측했던 값과 다른 값이 나올 때가 있어요.

이러한 문제를 동시성 문제라고 해요. 자바는 동시성 문제를 해결하기 위해 Mutex, Semaphore를 지원해줘요.

핵심 단어 정리

경쟁 조건(race condition): 공유 자원을 두 개 이상의 스레드가 동시에 접근하는 시나리오

임계 영역(critical section): 공유 자원을 접근하는 코드(블럭)

Mutex

뮤텍스는 경쟁 조건을 피하기 위해 임계 영역에 하나의 스레드만 접근 가능하게 하는 것으로 상호 배제 특징을 갖고 있어요.

뮤텍스는 임계 영역 앞에 뮤텍스(lock)를 설정하고 공유 자원에 대한 접근을 흭득하면 작업을 수행하고 완료되면 뮤텍스(lock) 객체를 반납(해제)해요. 중요한 것은 뮤텍스가 해제될 때 까지 다른 스레드들은 기다리는 점이에요.

이 방식의 대표적인 예로 synchronized가 있고, ReentrantLock이 있어요.

synchronized

synchronized는 자바의 예약어로 Thread-safe를 보장해줘요.
(Thread-safe란 여러 스레드로부터 안전하게 동시에 접근 가능한 상태나 객체를 의미해요.)

하나의 공유 데이터를 동시에 접근할 때 문제가 생긴다는 것은 변경을 갖고 있는 메서드가 인스턴스 변수를 수정하려고 할 때 발생하는 것을 의미해요.

즉, 특정 블록이나 메서드를 임계영역으로 지정하여 해당 영역이 해제될 때 까지 다른 스레드의 접근을 막아줘요.

synchronized를 사용하는 방식은 두 가지 있어요.

메서드 자체에 synchronized를 선언(synchronized method)
메소드 내의 특정 문장만 synchronized로 감싸는 방법(synchronized statements)

public synchronized void plus(int num) {
		amount += num;
}

public void plus(int num) {
   synchronized (this) {
	      amount += num;
   }
}

synchronized (this) 부분에 this는 잠금 처리를 위한 객체에요.

❗️ 생각해야 하는 부분

synchronized는 여러 스레드가 특정 객체의 있는 인스턴스 변수에 동시에 접근할 때 발생하는 문제를 해결하기 위한 것이에요. 각각 다른 객체의 인스턴스 변수(같은 이름의 변수라도)에 접근할 때는 의미가 없어요.

ReentrantLock

ReentrantLock은 java 1.5에서 도입되었고, synchronized보다 유연하고 제어 기능을 제공해줘요.

다음과 같은 이점을 얻을 수 있어요.

공정한 락 획득을 지원해요. 여러 스레드가 락을 요청했을 때, 먼저 요청한 스레드부터 차례대로 락을 주어 공정한 경쟁을 유지하고, 스레드의 실행 순서를 제어할 수 있어요.

인터럽트 가능한 락 획득을 지원해요. 스레드가 락을 흭득하기 위해 대기 중일 때, 다른 스레드가 해당 스레드를 인터럽트하면 락 흭득 대기상태에서 빠져나올 수 있어요. 이를 통해 스레드의 중단을 관리하고 조절할 수 있어요.
(자바에서 lock.lockInterruptibly() 은 락을 획득하려고 할 때 다른 스레드가 해당 스레드를 인터럽트할 수 있습니다.)

인터럽트 락 코드

import java.util.concurrent.locks.ReentrantLock;

public class InterruptLockExample {
   private static ReentrantLock lock = new ReentrantLock();

   public static void main(String[] args) {
       Thread thread1 = new Thread(() -> {
           try {
               lock.lockInterruptibly(); // 인터럽트 가능한 락 획득 시도
               try {
                   // 락을 획득한 후 수행할 작업
                   System.out.println("Thread 1 acquired the lock");
                   Thread.sleep(2000);
               } finally {
                   lock.unlock(); // 락 해제
                   System.out.println("Thread 1 released the lock");
               }
           } catch (InterruptedException e) {
               // 인터럽트가 발생하여 락을 획득하지 못한 경우
               System.out.println("Thread 1 interrupted while acquiring the lock");
           }
       });

       Thread thread2 = new Thread(() -> {
           try {
               // 일정 시간 후 인터럽트를 발생시킴
               Thread.sleep(1000);
               thread1.interrupt(); // Thread 1을 인터럽트하여 락 획득 대기 상태에서 벗어나도록 함
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       });

       thread1.start();
       thread2.start();

               /* result
               Thread 1 acquired the lock
               Thread 1 released the lock
               Thread 1 interrupted while acquiring the lock
               */
   }
}

타임아웃 기능을 지원해요. ReentrantLock은 흭득을 시도한 후 일정 시간 내에 락을 흭득하지 못하면 흭득을 포기하는 방식이에요. 이를 통해 deadlock 상황을 방지하고, 일정 시간 이상 락을 대기하지 않도록 할 수 있어요.

조건 변수(Condition)를 지원해요. 조건 변수를 사용하여 스레드 간의 통신과 상호작용을 할 수 있어요. 조건 변수를 통해 스레드의 대기, 신호 전달, 상태 체크 등을 관리할 수 있어요.

조건 변수 지원 코드

import java.util.concurrent.locks.ReentrantLock;
   
public class InterruptLockExample {
       private static ReentrantLock lock = new ReentrantLock();

       public static void main(String[] args) {
           Thread thread1 = new Thread(() -> {
               try {
                   lock.lockInterruptibly(); // 인터럽트 가능한 락 획득 시도
                   try {
                       // 락을 획득한 후 수행할 작업
                       System.out.println("Thread 1 acquired the lock");
                       Thread.sleep(2000);
                   } finally {
                       lock.unlock(); // 락 해제
                       System.out.println("Thread 1 released the lock");
                   }
               } catch (InterruptedException e) {
                   // 인터럽트가 발생하여 락을 획득하지 못한 경우
                   System.out.println("Thread 1 interrupted while acquiring the lock");
               }
           });

           Thread thread2 = new Thread(() -> {
               try {
                   // 일정 시간 후 인터럽트를 발생시킴
                   Thread.sleep(1000);
                   thread1.interrupt(); // Thread 1을 인터럽트하여 락 획득 대기 상태에서 벗어나도록 함
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           });

           thread1.start();
           thread2.start();

                   /* result
                   Thread 1 acquired the lock
                   Thread 1 released the lock
                   Thread 1 interrupted while acquiring the lock
                   */
       }
   }
    ```
 </div>

 </details>

ReentrantReadWriteLock

ReentrantReadWriteLock은 ReentrantLock의 특징을 갖고 있으면서 락을 읽기와 쓰기를 분리했어요.

🥸 왜 읽기와 쓰기를 분리했을까요?

읽기 작업은 멀티 스레드 환경에서도 데이터의 일관성과 정확성을 유지할 수 있어요. 읽기라는 것은 자원에 변화를 주지 않기 때문이에요. 그래서 ‘읽기 작업은 여러 스레드가 접근할 수 있게 허용해 주자’는 취지에요.

읽기와 쓰기를 분리함으로써 다음과 같은 이점을 얻어요.

읽기 작업은 자원을 변경시키지 않으므로 다른 스레드와 충돌 없이 실행되어 동시에 여러 스레드에서 실행될 수 있다.
읽기 작업이 많고 쓰기 작업이 적은 경우 동시에 실행할 수 있으므로 전반적으로 성능이 향상된다.
읽기 작업은 멀티 스레드로 쓰기 작업은 단일 스레드로 실행된다는 것은 스레드 간의 충돌이 없어지고 데이터의 일관성을 강화시켜 스레드의 안정성을 유지해준다.

ReentrantReadWriteLock에서 쓰기 작업은 읽기 작업와 충돌을 최소화하기 위해 독점적으로 실행해요.

🥸 락을 읽기와 쓰기로 분리했는데, 두 작업이 동시에 들어오면 누구를 먼저 실행시킬까요?
쓰기 작업이 우선권이 부여되고, 읽기 작업은 쓰기 작업이 완료되야 실행되요. 이를 다른 말로 표현한다면ReentrantLock과 달리 공정하지 않아요 !

Semaphore

ReentrantLock과 마찬가지로 Semaphore도 java 1.5에서 도입되었어요.

세마포어는 뮤텍스와 다르게 스레드를 고정된(지정한) 양의 정수만큼 임계 영역에 접근할 수 있어요.

만약 임계 영역에 접근 가능한 스레드 수를 1로 지정한다면 뮤텍스와 같아지게 돼요.

세마포어는 카운팅 변수를 활용해요. 공유 리소스 접근 전에 세마포어를 두어요. 그리고 공유자원에 접근하기 전에 카운팅 개수를 확인하고, 이보다 적으면 작업을 진행하고, 크다면 다른 스레드가 공유 자원을 반납할 때 까지 대기해요. 접근 후에 작업을 완료하면 세마포어의 수를 감소시키는 방식이에요.

세마포어는 두 가지 원자 연산 wait, signal 방식을 사용해요.
(signal은 자바에서는 notify() 메서드에요.)

Lock 을 가진 스레드가 다른 스레드에 Lock 을 넘겨준 이후에 대기해야 한다면 wait() 메서드를 사용하면 돼요. 그리고 대기 중인 임의의 스레드를 깨우려면 notify() 메서드를 통해 깨울 수 있어요. 대기 중인 모든 스레드를 깨우려면 notifyAll() 메서드를 통해 깨울 수 있는데, 이 경우에는 하나의 스레드만 Lock 을 획득하고 나머지 스레드는 다시 대기 상태에 들어가게 돼요.

❗️주의할 점

세마포어의 동작은 데드락을 피하기 위해 올바른 방식으로 구현해야 해요.

다음과 같이 사용할 때 데드락이 발생할 수 있어요.

쓰레드 2개와 세마포어 2개가 있다.
쓰레드 A의 작업은 세마포어 A를 흭득하고, 세마포어 B를 흭득하는 것이며, finally 블록에는 세마포어 A를 해제 후 세마포어 B를 해제한다.
쓰레드 B의 작업은 세마포어 B를 흭득하고, 세마포어 A를 흭득하는 것이며, finally 블록에는 세마포어 B륵 해제 후 세마포어 A를 해제한다.

import java.util.concurrent.Semaphore;

public class DeadLockExample {

private static final Semaphore semaphoreA = new Semaphore(1);
      private static final Semaphore semaphoreB = new Semaphore(1);

      public static void main(String[] args) {
          Thread threadA = new Thread(new Runnable() {
              @Override
              public void run() {
                  try {
                      semaphoreA.acquire();
                      Thread.sleep(1000);
                      semaphoreB.acquire();

                      System.out.println("Thread A running!");
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  } finally {
                      semaphoreA.release();
                      System.out.println("semaphore A in Thread A is released");
                      semaphoreB.release();
                      System.out.println("semaphore B in Thread A is released");
                  }
              }
          });

          Thread threadB = new Thread(new Runnable() {
              @Override
              public void run() {
                  try {
                      semaphoreB.acquire();
                      Thread.sleep(1000);
                      semaphoreA.acquire();

                      System.out.println("Thread B running");
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  } finally {
                      semaphoreB.release();
                      System.out.println("semaphore B in Thread A is released");
                      semaphoreA.release();
                      System.out.println("semaphore A in Thread A is released");
                  }
              }
          });

          threadA.start();
          threadB.start();
      }

  }

데드락을 발생시키지 않으려면 두 가지 조건을 지켜야해요.

락을 얻는 순서가 같아야 한다.
락을 얻은 순서의 역순으로 해제해야 한다.

import java.util.concurrent.Semaphore;

public class NoDeadLockExample {

    private static final Semaphore semaphoreA = new Semaphore(1);
    private static final Semaphore semaphoreB = new Semaphore(1);

    public static void main(String[] args) {
        Thread threadA = new Thread(() -> {
            try {
                semaphoreA.acquire();
                Thread.sleep(1000);
                semaphoreB.acquire();

                System.out.println("Thread A running!");

                semaphoreB.release();
                System.out.println("semaphore B in Thread A is released");
                semaphoreA.release();
                System.out.println("semaphore A in Thread A is released");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread threadB = new Thread(() -> {
            try {
                semaphoreA.acquire();
                Thread.sleep(1000);
                semaphoreB.acquire();

                System.out.println("Thread B running");

                semaphoreB.release();
                System.out.println("semaphore B in Thread A is released");
                semaphoreA.release();
                System.out.println("semaphore A in Thread A is released");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadA.start();
        threadB.start();

				/*
				Thread A running!
				semaphore B in Thread A is released
				semaphore A in Thread A is released
				Thread B running
				semaphore B in Thread A is released
				semaphore A in Thread A is released
				*/
    }

}