java.util 패키지에 있는 컬렉션 프레임워크는 원자적 연산을 제공할까? 예를 들어 하나의 ArrayList 인스턴스에 여러 스레드가 동시에 접근해도 괜찮을까? 참고로 여러 스레드가 동시에 접근해도 괜찮은 경우를 스레드 세이프(Thread Safe)하다고 한다. 그렇다면 ArrayList는 스레드 세이프 할까?
컬렉션에 데이터를 추가하는 add() 메서드를 생각해보면, 단순히 컬렉션에 데이터를 하나 추가하는 것 뿐이다. 따라서 이것은 마치 연산이 하나만 있는 원자적인 연산처럼 느껴진다. 원자적 연산은 쪼갤 수 없기 때문에 멀티스레드 상황에 문제가 되지 않는다. 물론 멀티스레드는 중간에 스레드의 실행 순서가 변경될 수 있으므로 [A, B] 또는 [B, A]로 데이터의 저장 순서는 변경될 수 있지만 결과적으로 데이터는 모두 안전하게 저장될 것 같다.
하지만, 컬렉션 프레임워크가 제공하는 대부분의 연산은 원자적인 연산이 아니다.
컬렉션을 아주 간단하게 직접 만들어보자.
SimpleList
package thread.collections.simple.list;
public interface SimpleList {
int size();
void add(Object o);
Object get(int index);
}
직접 만들 컬렉션의 인터페이스이다.
크기 조회, 데이터 추가, 데이터 조회의 3가지 메서드만 가진다.
BasicList
package thread.collections.simple.list;
import java.util.Arrays;
import static util.ThreadUtils.sleep;
public class BasicList implements SimpleList {
private static final int DEFAULT_CAPACITY = 5;
private Object[] elements;
private int size = 0;
public BasicList() {
elements = new Object[DEFAULT_CAPACITY];
}
@Override
public int size() {
return size;
}
@Override
public void add(Object o) {
elements[size] = o;
sleep(100); // 멀티 스레드 문제를 쉽게 확인하는 코드
size++;
}
@Override
public Object get(int index) {
return elements[index];
}
@Override
public String toString() {
return Arrays.toString(Arrays.copyOf(elements, size)) + ", size = " + size + ", capacity = " + elements.length;
}
}
가장 간단한 컬렉션의 구현이다. 내부에서는 배열을 사용해서 데이터를 보관한다.
ArrayList의 최소 구현 버전이라 생각하면 된다.
DEFAULT_CAPACITY: 최대 5개의 데이터를 저장할 수 있다.
size: 저장한 데이터의 크기를 나타낸다.
add(): 컬렉션에 데이터를 추가한다.
sleep(100): 잠시 대기한다. 이렇게 하면 멀티스레드 상황에 발생하는 문제를 확인하기 쉽다.
일단은 단일 스레드로 잘 동작하는지 확인해보자.
SimpleListMainV1
package thread.collections.simple;
import thread.collections.simple.list.BasicList;
import thread.collections.simple.list.SimpleList;
public class SimpleListMainV1 {
public static void main(String[] args) {
SimpleList basicList = new BasicList();
basicList.add("A");
basicList.add("B");
System.out.println("basicList = " + basicList);
}
}
실행 결과
basicList = [A, B], size = 2, capacity = 5
단일 스레드로 실행했기 때문에 전혀 문제 없이 잘 동작한다. 이제 멀티 스레드로 이 자료구조에 데이터를 추가해보자!
SimpleListMainV2
package thread.collections.simple;
import thread.collections.simple.list.BasicList;
import thread.collections.simple.list.SimpleList;
import static util.MyLogger.log;
public class SimpleListMainV2 {
public static void main(String[] args) throws InterruptedException {
test(new BasicList());
}
private static void test(SimpleList list) throws InterruptedException {
log(list.getClass().getSimpleName());
Runnable addA = new Runnable() {
@Override
public void run() {
list.add("A");
log("Thread-1: list.add(A)");
}
};
Runnable addB = new Runnable() {
@Override
public void run() {
list.add("B");
log("Thread-2: list.add(B)");
}
};
Thread thread1 = new Thread(addA, "Thread-1");
Thread thread2 = new Thread(addB, "Thread-2");
thread1.start();
thread2.start();
thread1.join();
thread2.join();
log(list);
}
}
실행 결과를 보면, size는 2인데, 데이터는 B 하나만 입력되어 있다. 어떻게 된 것일까?
참고로, 어떤 스레드가 먼저 실행됐냐에 따라 [A, null]이 결과가 될 수 있다.
@Override
public void add(Object o) {
elements[size] = o;
sleep(100); // 멀티 스레드 문제를 쉽게 확인하는 코드
size++;
}
스레드1, 스레드2가 element[size] = o; 이 코드를 동시에 수행한다. 여기서는 스레드1이 약간 빠르게 수행했다.
스레드1 수행: element[0] = A, element[0]의 값은 A가 된다.
스레드2 수행: element[0] = B, element[0]의 값은 A → B가 된다.
결과적으로 element[0]의 값은 B가 된다.
스레드1, 스레드2가 sleep()에서 잠시 대기한다. 여기서 sleep()을 사용한 이유는 동시성 문제를 쉽게 확인하기 위해서다. 이 코드를 제거하면 size++이 너무 빨리 호출되기 때문에 스레드1이 add()메서드를 완전히 수행하고 나서 스레드2가 add()메서드를 수행할 가능성이 높다. 당연한 이야기지만 sleep() 코드를 제거해도 멀티스레드 동시성 문제는 여전히 발생하고 있다. (확률을 더 높였을 뿐이다)
결론은 무엇이냐면 컬렉션 프레임워크 대부분은 스레드 세이프 하지 않다는 것이다.
우리가 일반적으로 자주 사용하는 ArrayList, LinkedList, HashSet, HashMap 등 수 많은 자료 구조들은 단순한 연산을 제공하는 것처럼 보인다. 예를 들어, 데이터를 추가하는 add()와 같은 연산은 마치 원자적 연산처럼 느껴진다. 하지만 그 내부에서는 수 많은 연산들이 함께 사용된다. 배열에 데이터를 추가하고, 사이즈를 변경하고, 배열을 새로 만들어서 배열의 크기도 늘리고, 노드를 만들어서 링크에 연결하는 등 수 많은 복잡한 연산이 함께 사용된다.
따라서, 일반적인 컬렉션들은 절대로! 스레드 세이프 하지 않다!
단일 스레드가 컬렉션에 접근하는 경우라면 아무런 문제가 되지 않지만, 멀티 스레드 상황에서 여러 스레드가 동시에 컬렉션에 접근하는 경우라면 java.util 패키지가 제공하는 일반적인 컬렉션들은 사용하면 안된다. (물론 일부 예외도 있다. 뒤에서 알아보자.)
동시성 컬렉션이 필요한 이유
컬렉션이 수많은 복잡한 연산으로 이루어져 있기 때문이다. 따라서 여러 스레드가 접근해야 한다면 synchronized, Lock등을 통해 안전한 임계 영역을 적절히 만들면 문제를 해결할 수 있다.
SyncList
package thread.collections.simple.list;
import java.util.Arrays;
import static util.ThreadUtils.sleep;
public class SyncList implements SimpleList {
private static final int DEFAULT_CAPACITY = 5;
private Object[] elements;
private int size = 0;
public SyncList() {
elements = new Object[DEFAULT_CAPACITY];
}
@Override
public synchronized int size() {
return size;
}
@Override
public synchronized void add(Object o) {
elements[size] = o;
sleep(100); // 멀티 스레드 문제를 쉽게 확인하는 코드
size++;
}
@Override
public synchronized Object get(int index) {
return elements[index];
}
@Override
public synchronized String toString() {
return Arrays.toString(Arrays.copyOf(elements, size)) + ", size = " + size + ", capacity = " + elements.length;
}
}
앞서 만든 BasicList를 복사해서 만든 SyncList에 synchronized 키워드만 추가했다.
아주 아주 잘 실행됐다. 이제 동시에 여러 스레드가 접근하더라도 걱정없이 사용할 수 있다. 근데! 문제가 있다.
BasicList 코드가 있는데, 이 코드를 거의 그대로 복사해서 synchronized 기능만 추가한 SyncList를 만들었다. 하지만 이렇게 되면 모든 컬렉션을 다 복사해서 동기화 용으로 새로 구현해야 한다. 이게 매우 비효율적이다.
프록시 도입
위에서 말한 문제를 다시 상기해보면, 고작 synchronized 키워드 하나를 추가하기 위해 같은 코드를 복사해서 새로운 클래스를 만들어내야 한다는 점이다. 그럼 다른 자료구조를 사용한다고 하면 그것 역시 또 새로운 클래스를 만들어야 한다. 즉, 단일 스레드용 클래스와 멀티 스레드용 클래스가 나뉘어진다는 점이다. 다음과 같이 말이다.
ArrayList → SyncArrayList
LinkedList → SyncLinkedList
원하는건 기존 코드를 그대로 사용하되 synchronized 기능만 살짝 추가하고 싶은 것이다. 이럴때 프록시를 사용하면 좋다.
프록시(Proxy)
대리자, 대체자라는 뜻으로 스프링에서도 굉장히 자주 등장하고 많이 사용된다. 요청을 하는 클라이언트와 요청을 받는 서버가 원래는 이런 형태였다면,
클라이언트 → 서버
다음과 같은 형태로 변형되는 것이다.
클라이언트 → 프록시 → 서버
중요한건 이렇게 변경이 되어도 클라이언트 코드는 바꿀게 하나도 없다는 것. 이게 바로 핵심이다.
SyncProxyList
package thread.collections.simple.list;
public class SyncProxyList implements SimpleList {
private SimpleList target;
public SyncProxyList(SimpleList target) {
this.target = target;
}
@Override
public synchronized int size() {
return target.size();
}
@Override
public synchronized void add(Object o) {
target.add(o);
}
@Override
public synchronized Object get(int index) {
return target.get(index);
}
@Override
public String toString() {
return target.toString() + " by " + this.getClass().getSimpleName();
}
}
프록시 역할을 하는 클래스이다.
SyncProxyList는 BasicList와 같은 SimpleList 인터페이스를 구현한다.
이 클래스는 생성자를 통해 SimpleList target을 주입받는다. 여기에 실제 호출되는 대상이 들어간다.
이 클래스는 빈 껍데기다. 이 클래스의 역할은 모든 메서드에 synchronized를 걸어주는 일 뿐이다. 그리고나서 target에 있는 같은 기능을 호출한다.
이 프록시 클래스는 synchronized만 걸고, 그 다음에 바로 실제 호출해야 하는 원본 대상(target)을 호출한다.
그리고 다음과 같이 호출하는 test() 메서드에 파라미터로 SyncProxyList를 던져주면 된다.
test() 메서드를 클라이언트라고 가정하면 test()메서드는 SimpleList라는 인터페이스에만 의존한다. 이것을 추상화에 의존한다고 표현한다.
덕분에 SimpleList 인터페이스의 구현체인 BasicList, SyncList, SyncProxyList 중에 어떤 것을 사용하든 클라이언트인 test()의 코드는 전혀 변경하지 않아도 된다.
클라이언트인 test() 입장에서 생각해보면 BasicList가 넘어올지, SyncProxyList가 넘어올지 알 수 없다. 단순히 SimpleList의 구현체 중 하나가 넘어와서 실행된다는 정도만 알 수 있다. 그래서 클라이언트인 test()는 매우 유연하다. SimpleList의 어떤 구현체든지 다 받아들일 수 있다.
런타임 의존 관계 - BasicList
먼저 BasicList를 사용하는 예를 보자.
그림과 같이 실제 런타임에 발생하는 인스턴스의 의존 관계를 런타임 의존 관계라 한다. 먼저 간단한 BasicList를 직접 사용하는 경우부터 알아보자.
test() 메서드에서 스레드를 만들고, 스레드에 있는 run()에서 list.add()를 호출한다.
그림은 간단하게 test()에서 호출하는 것으로 표현하겠다.
BasicList(x001) 인스턴스에 있는 add()가 호출된다.
런타임 의존 관계 - SyncProxyList
이번엔 BasicList가 아니라 SyncProxyList를 사용하는 예를 보자.
test(new SyncProxyList(new BasicList()));
먼저 BasicList(x001) 인스턴스가 만들어진다.
앞서 만든 BasicList(x001)의 참조를 SyncProxyList의 생성자에 전달하여 SyncProxyList(x002)가 만들어진다.
내부에는 원본 대상을 가르키는 target 변수를 포함하고 있다. 이 변수는 BasicList(x001)의 참조를 보관한다.
test()메서드는 SyncProxyList(x002) 인스턴스를 사용하게 된다.
SyncProxyList - add() 호출 과정
test()메서드에서 스레드를 만들고, 스레드에 있는 run()에서 list.add()를 호출한다.
SyncProxyList(x002)에 있는 add()가 호출된다.
그림은 간단하게 test()에서 호출하는 것으로 표현하겠다.
프록시인 SyncProxyList는 synchronized를 적용한다. 그리고 나서 target에 있는 add()를 호출한다.
원본 대상인 BasicList(x001)의 add()가 호출된다.
원본 대상의 호출이 끝나면 결과를 반환한다.
SyncProxyList에 있는 add()로 흐름이 돌아온다. 메서드를 반환하면서 synchronized를 해제한다.
test()로 흐름이 돌아온다.
프록시 정리
프록시인 SyncProxyList는 원본인 BasicList와 똑같은 SimpleList를 구현한다. 따라서 클라이언트인 test() 입장에서는 원본 구현체가 전달되든, 아니면 프록시 구현체가 전달되든 아무런 상관이 없다. 단지 수많은 SimpleList의 구현체 중 하나가 전달되었다고 생각할 뿐이다.
클라이언트 입장에서 보면 프록시는 원본과 똑같이 생겼고, 호출할 메서드도 똑같다. 단지 SimpleList의 구현체일 뿐이다.
프록시는 내부에 원본을 가지고 있다. 그래서 프록시가 필요한 일부의 일을 처리하고, 그 다음에 원본을 호출하는 구조를 만들 수 있다. 여기서 프록시는 synchronized를 통한 동기화를 적용한다.
프록시가 동기화를 적용하고 원본을 호출하기 때문에 원본 코드도 이미 동기화가 적용된 상태로 호출된다.
여기서 핵심은 원본 코드인 BasicList를 전혀 손대지 않고 프록시인 SyncProxyList를 통해 동기화 기능을 적용했다는 점이다. 또한 이후에 SimpleList를 구현한 BasicLinkedList 같은 연결 리스트를 만들더라도 서로 같은 인터페이스를 사용하기 때문에 SyncProxyList를 그대로 활용할 수 있다. 쉽게 이야기해서 SyncProxyList 프록시 하나로 SimpleList 인터페이스의 모든 구현체를 동기화 할 수 있다.
이런 프록시를 사용하는 걸 프록시 패턴이라고 하고 정말 자주 종종 사용되는 패턴이다. 특히 스프링의 AOP는 프록시의 끝판왕으로 생각하면 된다. 프록시 패턴의 주요 목적은 다음과 같다.
접근 제어: 실제 객체에 대한 접근을 제한하거나 통제할 수 있다.
성능 향상: 실제 객체의 생성을 지연시키거나 캐싱하여 성능을 최적화할 수 있다.
부가 기능 제공: 실제 객체에 추가적인 기능(로깅, 인증, 동기화 등)을 투명하게 제공할 수 있다.
그래서 자바는 어떤 동시성 컬렉션을 제공해왔을까? 알아보자!
자바 동시성 컬렉션 - synchronized
자바가 제공하는 java.util 패키지에 있는 컬렉션 프레임워크들은 대부분 스레드 안전하지 않다. 우리가 일반적으로 사용하는 ArrayList, LinkedList, HashSet, HashMap 등 수 많은 자료 구조들은 내부에서 수많은 연산들이 함께 사용된다. 그렇다면 처음부터 모든 자료 구조에 synchronized를 사용해서 동기화를 해두면 어떨까? synchronized, Lock, CAS등 모든 방식은 정도의 차이가 있지만 성능과 트레이드 오프가 있다. 결국 동기화를 사용하지 않는 것이 가장 빠르다.
그리고 컬렉션이 항상 멀티스레드에서 사용되는것도 아니다. 미리 동기화를 해둔다면 단일 스레드에서 사용할 때 동기화로 인해 성능이 저하된다. 따라서 동기화의 필요성을 정확히 판단하고 꼭 필요한 경우에만 동기화를 적용하는 것이 필요하다.
좋은 대안으로는 우리가 앞서 배운 것처럼 synchronized를 대신 적용해 주는 프록시를 만드는 방법이 있다. List, Set, Map 등 주요 인터페이스를 구현해서 synchronized를 적용할 수 있는 프록시를 만들면 된다. 이 방법을 사용하면 기존 코드를 유지하면서 필요한 경우에만 동기화를 적용할 수 있다.
자바는 컬렉션을 위한 프록시 기능을 제공한다.
SynchronizedListMain
package thread.collections.java;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
public class SynchronizedListMain {
public static void main(String[] args) {
List<String> list = Collections.synchronizedList(new ArrayList<>());
list.add("data1");
list.add("data2");
list.add("data3");
System.out.println(list.getClass());
System.out.println("list = " + list);
}
}
실행 결과
class java.util.Collections$SynchronizedRandomAccessList
list = [data1, data2, data3]
위에서 프록시를 직접 만들어 본 것처럼 이것도 역시 파리미터로 건네주는 자료 구조를 동기화 해주는 프록시 클래스이다.
예를 들어 이 클래스의 add() 메서드를 보면, synchronized 블록을 적용하고 그 다음에 원본 대상의 add()를 호출하는 것을 알 수 있다.
Collections는 다음과 같이 다양한 synchronized 동기화 메서드를 지원한다. 이 메서드를 사용하면 List, Collection, Map, Set 등 다양한 동기화 프록시를 만들 수 있다.
synchronizedList()
synchronizedCollection()
synchronizedMap()
synchronizedSet()
synchronizedNavigableMap()
synchronizedNavigableSet()
synchronizedSortedMap()
synchronizedSortedSet()
Collections가 제공하는 동기화 프록시 기능 덕분에 스레드 안전하지 않은 수많은 컬렉션들을 매우 편리하게 스레드 안전한 컬렉션으로 변경해서 사용할 수 있다.
synchronized 프록시 방식의 단점
하지만 synchronized 프록시를 사용하는 방식은 다음과 같은 단점이 있다.
첫째, 동기화 오버헤드가 발생한다. 비록 synchronized 키워드가 멀티 스레드 환경에서 안전한 접근을 보장하지만, 각 메서드 호출 시마다 동기화 비용이 추가된다. 이로 인해 성능 저하가 발생할 수 있다.
둘째, 전체 컬렉션에 대해 동기화가 이루어지기 때문에, 잠금 범위가 넓어질 수 있다. 이는 잠금 경합(lock contention)을 증가시키고, 병렬 처리의 효율성을 저하시키는 요인이 된다. 모든 메서드에 대해 동기화를 적용하다 보면, 특정 스레드가 컬렉션을 사용하고 있을 때 다른 스레드들이 대기해야 하는 상황이 빈번해질 수 있다.
셋째, 정교한 동기화가 불가능하다. synchronized 프록시를 적용하면 컬렉션 전체에 대한 동기화가 이루어지지만, 특정 부분이나 메서드에 대해 선택적으로 동기화를 적용하는 것은 어렵다. 이는 과도한 동기화로 이어질 수 있다.
쉽게 이야기해서 이 방식은 단순 무식하게 모든 메서드에 synchronized를 걸어버리는 것이다. 따라서 동기화에 대한 최적화가 이루어지지 않는다. 자바는 이런 단점을 보완하기 위해 java.util.concurrent 패키지에 동시성 컬렉션을 제공한다.
자바가 제공하는 동시성 컬렉션
위 자바가 제공하는 Collections.synchronizedXxx() 프록시 방식은 여러 스레드가 동시에 접근해도 동시성 문제가 발생하지 않도록 해준다. 그래서 안전하게 멀티 스레드 환경에서 사용할 수 있다. 그러나, 단점이 있는데 모든 메서드가 다 synchronized가 걸려있어서 필요없을 때 조차도 병렬 처리가 불가능하고 한마디로 무겁고 비용이 많이 든다.
그래서 java.util.concurrent 패키지에는 고성능 멀티 스레드 환경을 지원하는 다양한 동시성 컬렉션 클래스들을 제공한다. 예를 들어, ConcurrentHashMap, CopyOnWriteArrayList, BlockingQueue 등이 있다. 이 컬렉션들은 더 정교한 잠금 메커니즘을 사용하여 동시 접근을 효율적으로 처리하며, 필요한 경우 일부 메서드에 대해서만 동기화를 적용하는 등 유연한 동기화 전략을 제공한다.
여기에 다양한 성능 최적화 기법들이 적용되어 있는데, synchronized, Lock(ReentrantLock), CAS, 분할 잠금 기술(segment lock)등 다양한 방법을 섞어서 매우 정교한 동기화를 구현하면서 동시에 성능도 최적화했다. 각각의 최적화는 매우 어렵게 구현되어 있기 때문에, 자세히 구현을 이해하는 것보다는 멀티스레드 환경에 필요한 동시성 컬렉션들을 잘 선택해서 사용할 수 있으면 충분하다.
동시성 컬렉션의 종류
List
CopyOnWriteArrayList → ArrayList의 대안
Set
CopyOnWriteArraySet → HashSet의 대안
ConcurrentSkipListSet → TreeSet의 대안(정렬된 순서 유지, Comparator 사용 가능)
Map
ConcurrentHashMap → HashMap의 대안
ConcurrentSkipListMap → TreeMap의 대안(정렬된 순서 유지, Comparator 사용 가능)
Queue
ConcurrentLinkedQueue: 동시성 큐, 비 차단 큐이다.
Deque
ConcurrentLinkedDeque: 동시성 데크, 비 차단 큐이다.
참고로, LinkedHashSet, LinkedHashMap 처럼 입력 순서를 유지하는 동시에 멀티 스레드 환경에서 사용할 수 있는 Set, Map 구현체는 제공하지 않는다. 필요하다면 Collections.synchronizedXxx()를 사용해야 한다.
스레드를 차단하는 블로킹 큐도 알아보자.
BlockingQueue
ArrayBlockingQueue
크기가 고정된 블로킹 큐
공정(fair) 모드를 사용할 수 있다. 공정 모드를 사용하면 성능이 저하될 수 있다.
LinkedBlockingQueue
크기가 무한하거나 고정된 블로킹 큐
PriorityBlockingQueue
우선순위가 높은 요소를 먼저 처리하는 블로킹 큐
SynchronousQueue
데이터를 저장하지 않는 블로킹 큐로, 생산자가 데이터를 추가하면 소비자가 그 데이터를 받을 때까지 대기한다. 생산자 - 소비자 간 직접적인 핸드오프 메커니즘을 제공한다. 쉽게 이야기해서 중간에 큐 없이 생산자 - 소비자가 직접 거래한다.
DelayQueue
지연된 요소를 처리하는 블로킹 큐로, 각 요소는 지정된 지연 시간이 지난 후에야 소비될 수 있다. 일정 시간이 지난 후 작업을 처리해야 하는 스케쥴링 작업에 사용된다.
ListMain (List 예시)
package thread.collections.java;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
public class ListMain {
public static void main(String[] args) {
List<String> list = new CopyOnWriteArrayList<>();
list.add("a");
list.add("b");
System.out.println("list = " + list);
}
}
실행 결과
list = [a, b]
물론, 지금 실행 결과는 단일 스레드의 실행 결과이지만 이 CopyOnWriteArrayList는 ArrayList에 대한 동시성 접근을 잘 처리한 자료 구조이다. 지금처럼 이렇게 자바가 잘 만들어놓은 자료 구조를 사용하면 이렇게 list.add("a"), list.add("b") 와 같은 메서드만 있다면 굳이 synchronized 같은 동기화 기법을 사용할 필요가 없다. 내부적으로 동기화 기법이 잘 적용된 상태니까.
그러니까 쉽게 말해서, 자료 구조에 대한 작업을 위해서는 동기화 작업을 직접적으로 개발자가 따로 걸어줄 필요가 없다는 소리다. 당연히 그게 아니라 로직상에 원자적 연산이 아닌 코드가 있다면 그 부분에 대해서는 개발자가 직접 동기화 작업을 위한 코드를 작성해야 겠지만 위 코드처럼 이미 아주 효율적으로 동기화 작업이 되어 있는 자료 구조에 데이터를 추가하고 뭐 하고 하는 부분만 있으면 동기화 작업이 따로 필요 없다.
SetMain (Set 예시)
package thread.collections.java;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArraySet;
public class SetMain {
public static void main(String[] args) {
Set<Integer> copySet = new CopyOnWriteArraySet<>();
copySet.add(1);
copySet.add(2);
copySet.add(3);
System.out.println("copySet = " + copySet);
ConcurrentSkipListSet<Object> skipSet = new ConcurrentSkipListSet<>();
skipSet.add(2);
skipSet.add(1);
skipSet.add(3);
System.out.println("skipSet = " + skipSet);
}
}
실행 결과
copySet = [1, 2, 3]
skipSet = [1, 2, 3]
CopyOnWriteArraySet은 HashSet의 대안이다.
ConcurrentSkipListSet은 TreeSet의 대안이다. 데이터의 정렬 순서를 유지한다. (Comparator 사용 가능)
ConcurrentSkipListMap은 TreeMap의 대안이다. 데이터의 정렬 순서를 유지한다. (Comparator 사용 가능)
정리
자바가 제공하는 동시성 컬렉션은 멀티 스레드 상황에 최적의 성능을 낼 수 있도록 다양한 최적화 기법이 적용되어 있다. 따라서 Collections.synchronizedXxx를 사용하는 것보다 더 좋은 성능을 제공한다. 당연한 이야기지만 동시성은 결국 성능과 트레이드 오프가 있다. 따라서 단일 스레드가 컬렉션을 사용하는 경우에는 동시성 컬렉션이 아닌 일반 컬렉션을 사용해야 한다.
반대로 멀티 스레드 상황에서 일반 컬렉션을 사용하면 정말 해결하기 어려운 버그를 만날 수 있다. 세상에서 가장 해결하기 어려운 버그가 멀티스레드로 인해 발생한 버그이다. 이러한 이유로 멀티스레드 환경에서는 동시성 컬렉션을 적절히 활용해서 버그를 예방하고 성능을 최적화하는 것이 중요하다. 동시성 컬렉션을 사용하면 코드의 안정성과 효율성을 높일 수 있으며, 예상치 못한 동시성 문제도 방지할 수 있다.
컴퓨터 과학에서 사용하는 원자적 연산(atomic operation)의 의미는 해당 연산이 더 이상 나눌 수 없는 단위로 수행된다는 것을 의미한다. 즉, 원자적 연산은 중단되지 않고 다른 연산과 간섭 없이 완전히 실행되거나 전혀 실행되지 않는 성질을 가지고 있다. 쉽게 이야기해서 멀티 스레드 환경에서 다른 스레드의 간섭 없이 안전하게 처리되는 연산이라는 뜻이다.
예를 들어 다음과 같은 필드가 있을 때,
volatile int i = 0;
다음 연산은 둘로 쪼갤 수 없는 원자적 연산이다.
i = 1
왜냐하면 이 연산은 다음 단 하나의 순서로 실행되기 때문이다.
오른쪽에 있는 1의 값을 왼쪽의 i변수에 대입한다.
하지만 다음 연산은 원자적 연산이 아니다.
i = i + 1
왜냐하면 이 연산은 다음 순서로 나누어 실행되기 때문이다.
오른쪽에 있는 i의 값을 읽는다. (i의 값을 10이라고 가정)
읽은 값에 1을 더해서 11을 만든다.
더한 11을 왼쪽 i 변수에 대입한다.
원자적 연산은 멀티 스레드 상황에서 아무런 문제가 발생하지 않는다. 하지만 원자적 연산이 아닌 경우에는 synchronized 블록이나 Lock등을 사용해서 안전한 임계 영역을 만들어야 한다.
원자적 연산 시작
원자적이지 않은 연산을 멀티스레드 환경에서 실행하면 어떤 문제가 발생하는지 코드로 알아보자.
IncrementInteger는 숫자 값을 하나씩 증가시키는 기능을 제공한다. 예를 들어서 지금까지 접속한 사용자의 수 등을 계산할 때 사용할 수 있다.
IncrementInteger
package thread.cas.increment;
public interface IncrementInteger {
void increment();
int get();
}
IncrementInteger는 값을 증가하는 기능을 가진 숫자 기능을 제공하는 인터페이스이다.
increment(): 값을 하나 증가
get(): 값을 조회
BasicInteger
package thread.cas.increment;
public class BasicInteger implements IncrementInteger {
private int value;
@Override
public void increment() {
value++;
}
@Override
public int get() {
return value;
}
}
IncrementInteger 인터페이스의 가장 기본 구현이다.
increment()를 호출하면 value++를 통해서 값을 하나 증가한다.
value 값은 인스턴스의 필드이기 때문에, 여러 스레드가 공유할 수 있다. 이렇게 공유 가능한 자원에 ++와 같은 원자적이지 않은 연산을 사용하면 멀티스레드 상황에 문제가 발생할 수 있다.
IncrementThreadMain
package thread.cas.increment;
import java.util.ArrayList;
import java.util.List;
import static util.ThreadUtils.sleep;
public class IncrementThreadMain {
public static final int THREAD_COUNT = 1000;
public static void main(String[] args) throws InterruptedException {
test(new BasicInteger());
}
private static void test(IncrementInteger incrementInteger) throws InterruptedException {
Runnable runnable = new Runnable() {
@Override
public void run() {
sleep(10);
incrementInteger.increment();
}
};
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < THREAD_COUNT; i++) {
Thread thread = new Thread(runnable);
threads.add(thread);
thread.start();
}
for (Thread thread : threads) {
thread.join();
}
int result = incrementInteger.get();
System.out.println(incrementInteger.getClass().getSimpleName() + " result: " + result);
}
}
THREAD_COUNT 수 만큼 스레드를 생성하고 incrementInteger.increment()를 호출한다.
스레드를 1000개 생성했다면, increment() 메서드도 1000번 호출하기 때문에 결과는 1000이 되어야 한다.
참고로 스레드가 너무 빨리 실행되기 때문에 여러 스레드가 동시에 실행되는 상황을 확인하기 어렵다. 그래서 run() 메서드에 sleep(10)을 두어, 최대한 많은 스레드가 동시에 increment()를 호출하도록 한다.
실행결과
BasicInteger result: 950
실행결과를 보면 기대한 1000이 아니라 다른 숫자가 보인다. 이 문제는 앞서 설명한 것처럼 여러 스레드가 동시에 원자적이지 않은 value++을 호출했기 때문이다. 물론 멀티 스레드 환경에서는 공유 자원에 여러 스레드가 아무런 안전 장치 없이 자원에 쓰기 작업을 하면 문제가 발생하는것을 이제는 너무 잘 알지만 원자적 연산 관점으로 한번 생각을 해보자. 결국 공유 가능한 자원에 원자적이지 않은 연산을 하면 멀티스레드 환경에선 문제가 될 수 있다가 핵심이다!
참고로, value++은 value = value + 1;이다.
그럼 volatile, synchronized를 적용해보면 어떻게 나올지 결과를 보자!
VolatileInteger
package thread.cas.increment;
public class VolatileInteger implements IncrementInteger {
volatile private int value;
@Override
public void increment() {
value++;
}
@Override
public int get() {
return value;
}
}
SyncInteger
package thread.cas.increment;
public class SyncInteger implements IncrementInteger {
private int value;
@Override
public synchronized void increment() {
value++;
}
@Override
public synchronized int get() {
return value;
}
}
이 결과도 예측 가능한 결과였다. 이젠 volatile은 동시성 문제에 아무런 해결 방안이 되지 않는다는 것을 알고 있기 때문에. 그래서 synchronized 블록을 사용했을 때 드디어 원하는 결과가 나왔다. 근데 이럴때 그냥 원자적 연산을 가능하게 해주는 기능이 따로 있으면 편하지 않을까?
AtomicInteger
이거 얘기하려고 이만큼 빌드업했다..! 자바는 앞서 만든 SyncInteger와 같이 멀티 스레드 환경에서 안전하게 증가 연산을 수행할 수 있는 AtomicInteger라는 클래스를 제공한다. 이름 그대로 원자적인 Integer라는 뜻이다. 다음과 같이 MyAtomicInteger 클래스를 만들고, 자바가 제공하는 AtomicInteger를 사용해보자.
MyAtomicInteger
package thread.cas.increment;
import java.util.concurrent.atomic.AtomicInteger;
public class MyAtomicInteger implements IncrementInteger {
AtomicInteger atomicInteger = new AtomicInteger(0);
@Override
public void increment() {
atomicInteger.incrementAndGet();
}
@Override
public int get() {
return atomicInteger.get();
}
}
실행 결과를 보면 AtomicInteger를 사용하면 이 결과 역시 1000이 잘 찍힌것을 알 수 있다. 1000개의 스레드가 안전하게 증가 연산을 수행한 것이다. AtomicInteger는 멀티스레드 상황에 안전하고 또 다양한 값 증가, 감소 연산을 제공한다. 특정 값을 증가하거나 감소해야 하는데 여러 스레드가 해당 값을 공유해야 한다면, AtomicInteger를 사용하면 된다.
참고로, AtomicInteger, AtomicLong, AtomicBoolean 등 다양한 AtomicXXX 클래스가 존재한다.
원자적 연산 성능 테스트
AtomicInteger의 비밀을 하나씩 파헤쳐보자. 우선 한번 지금까지 만든 클래스들의 성능을 비교해보자.
IncrementPerformanceMain
package thread.cas.increment;
public class IncrementPerformanceMain {
public static final long COUNT = 100_000_000_0;
public static void main(String[] args) {
test(new BasicInteger());
test(new VolatileInteger());
test(new SyncInteger());
test(new MyAtomicInteger());
}
private static void test(IncrementInteger incrementInteger) {
long startMs = System.currentTimeMillis();
for (int i = 0; i < COUNT; i++) {
incrementInteger.increment();
}
long endMs = System.currentTimeMillis();
System.out.println(incrementInteger.getClass().getSimpleName() + ": ms= " + (endMs - startMs));
}
}
안전한 임계 영역도 없고, volatile도 사용하지 않기 때문에 멀티 스레드 상황에는 사용할 수 없다.
단일 스레드가 사용하는 경우엔 효율적이다.
VolatileInteger
volatile을 사용해서 CPU 캐시를 사용하지 않고 메인 메모리를 사용한다.
안전한 임계 영역이 없기 때문에 멀티 스레드 상황에는 사용할 수 없다.
단일 스레드가 사용하기엔 BasicInteger보다 느리다. 그리고 멀티 스레드 환경에서도 안전하지 않다.
SyncInteger
synchronized를 사용한 안전한 임계 영역이 있기 때문에 멀티 스레드 환경에서도 안전하게 사용할 수 있다.
MyAtomicInteger보다 성능이 느리다.
MyAtomicInteger
자바가 제공하는 AtomicInteger를 사용한다. 멀티 스레드 상황에 안전하게 사용할 수 있다.
성능도 synchronized, Lock(ReentrantLock)을 사용하는 경우보다 1.5 ~ 2배 정도 빠르다.
SyncInteger 처럼 락을 사용하는 경우보다, AtomicInteger가 더 빠른 이유는 무엇일까? i++ 연산은 원자적인 연산이 아니다. 따라서 분명히 synchronized, Lock(ReentrantLock)과 같은 락을 통해 안전히 임계 영역을 만들어야 할 것 같다. 놀랍게도 AtomicInteger가 제공하는 incrementAndGet() 메서드는 락을 사용하지 않고, 원자적 연산을 만들어 낸다.
CAS 연산
락 기반 방식의 문제점
SyncInteger와 같은 클래스는 데이터를 보호하기 위해 락을 사용한다. 여기서 말하는 락은 synchronized, Lock(ReentrantLock)등을 사용하는 것을 말한다. 락은 특정 자원을 보호하기 위해 스레드가 해당 자원에 대해 접근하는 것을 제한한다. 락이 걸려 있는 동안 다른 스레드들은 해당 자원에 접근할 수 없고, 락이 해제될 때까지 기다린다. 또한 락 기반 접근에서는 락을 획득하고 해제하는 데 시간이 소요된다.
예를 들어 락을 사용하는 연산이 있다고 하면 이런 흐름으로 동작한다.
락이 있는지 확인한다.
락을 획득하고 임계 영역에 들어간다.
작업을 수행한다.
락을 반납한다.
여기서 락을 획득하고 반납하는 과정이 계속 반복된다. 10000번의 연산이 있다면 10000번의 연산 모두 같은 과정을 반복한다. 이렇듯 락을 사용하는 방식은 직관적이지만 상대적으로 무겁다는 단점이 있다.
CAS
이런 문제를 해결하기 위해 락을 걸지 않고 원자적인 연산을 수행할 수 있는 방법이 있는데 이것을 CAS(Compare-And-Swap, Compare-And-Set) 연산이라고 한다. 이 방법은 락을 사용하지 않기 때문에 락 프리(lock-free) 기법이라고 한다. 참고로 CAS 연산은 락을 완전히 대체하는 게 아니라 작은 단위의 일부 영역에 적용할 수 있다. 기본은 락을 사용하고 특별한 경우에 CAS를 적용할 수 있다고 생각하면 된다.
다음 코드를 보자.
CasMainV1
package thread.cas;
import java.util.concurrent.atomic.AtomicInteger;
public class CasMainV1 {
public static void main(String[] args) {
AtomicInteger atomicInteger = new AtomicInteger(0);
System.out.println("start value = " + atomicInteger.get());
boolean result1 = atomicInteger.compareAndSet(0, 1);
System.out.println("result1 = " + result1 + ", value = " + atomicInteger.get());
boolean result2 = atomicInteger.compareAndSet(0, 1);
System.out.println("result2 = " + result2 + ", value = " + atomicInteger.get());
}
}
new AtomicInteger(0); 내부에 있는 기본 숫자 값을 0으로 설정한다.
자바는 AtomicXxx의 compareAndSet() 메서드를 통해 CAS 연산을 지원한다.
실행 결과
start value = 0
result1 = true, value = 1
result2 = false, value = 1
compareAndSet(0, 1)
atomicInteger가 가지고 있는 값이 현재 0이면 이 값을 1로 변경하라는 매우 단순한 메서드이다.
만약 atomicInteger의 값이 현재 0이라면 atomicInteger의 값은 1로 변경된다. 이 경우 true를 반환한다.
만약 atomicInteger의 값이 현재 0이 아니라면 atomicInteger의 값은 변경되지 않는다. 이 경우 false를 반환한다.
여기서 가장 중요한 내용이 있는데, 이 메서드는 원자적으로 실행된다는 점이다. 그리고 이 메서드가 제공하는 기능이 바로 CAS(compareAndSet)연산이다.
여기서는 AtomicInteger 내부에 있는 value의 값이 0이라면, 1로 변경하고 싶다.
그런데 생각해보면 이 명령어는 2개로 나누어진 명령어이다. 따라서 원자적이지 않은 연산처럼 보인다.
1. 먼저 메인 메모리에 있는 값을 확인
2. 해당 값이 기대하는 값이라면 원하는 값으로 변경
CPU 하드웨어의 지원
원자적이지 않은 연산도 원자적 연산으로 할 수 있는 이유는 바로 CPU의 지원 때문이다!
CAS 연산은 이렇게 원자적이지 않은 두 개의 연산을 CPU 하드웨어 차원에서 특별하게 하나의 원자적인 연산으로 묶어서 제공하는 기능이다. 이것은 소프트웨어가 제공하는 기능이 아니라 하드웨어가 제공하는 기능이다. 대부분의 현대 CPU들은 CAS 연산을 위한 명령어를 제공한다.
CPU는 다음 두 과정을 묶어서 하나의 원자적인 명령으로 만들어버린다. 따라서 중간에 다른 스레드가 개입할 수 없다.
x001의 값을 확인한다.
읽은 값이 0이면 1로 변경한다.
CPU는 두 과정을 하나의 원자적인 명령으로 만들기 위해 1번과 2번 사이에 다른 스레드가 x001의 값을 변경하지 못하게 막는다. 참고로 1번과 2번 사이의 시간은 CPU 입장에서 보면 진짜 아주 잠깐 찰나의 순간이다. 생각을 해보자. 1초에 몇번의 연산을 CPU가 할 수 있는지? 수억번이다 수억번. 그러니까 저 1번과 2번 사이에 다른 스레드가 변경하지 못하게 막는 행위가 성능에 큰 영향을 끼치지도 않는다.
value의 값이 0 → 1이 되었다.
CAS 연산으로 값을 성공적으로 변경하고 나면 true를 반환한다.
여기까지 듣고보면 CAS 연산도 별 게 아니다. 그냥 CPU 차원에서 하드웨어적으로 원자적 연산을 가능하게 해준다는 것이다. CPU 입장에서 그 찰나의 순간은 사람이 느끼지도 못할 정도의 시간이니까 성능의 차이도 딱히 없다. 그럼 결국 1. 값을 확인하고, 2. 값을 변경하는 두 연산을 하나로 묶어 원자적으로 제공한다는 것을 이해했을 것이다. 그런데 이 기능이 어떻게 락을 일부 대체할 수 있다는 걸까?
어떤 값을 하나 증가하는 value++ 연산은 원자적 연산이 아니다. 이 연산은 다음과 같다.
value = value + 1;
이 연산은 다음 순서로 실행된다. value의 초기값은 0으로 가정하겠다.
오른쪽에 있는 value의 값을 읽는다. value의 값은 0이다.
읽은 0에 1을 더해서 1을 만든다.
더한 1을 왼쪽에 value 변수에 대입한다.
1번과 3번 연산 사이에 다른 스레드가 value의 값을 변경할 수 있기 때문에, 문제가 될 수 있다. 따라서 value++ 연산을 여러 스레드에서 사용한다면 락을 건 다음에 값을 증가해야 한다.
CAS 연산을 활용해서 락 없이 값을 증가하는 기능을 만들어보자. AtomicInteger가 제공하는 incrementAndGet()메서드가 어떻게 CAS 연산을 활용해서 락 없이 만들어졌는지 직접 구현해보자.
CasMainV2
package thread.cas;
import java.util.concurrent.atomic.AtomicInteger;
import static util.MyLogger.log;
public class CasMainV2 {
public static void main(String[] args) {
AtomicInteger atomicInteger = new AtomicInteger(0);
System.out.println("start value = " + atomicInteger.get());
int resultValue1 = incrementAndGet(atomicInteger);
System.out.println("resultValue1 = " + resultValue1);
int resultValue2 = incrementAndGet(atomicInteger);
System.out.println("resultValue1 = " + resultValue2);
}
private static int incrementAndGet(AtomicInteger atomicInteger) {
int getValue;
boolean result;
do {
getValue = atomicInteger.get();
log("getValue : " + getValue);
result = atomicInteger.compareAndSet(getValue, getValue + 1);
log("result: " + result);
} while (!result);
return getValue + 1;
}
}
여기서 만든 incrementAndGet()은 atomicInteger 내부의 value 값을 하나 증가하는 메서드이다. 사실 atomicInteger도 이 메서드를 제공하지만 여기서는 이해를 위해 직접 구현해보자.
CAS 연산을 사용하면 여러 스레드가 같은 값을 사용하는 상황에서도 락을 걸지 않고, 안전하게 값을 증가할 수 있다. 여기서는 락을 걸지 않고 CAS 연산을 사용해서 값을 증가했다.
getValue = atomicInteger.get()을 사용해서 value값을 읽는다.
compareAndSet(getValue, getValue + 1)을 사용해서 방금 읽은 value값이 메모리의 value값과 같다면 value 값을 하나 증가한다. 여기서 CAS연산을 사용한다.
보면 결과가 false가 여러번 나오는 것을 확인할 수 있다. 그렇지만 결국 결과는 완벽하게 100을 찍었다. (스레드 개수를 100으로 했을 때 결과)
당연히 멀티 스레드 환경에서는 공유 자원에 대해 여러 스레드가 동시에 값을 읽고 쓰고 할 수 있기 때문에 위 실행 결과에서 Thread-69 입장에서 본인이 읽었을 때 시점과 0.1초 후에 다시 CAS 연산을 시도했을 때 예상한 원래 값이 달라질 수 있을 것이라는 추측이 가능하다. 그럼에도 원하는 결과를 정확히 찍을 수 있는 이유는 연산 시점에서는 적어도? 다른 스레드가 접근하지 못하도록 CPU 차원에서 막고 있기 때문이다. 이게 바로 CAS 연산이다.
정리를 하자면,
AtomicInteger가 제공하는 incrementAndGet() 코드가 우리가 직접 작성한 incrementAndGet() 메서드와 똑같이 CAS를 활용하도록 작성되어 있다. 그리고 조건에 맞을때까지 루프를 돌면서 확인하는 코드도 똑같다. CAS를 활용하면 락을 사용하지 않지만, 대신에 다른 스레드가 값을 먼저 증가해서 문제가 발생하는 경우 루프를 다시 돌아 재시도를 하는 방식으로 사용한다.
이 방식은 다음과 같이 동작한다.
현재 변수의 값을 읽어온다.
변수의 값을 1 증가시킬 때, 원래 값이 같은지 확인한다. (CAS 연산 활용)
동일하다면 증가된 값을 변수에 저장하고 종료한다.
동일하지 않다면 다른 스레드가 값을 중간에 변경한 것이므로, 다시 처음으로 돌아가 위 과정을 반복한다.
두 스레드가 동시에 실행되면서 문제가 발생하는 상황을 스레드가 충돌했다고 표현한다.
이 과정에서 충돌이 발생할 때 마다 반복해서 다시 시도하므로, 결과적으로 락 없이 데이터를 안전하게 변경할 수 있다. CAS 연산을 사용하는 방식은 충돌이 드물게 발생하는 환경에서는 락을 사용하지 않으므로 높은 성능을 발휘할 수 있다. 이는 락을 사용하는 방식과 비교했을 때, 스레드가 락을 획득하기 위해 대기하지 않기 때문에 대기 시간과 오버헤드가 줄어드는 장점이 있다.
그러나, 충돌이 빈번하게 발생하는 환경에서는 성능에 문제가 될 수 있다. 여러 스레드가 자주 동시에 동일한 변수의 값을 변경하려고 시도할 때, CAS는 자주 실패하고 재시도해야 하므로 성능 저하가 발생할 수 있다. 이런 상황에서는 반복문을 계속 돌기 때문에 CPU 자원을 많이 소모하게 된다.
CAS(Compare-And-Set)와 락(Lock)방식의 비교
락 방식
비관적(pessimistic) 접근법 (기본적으로 "여기선 충돌이 날거야!" 라고 생각하고 아예 입구를 틀어막는 관점)
데이터에 접근하기 전에 항상 락을 획득
다른 스레드의 접근을 막음
"다른 스레드가 방해할 것이다"라고 가정
CAS 방식
낙관적(optimistic) 접근법
락을 사용하지 않고 데이터에 바로 접근
충돌이 발생하면 그때 재시도
"대부분의 경우 충돌이 없을 것이다"라고 가정
정리하면, 충돌이 많이 없는 경우에 CAS 연산이 빠른 것을 확인할 수 있다.
그럼 충돌이 많이 발생하지 않는 연산은 어떤 것이 있을까? 언제 CAS 연산을 사용하면 좋을까?
사실 간단한 CPU 연산은 너무 빨리 처리되기 때문에 충돌이 자주 발생하지 않는다. 충돌이 발생하기도 전에 이미 연산을 완료하는 경우가 더 많다.
즉, 간단한 CPU 연산에는 락 보단 CAS 연산을 사용하면 더 효율적이고 복잡하고 어려운 과정이 들어가있는, 시간이 많이 소모되는 연산에 대해서는 락 방식으로 완전히 안전한 임계 영역을 만들어 사용하면 될 것 같다.
CAS 락
CAS는 단순 연산 뿐만 아니라, 락을 구현하는데도 사용할 수 있다. synchronized, Lock(ReentrantLock)없이 CAS를 활용해서 락을 구현해보자. 먼저 CAS의 필요성을 이해하기 위해 CAS없이 직접 락을 구현해보자.
SpinLockBad
package thread.cas.spinlock;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class SpinLockBad {
private volatile boolean lock = false;
public void lock() {
log("락 획득 시도");
while (true) {
if (!lock) { // 1. 락 사용 여부 확인
sleep(100);
lock = true; // 2. 락의 값 변경
break;
} else {
log("락 획득 실패 - 스핀 대기");
}
}
log("락 획득 완료");
}
public void unlock() {
lock = false;
log("락 반납 완료");
}
}
구현 원리는 매우 단순하다.
스레드가 락을 획득하면 lock의 값이 true가 된다.
스레드가 락을 반납하면 lock의 값이 false가 된다.
스레드가 락을 획득하면 while문을 빠져나온다.
스레드가 락을 획득하지 못하면 락을 획득할 때까지 while문을 계속 반복 실행한다.
"어?왜 스핀 락인가요?" 락이 해제되기를 기다리면서 계속해서 반복문을 통해 확인하는 모습이 마치 계속 빙글빙글 돌고 있는것 같다고 해서 스핀 락이라고 불린다.
SpinLockMain
package thread.cas.spinlock;
import static util.MyLogger.log;
public class SpinLockMain {
public static void main(String[] args) {
SpinLockBad spinLock = new SpinLockBad();
Runnable task = new Runnable() {
@Override
public void run() {
spinLock.lock();
try {
log("비즈니스 로직 실행");
} finally {
spinLock.unlock();
}
}
};
Thread t1 = new Thread(task, "Thread-1");
Thread t2 = new Thread(task, "Thread-2");
t1.start();
t2.start();
}
}
실행 결과
2024-07-26 15:43:41.447 [ Thread-1] 락 획득 시도
2024-07-26 15:43:41.447 [ Thread-2] 락 획득 시도
2024-07-26 15:43:41.553 [ Thread-1] 락 획득 완료
2024-07-26 15:43:41.553 [ Thread-2] 락 획득 완료
2024-07-26 15:43:41.553 [ Thread-1] 비즈니스 로직 실행
2024-07-26 15:43:41.553 [ Thread-2] 비즈니스 로직 실행
2024-07-26 15:43:41.554 [ Thread-1] 락 반납 완료
2024-07-26 15:43:41.554 [ Thread-2] 락 반납 완료
실행 결과를 보면 기대와 다르게 Thread-0, Thread-1 모두 둘 다 동시에 락을 획득했다. 이제는 왜 그런지 안다. 동시성 문제를 해결하지 못한 코드이기 때문이다. volatile은 동시성 문제를 해결하는 방안이 아니다.
그럼 여기서 어떤 부분이 문제일까? 바로 다음 두 부분이 원자적이지 않다는 것이다.
락 사용 여부 확인
락의 값 변경
이 둘은 한번에 하나의 스레드만 실행해야 한다. 따라서 synchronized 또는 Lock을 사용해서 두 코드를 동기화해서 안전한 임계 영역을 만들어야 한다. 여기서! 다른 해결 방안도 있다. 바로 두 코드를 하나로 묶어서 원자적 연산처리를 하는 것이다.
CAS 연산을 사용하면 두 연산을 하나로 묶어서 하나의 원자적인 연산으로 처리할 수 있다.
락의 사용 여부를 확인하고, 그 값이 기대하는 값과 같다면 변경하는 것이다. CAS 연산에 딱 들어 맞는다!
SpinLock
package thread.cas.spinlock;
import java.util.concurrent.atomic.AtomicBoolean;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class SpinLock {
private final AtomicBoolean lock = new AtomicBoolean(false);
public void lock() {
log("락 획득 시도");
while (!lock.compareAndSet(false, true)) {
log("락 획득 실패 - 스핀 대기");
}
log("락 획득 완료");
}
public void unlock() {
lock.set(false);
log("락 반납 완료");
}
}
CAS 연산을 지원하는 AtomicBoolean을 사용했다.
구현 원리는 단순하다.
스레드가 락을 획득하면 lock의 값이 true가 된다.
스레드가 락을 반납하면 lock의 값이 false가 된다.
스레드가 락을 획득하면 while문을 빠져나온다.
스레드가 락을 획득하지 못하면 락을 획득할 때까지 while문을 계속 반복 실행한다.
락을 획득할 때 매우 중요한 부분이 있다. 바로 다음 두 연산을 하나로 만들어야 한다는 것이다.
락 사용 여부 확인
락의 값 변경
여기에 딱 맞는 방법이 바로 다음 코드 한 줄이다.
lock.compareAndSet(false, true);
→ 현재 lock의 값이 false라면 true로 변경해라.
이것은 CAS 연산으로 수행된다.
이 SpinLock을 사용해서 다시 실행해보자!
package thread.cas.spinlock;
import static util.MyLogger.log;
public class SpinLockMain {
public static void main(String[] args) {
SpinLock spinLock = new SpinLock();
Runnable task = new Runnable() {
@Override
public void run() {
spinLock.lock();
try {
log("비즈니스 로직 실행");
} finally {
spinLock.unlock();
}
}
};
Thread t1 = new Thread(task, "Thread-1");
Thread t2 = new Thread(task, "Thread-2");
t1.start();
t2.start();
}
}
실행 결과
2024-07-26 15:50:56.510 [ Thread-1] 락 획득 시도
2024-07-26 15:50:56.510 [ Thread-2] 락 획득 시도
2024-07-26 15:50:56.516 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:50:56.516 [ Thread-1] 락 획득 완료
2024-07-26 15:50:56.516 [ Thread-1] 비즈니스 로직 실행
2024-07-26 15:50:56.516 [ Thread-1] 락 반납 완료
2024-07-26 15:50:56.516 [ Thread-2] 락 획득 완료
2024-07-26 15:50:56.517 [ Thread-2] 비즈니스 로직 실행
2024-07-26 15:50:56.517 [ Thread-2] 락 반납 완료
실행 결과를 보면 락이 잘 적용된 것을 알 수 있다.
이렇게 CAS는 연산뿐 아니라 락을 구현해낼 수도 있다. 그렇지만 단점도 있다. 만약, 다음 코드로 변경하면 어떻게 될까?
실제로 실행되는 비즈니스 로직, 그러니까 스레드가 실행하는 로직이 1 MS만 늘어나도 이 CAS를 사용한 락은 CPU를 많이 갉아먹을 것이다.
실행 결과
2024-07-26 15:55:23.806 [ Thread-1] 락 획득 시도
2024-07-26 15:55:23.806 [ Thread-2] 락 획득 시도
2024-07-26 15:55:23.812 [ Thread-1] 락 획득 완료
2024-07-26 15:55:23.812 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.813 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.813 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.813 [ Thread-1] 비즈니스 로직 실행
2024-07-26 15:55:23.813 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.814 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.814 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.814 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.815 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.815 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.815 [ Thread-2] 락 획득 실패 - 스핀 대기
2024-07-26 15:55:23.815 [ Thread-1] 락 반납 완료
2024-07-26 15:55:23.815 [ Thread-2] 락 획득 완료
2024-07-26 15:55:23.816 [ Thread-2] 비즈니스 로직 실행
2024-07-26 15:55:23.817 [ Thread-2] 락 반납 완료
무슨 말이냐면, 비즈니스 로직의 실행 처리 속도나 복잡도가 올라가면 올라갈수록 이 락을 획득하기 위해 계속해서 스핀락은 시도를 할거고 그 시도엔 CPU 자원을 사용한다는 점이다.
그래서, 계속 말하지만 결론은, 연산이 길지 않고 매우매우매우 짧게 끝날 때는 CAS가 더 효율적일 수 있고, 연산이 복잡하고 길다면 무조건 Lock(ReentrantLock)을 사용하면 된다. 그러니까 더 짧은 결론은 기본적으론 Lock(ReentrantLock)을 사용하는데 아주 특별한 경우에 한정해서 CAS를 사용하면 최적화할 수 있다.
정리
동기화 락을 사용하는 방식과 CAS를 활용하는 락 프리 방식의 장단점을 비교해보자!
CAS
장점
낙관적 동기화: 락을 걸지 않고도 값을 안전하게 업데이트 할 수 있다. CAS는 충돌이 자주 발생하지 않을 것이라고 가정한다. 이는 충돌이 적은 환경에서 높은 성능을 발휘한다.
락 프리: CAS는 락을 사용하지 않기 때문에, 락을 획득하기 위해 대기하는 시간이 없다. 따라서 스레드가 블로킹되지 않으며, 병렬 처리가 더 효율적일 수 있다.
단점
충돌이 빈번한 경우: 여러 스레드가 동시에 동일한 변수에 접근하여 업데이트를 시도할 때 충돌이 발생할 수 있다. 충돌이 발생하면 CAS는 루프를 돌며 재시도해야 하며, 이에 따라 CPU 자원을 계속 소모할 수 있다. 반복적인 재시도로 인해 오버헤드가 발생할 수 있다.
스핀락과 유사한 오버헤드: CAS는 충돌 시 반복적인 재시도를 하므로, 이 과정이 계속 반복되면 스핀락과 유사한 성능 저하가 발생할 수 있다. 특히 충돌 빈도가 높을수록 이런 현상이 두드러진다.
동기화 락
장점
충돌 관리: 락을 사용하면 하나의 스레드만 리소스에 접근할 수 있으므로 충돌이 발생하지 않는다. 여러 스레드가 경쟁할 경우에도 안정적으로 동작한다.
안정성: 복잡한 상황에서도 락은 일관성 있는 동작을 보장한다.
스레드 대기: 락을 대기하는 스레드는 CPU를 거의 사용하지 않는다.
단점
락 획득 대기 시간: 스레드가 락을 획득하기 위해 대기해야 하므로, 대기 시간이 길어질 수 있다.
컨텍스트 스위칭 오버헤드: 락을 사용하면, 락 획득을 대기하는 시점과 또 락을 획득하는 시점에 스레드의 상태가 변경된다. 이에 컨텍스트 스위칭이 발생할 수 있으며 이로 인해 오버헤드가 증가할 수 있다.
결론
일반적으로 동기화 락을 사용하고, 아주 특별한 경우에 한정해서 CAS를 활용해서 최적화해야 한다. CAS를 통한 최적화가 더 나은 경우는 스레드가 RUNNABLE → BLOCKED, WAITING 상태에서 다시 RUNNABLE 상태로 가는 것 보다는, 스레드를 RUNNABLE로 살려둔 상태에서 계속 락 획득을 반복 체크하는 것이 더 효율적인 경우에 사용해야 한다. 하지만 이 경우 대기하는 스레드가 CPU 자원을 계속 소모하기 때문에 대기 시간이 아주아주아주 짧아야 한다. 따라서 임계 영역이 필요는 하지만, 연산이 길지 않고 매우매우매우 짧게 끝날 때 사용해야 한다. 예를 들어 숫자 값의 증가, 자료 구조의 데이터 추가, 삭제와 같이 CPU 사이클이 금방 끝나지만 안전한 임계 영역, 또는 원자적인 연산이 필요한 경우에 사용해야 한다.
반면에 데이터베이스를 기다린다거나, 다른 서버의 요청을 기다리는 것처럼 오래 기다리는 작업에 CAS를 사용하면 CPU를 계속 사용하며 기다리는 최악의 결과가 나올 수도 있다. 이런 경우에는 동기화 락을 사용해야 한다.
또한 CAS는 충돌 가능성이 낮은 환경에서 매우 효율적이지만, 충돌 가능성이 높은 환경에서는 성능 저하가 발생할 수 있다. 이런 경우에는 상황에 맞는 적절한 동기화 전략을 사용하는 것이 중요하다. 때로는 락이 더 나은 성능을 발휘할 수 있으며 CAS가 항상 더 빠르다고 단정할 수는 없다. 따라서 각 접근 방식의 특성을 이해하고 애플리케이션의 특정 요구사항과 환경에 맞는 방식을 선택하는 것이 중요하다.
실무 관점
실무 관점에서 보면 대부분의 애플리케이션들은 공유 자원을 사용할 때 충돌할 가능성보다 충돌하지 않을 가능성이 훨씬 높다. 예를 들어, 여러 스레드에서 발생하는 주문 수를 실시간으로 증가하면서 카운트 한다고 가정해보자. 그리고 특정 피크시간에 주문이 100만건 들어오는 서비스라고 가정해보자 (이 정도면 국내 업계 탑이다).
1,000,000 / 60분 = 1분에 16,666건, 1초에 277건
CPU가 1초에 얼마나 많은 연산을 처리하는지 생각해보면, 백만 건 중에 충돌이 나는 경우는 아주 넉넉하게 잡아도 몇 십 건 이하일 것이다. 따라서 실무에서는 주문 수 증가와 같은 단순한 연산의 경우, 락을 걸고 시작하는 것 보다는, CAS처럼 낙관적인 방식이 더 나은 성능을 보인다.
그런데 여기서 중요한 핵심은 주문 수 증가와 같은 단순한 연산이라는 점이다. 이런 경우에는 AtomicInteger와 같은 CAS 연산을 사용하는 방식이 더 효과적이다. 이런 연산은 나노 초 단위로 발생하는 연산이다. 반면에 데이터베이스를 기다린다거나, 다른 서버의 요청을 기다리는 것 처럼 수 밀리초 이상의 시간이 걸리는 작업이라면 CAS를 사용하는 것보단 동기화 락을 사용하거나 스레드가 대기하는 방식이 더 효과적이다.
저번 포스팅에서 다뤘던 생산자 소비자 문제의 두번째 포스팅이다. 이 포스팅에선 어떻게 저번 포스팅에 말했던 문제를 해결하는지를 알아보자. 우선 저번 포스팅에서 말했던 문제는 생산자가 생산자를 깨워버릴 수 있고, 소비자가 소비자를 깨워버릴 수 있다는 문제였다. 그렇게 되면 결국 깨어난 스레드는 아무것도 하지 못하고 다시 기다리는 상태로 돌아가야 한다.
Lock Condition
이제, synchronized를 사용하지 않겠다. 이전에 synchronized의 단점을 극복하기 위해 사용했던 ReentrantLock을 사용하면 이 문제에서도 역시나 도움을 준다.
어떻게 도움을 주냐? 기존의 문제는 생산자가 생산자를 깨울 가능성이 있고 반대로 소비자가 소비자를 깨울 가능성이 있기 때문에 비효율이 발생하는 것이다. 그리고 그 근본 원인은? 스레드 대기 집합이 딱 하나이기 때문이다. 이 스레드 대기 집합은 모든 객체가 기본으로 가지고 있다. 모니터 락과 synchronized와 같이 사용되는 것이다.
그럼 스레드 대기 집합을 둘로 나누면 된다. 생산자용 대기 집합과 소비자용 대기 집합으로. 그리고 깨울땐 소비자라면 생산자용 스레드 대기 집합에 알리면 되고, 생산자라면 소비자용 스레드 대기 집합에 알리면 된다!
우선, 대기 집합을 분리하기 전에 앞에서 사용했던 synchronized, wait(), notify()를 통해 작성한 코드를 Lock 인터페이스와 ReentrantLock 구현체를 이용해서 다시 구현해보자!
BoundedQueueV4
package thread.bounded;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import static util.MyLogger.log;
public class BoundedQueueV4 implements BoundedQueue {
private final Queue<String> queue = new ArrayDeque<>();
private final int max;
private final Lock lock = new ReentrantLock();
private final Condition condition = lock.newCondition();
public BoundedQueueV4(int max) {
this.max = max;
}
@Override
public void put(String data) {
lock.lock();
try {
while (queue.size() == max) {
log("[put] 큐가 가득 찼습니다. 생산자는 대기합니다.");
try {
condition.await();
log("[put] 생산자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
queue.offer(data);
log("[put] 생산자 데이터 저장, notify() 호출");
condition.signal();
} finally {
lock.unlock();
}
}
@Override
public String take() {
lock.lock();
try {
while (queue.isEmpty()) {
log("[take] 큐에 데이터가 없습니다. 소비자는 대기합니다");
try {
condition.await();
log("[take] 소비자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
String data = queue.poll();
log("[take] 소비자 데이터 획득, notify() 호출");
condition.signal();
return data;
} finally {
lock.unlock();
}
}
@Override
public String toString() {
return queue.toString();
}
}
synchronized 대신 Lock lock = new ReentrantLock을 사용한다.
Condition이 등장한다. 이 Condition은 ReentrantLock을 사용하는 스레드가 대기하는 스레드 대기 공간이다.
lock.newCondition() 메서드를 호출하면 스레드 대기 공간이 만들어진다. Lock(ReentrantLock)의 스레드 대기 공간은 이렇게 만들 수가 있다.
참고로 synchronized, 모니터 락, wait(), notify()에서 사용한 스레드 대기 공간은 모든 객체 인스턴스가 내부에 기본으로 가지고 있다. 반면에 Lock(ReentrantLock)을 사용하는 경우 이렇게 스레드 대기 공간을 직접 만들어서 사용해야 한다.
condition.await()
Object.wait()과 유사한 기능이다. 지정한 condition에 현재 스레드를 대기(WAITING)상태로 보관한다. 이때 ReentrantLock에서 획득한 락을 반납하고 대기 상태로 condition에 보관된다.
condition.signal()
Object.notify()와 유사한 기능이다. 지정한 condition에서 대기중인 스레드를 하나 깨운다. 깨어난 스레드는 condition에서 빠져나온다.
Lock lock = new ReentrantLock()
이 그림에서 lock은 synchronized에서 사용하는 객체 내부에 모니터 락이 아니라, ReentrantLock 락을 뜻한다. ReentrantLock은 내부에 락과, 락 획득을 대기하는 스레드를 관리하는 대기 큐가 있다.
이 그림에서 스레드 대기 공간은 synchronized에서 사용하는 스레드 대기 공간이 아니라, 다음 코드를 뜻한다.
Condition condition = lock.newCondition()
ReentrantLock을 사용하면, condition이 스레드 대기 공간이다.
지금까지는 synchronized, wait(), notify()를 사용한 이전 코드와 거의 비슷하다. 아직 생산자용, 소비자용 스레드 대기 공간을 따로 분리하지 않았기 때문에 기존 방식과 같다고 보면 된다. 다만 구현을 synchronized로 했는가 아니면 ReentrantLock을 사용해서 했는가에 차이가 있을 뿐이다.
이대로 실행해보면 실행결과는 기존과 똑같을 것이다. 이제 스레드 대기 공간을 생산자용과 소비자용으로 분리해보자!
생산자, 소비자 대기 공간 분리
이런 그림을 만들어보자! 생각보다 엄청 간단하다.
BoundedQueueV5
package thread.bounded;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import static util.MyLogger.log;
public class BoundedQueueV5 implements BoundedQueue {
private final Queue<String> queue = new ArrayDeque<>();
private final int max;
private final Lock lock = new ReentrantLock();
private final Condition producerCond = lock.newCondition();
private final Condition consumerCond = lock.newCondition();
public BoundedQueueV5(int max) {
this.max = max;
}
@Override
public void put(String data) {
lock.lock();
try {
while (queue.size() == max) {
log("[put] 큐가 가득 찼습니다. 생산자는 대기합니다.");
try {
producerCond.await();
log("[put] 생산자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
queue.offer(data);
log("[put] 생산자 데이터 저장, notify() 호출");
consumerCond.signal();
} finally {
lock.unlock();
}
}
@Override
public String take() {
lock.lock();
try {
while (queue.isEmpty()) {
log("[take] 큐에 데이터가 없습니다. 소비자는 대기합니다");
try {
consumerCond.await();
log("[take] 소비자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
String data = queue.poll();
log("[take] 소비자 데이터 획득, notify() 호출");
producerCond.signal();
return data;
} finally {
lock.unlock();
}
}
@Override
public String toString() {
return queue.toString();
}
}
여기서 핵심은 lock.newCondition()을 두 번 호출해서 ReentrantLock을 사용하는 스레드 대기 공간을 2개 만드는 것이다.
private final Lock lock = new ReentrantLock();
private final Condition producerCond = lock.newCondition();
private final Condition consumerCond = lock.newCondition();
Condition 분리
producerCond: 생산자를 위한 스레드 대기 공간
consumerCond: 소비자를 위한 스레드 대기 공간
이렇게 하면 생산자 스레드, 소비자 스레드를 정확하게 나누어 관리하고 깨울 수 있다!
@Override
public void put(String data) {
lock.lock();
try {
while (queue.size() == max) {
log("[put] 큐가 가득 찼습니다. 생산자는 대기합니다.");
try {
producerCond.await();
log("[put] 생산자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
queue.offer(data);
log("[put] 생산자 데이터 저장, notify() 호출");
consumerCond.signal();
} finally {
lock.unlock();
}
}
put(data) 메서드는 결국 생산자가 큐에 데이터를 생산해내는 것이다. 그렇다면 이 메서드에서 만약 큐가 가득찼다면 어디로 들어가면 될까? producerCond 안으로 들어가면 된다. 그래서 producerCond.await()을 호출한다.
그리고 큐가 가득차지 않아서 데이터를 잘 넣었다면 누굴 깨우면 될까? 소비자 스레드를 깨우면 된다. 그래서 consumerCond.signal()을 호출한다.
@Override
public String take() {
lock.lock();
try {
while (queue.isEmpty()) {
log("[take] 큐에 데이터가 없습니다. 소비자는 대기합니다");
try {
consumerCond.await();
log("[take] 소비자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
String data = queue.poll();
log("[take] 소비자 데이터 획득, notify() 호출");
producerCond.signal();
return data;
} finally {
lock.unlock();
}
}
take() 메서드는 소비자 스레드가 사용하는 메서드이다. 만약 큐에 데이터가 비었다면 어디로 들어가면 될까? 소비자용 스레드 대기 공간으로 들어가면 된다. 그래서 consumerCond.await()을 호출한다.
큐에 데이터가 있어서 데이터를 잘 소비했다면 누굴 깨우면 될까? 생산자 스레드를 깨우면 된다. 그래서 producerCond.signal()을 호출한다.
이렇게 굉장히 간단하게 딱 필요한 스레드 대기 공간에 알리고, 자기가 들어갈 곳을 잘 들어갈 수 있게 됐다. 그리고 실제로 이 코드로 실행해보면 비효율은 하나도 발생하지 않는다! 왜냐!? 생산자 스레드는 소비자 스레드만을 깨우고, 소비자 스레드는 생산자 스레드만을 깨우기 때문에.
실행 결과 로그는 따로 작성하지 않겠다. 직접 실행해보면 될 것 같다.
한번 이 큰 그림을 그림으로 하나씩 분석해보자.
참고로, 지금 그림을 가지고 하는 설명은 위 예제 실행 결과와 살짝 다르다. 이해하기 쉽게 조금 변형했다고 보면 된다
생산자 먼저 실행
c1, c2, c3는 소비자 스레드 전용 대기 공간(consumerCond)에 대기중이다.
p1, p2, p3는 생산자 스레드 전용 대기 공간(producerCond)에 대기중이다.
큐에 데이터가 비어있다.
생산자인 p0 스레드가 실행 예정이다.
p0 스레드는 ReentrantLock의 락을 획득하고 큐에 데이터를 보관한다.
생산자 스레드가 큐에 데이터를 보관했기 때문에, 소비자 스레드가 가져갈 데이터가 추가되었다. 따라서 소비자 대기 공간(consumerCond)에 signal()을 통해 알려준다.
소비자 스레드 중에 하나가 깨어난다. c1이 깨어났다고 가정하자.
c1은 락 획득까지 잠시 대기하다가 이후에 p0가 반납한 ReentrantLock의 락을 획득한다. 그리고 큐의 데이터를 획득한 다음에 완료된다.
소비자 먼저 실행
이 설명 역시 예제 결과와 살짝 다르다.
c1, c2, c3는 소비자 스레드 전용 대기 공간(consumerCond)에 대기중이다.
p1, p2, p3는 생산자 스레드 전용 대기 공간(producerCond)에 대기중이다.
큐에 데이터가 가득 차 있다.
소비자인 c0 스레드가 실행 예정이다.
c0 스레드는 ReentrantLock의 락을 획득하고 큐에 있는 데이터를 획득한다.
큐에 데이터를 획득했기 때문에, 큐에 데이터를 생산할 수 있는 빈 공간이 생겼다. 생산자 대기 공간(producerCond)에 signal()을 통해 알려준다.
생산자 스레드 중에 하나가 깨어난다. p3가 깨어났다고 가정하자.
p3는 이후에 c0가 반납한 ReentrantLock의 락을 획득하고, 큐의 데이터를 저장한 다음에 완료된다.
Object.notify() vs Condition.signal()
Object.notify()
대기 중인 스레드 중 임의의 하나를 선택해 깨운다. 스레드가 깨어나는 순서는 보장되어 있지 않으며, JVM 구현에 따라 다르다. 보통은 먼저 들어온 스레드가 먼저 수행되지만 구현에 따라 다를 수 있다.
synchronized 블록 내에서 모니터 락을 가지고 있는 스레드가 호출해야 한다.
Condition.signal()
대기중인 스레드 중 하나를 깨우며, 일반적으로는 FIFO 순서로 깨운다. 이 부분은 자바 버전과 구현에 따라 달라질 수 있지만, 보통 Condition의 구현은 Queue 구조를 사용하기 때문에 FIFO 순서로 깨운다.
ReentrantLock의 락을 가지고 있는 스레드가 호출해야 한다.
스레드의 대기
사실 지금까지 얘기하지 않고 있던 부분이 하나 있다.
이제는 synchronized, ReentrantLock의 대기 상태에 대해 이야기할 차례가 됐다.
먼저 synchronized의 대기 상태부터 얘기해보자면, 잘 생각해보면 2가지 단계의 대기 상태가 존재한다.
synchronized대기
대기1: 락 획득 대기
BLOCKED 상태로 모니터 락 획득 대기
synchronized를 시작할 때 락이 없으면 대기
다른 스레드가 synchronized를 빠져나갈 때 모니터 락 획득 시도
대기2: wait()대기
WAITING 상태로 대기
wait()을 호출했을 때 스레드 대기 집합에서 대기
다른 스레드가 notify()를 호출했을 때 빠져나감
소비자 스레드: c1, c2, c3
생산자 스레드: p1, p2, p3
소비자 스레드 c1, c2, c3가 먼저 동시에 실행된다고 가정하자.
소비자 스레드 c1이 가장 먼저 락을 획득한다.
c2, c3는 락 획득을 대기하며 BLOCKED 상태가 된다.
c2, c3는 락 획득을 시도하지만, 모니터 락이 없기 때문에 락을 대기하며 BLOCKED 상태가 된다. c1은 나중에 락을 반납할 것이다. 그러면 c2, c3 중에 하나가 락을 획득해야 한다. 그런데 잘 생각해보면 락을 기다리는 c2, c3도 어딘가에서 관리가 되어야 한다. 그래야 락이 반환되었을 때 자바가 c2, c3중에 하나를 선택해서 락을 제공할 수 있다. 예를 들어서 List, Set, Queue와 같은 자료구조에 관리가 되어야 한다. 그림에서는 c2, c3가 단순히 BLOCKED 상태로 변경만 되었다. 그래서 관리되는 것처럼 보이지는 않는다.
사실은 BLOCKED 상태의 스레드도 자바 내부에서 따로 관리된다. 다음 그림을 보자.
이 그림은 이전 그림과 같은 상태를 좀 더 자세히 그린 그림이다.
그림을 보면 락 대기 집합이라는 곳이 있다. 이곳은 락을 기다리는 BLOCKED 상태의 스레드들을 관리한다.
락 대기 집합은 자바 내부에 구현되어 있기 때문에 모니터 락과 같이 개발자가 확인하기는 어렵다.
여기서는 BLOCKED 상태의 스레드 c2, c3가 관리된다.
언젠가 c1이 락을 반납하면 락 대기 집합에서 관리되는 스레드 중 하나가 락을 획득한다.
락 대기 집합이 지금에서야 나온 이유는,
단순하게 설명하기 위해 BLOCKED 상태에서 사용하는 락 대기 집합을 일부러 얘기하지 않았다. 지금쯤이면 이 내용을 말해도 이해하는데 어려움이 없을 것이다.
c1이 큐에 데이터가 없으므로 wait()을 호출하고 스레드 대기 집합에 들어간다.
c1은 락을 반납 후 스레드 대기 집합에 들어가고 WAITING 상태가 된다.
이후에 락이 반납됐으니 락을 기다리고있는 BLOCKED 상태의 스레드 중 하나가 락을 획득한다. 어차피 c2, c3 둘 다 소비자 스레드이기 때문에 락을 획득하고 임계 영역에 들어와도 아무것도 하지 못한채 wait()을 호출하고 둘 다 스레드 대기 집합으로 들아간 모습이다.
p1이 락을 획득하고 데이터를 저장한 다음 notify()를 호출하여 스레드 대기 집합에 이 사실을 알린다.
스레드 대기 집합에 있는 c1이 스레드 대기 집합을 빠져나간다.
하지만 아직 끝난 것이 아니다. 락을 얻지 못한 상태이니까 BLOCKED 상태가 된다. 그리고 락을 기다리는 BLOCKED 상태의 스레드들은 락 대기 집합에서 대기한다.
그러니까 사실 2번의 대기 과정이 있고 대기 장소가 있는것이다. 그래서 이 그림이 된다.
c1은 BLOCKED 상태에서 락을 얻을 때까지 락 대기 집합에서 대기한다.
드디어 p1이 락을 반납한다.
락이 반납되면 락 대기 집합에 있는 스레드 중 하나가 락을 획득한다. 여기서는 c1이 획득한다.
c1은 드디어 락 대기 집합까지 탈출하고 임계 영역을 수행할 수 있다.
정리를 하자면,
자바의 모든 객체 인스턴스는 멀티 스레드와 임계 영역을 다루기 위해 내부에 3가지 기본 요소를 가진다.
모니터 락
락 대기 집합(모니터 락 대기 집합)
스레드 대기 집합
여기서 락 대기 집합이 1차 대기소이고, 스레드 대기 집합이 2차 대기소라 생각하면 된다. 2차 대기소에 들어간 스레드는 2차, 1차 대기소를 모두 빠져나와야 임계 영역을 수행할 수 있다. 이 3가지 요소는 서로 맞물려 들어간다.
synchronized를 사용한 임계 영역에 들어가려면 모니터 락이 필요하다.
모니터 락이 없으면 락 대기 집합에 들어가서 BLOCKED 상태로 락을 기다린다.
모니터 락을 반납하면 락 대기 집합에 있는 스레드 중 하나가 락을 획득하고 BLOCKED → RUNNABLE 상태가 된다.
wait()을 호출해서 스레드 대기 집합에 들어가기 위해선 모니터 락이 필요하다.
스레드 대기 집합에 들어가면 모니터 락을 반납한다.
스레드가 notify()를 호출하면 스레드 대기 집합에 있는 스레드 중 하나가 스레드 대기 집합을 빠져나온다. 그리고 모니터 락 획득을 시도한다.
모니터 락을 획득하면 임계 영역을 수행한다.
모니터 락을 획득하지 못하면 락 대기 집합에 들어가서 BLOCKED 상태로 락을 기다린다.
여기까지가 synchronized를 사용한 스레드 대기의 전체 과정이다. 그럼 ReentrantLock은 크게 다를까? 똑같다. 살짝살짝만 다른 부분이 있고 이 메커니즘은 똑같다고 보면 된다.
ReentrantLock대기
대기1: ReentrantLock 락 획득 대기
ReentrantLock의 대기 큐에서 관리
WAITING 상태로 락 획득 대기
lock.lock()을 호출 했을 때 락이 없으면 대기
다른 스레드가 lock.unlock()을 호출 했을 때 락 획득을 시도, 락을 획득하면 대기 큐를 빠져나감
대기2: await() 대기
condition.await()을 호출 했을 때, condition 객체의 스레드 대기 공간에서 관리
WAITING 상태로 대기
다른 스레드가 condition.signal()을 호출 했을 때 condition 객체의 스레드 대기 공간에서 빠져나감
이 ReentrantLock도 synchronized와 마찬가지로 대기소가 2단계로 되어 있다. 2단계 대기소인 condition 객체의 스레드 대기 공간을 빠져나간다고 바로 실행되는 것이 아니다. 임계 영역 안에서는 항상 락이 있는 하나의 스레드만 실행될 수 있다. 여기서는 ReentrantLock의 락을 획득해야 RUNNABLE 상태가 되면서 그 다음 코드를 실행할 수 있다. 락을 획득하지 못하면 WAITING 상태로 락을 획득할 때 까지 ReentrantLock의 대기 큐에서 대기한다.
차이점이라고 하면, synchronized는 락 대기 집합에서 스레드들은 BLOCKED 상태라는 거고 ReentrantLock의 대기 큐에서 스레드들은 WAITING 상태라는 것이다. 그래서, 이런 사소한 차이가 있고 결국 대기소는 둘 다 2개가 있다라는 점이다. 물론 ReentrantLock을 사용하면 더 유연하고 많은 장점이 있다. 무한 대기를 하지 않아도 되는 점과 공정성을 해결할 수 있다는 것과 스레드 대기 집합을 여러개로 분리할 수 있다는 것이다.
중간 정리
여기까지 잘 익혔다면, 진짜 어지간한 멀티 스레드 환경과 흐름을 이해했다고 봐도 무방하다. 실무에서도 사용할 수 있는 정도의 수준이라고 말할 수 있다. 사실 이 예제는 이미 자바가 만들어 둔 java.util.concurrent.BlockingQueue를 사용했으면 되는 문제였다. 정확히 이 한정된 버퍼에 더 채울 수 있는 공간이 없으면 데이터 추가를 공간이 생길 때까지 차단해주고, 한정된 버퍼에 데이터가 없으면 데이터가 생길 때까지 데이터를 소비하는 것을 차단해주는 큐이다.
근데, 하나씩 풀어가면서 왜 이런 상황이 발생하고,왜 생산자 스레드 입장에서 데이터가 비워질때까지(한정된 버퍼에 데이터를 넣을 공간이 생길때까지) 단순히 기다리기만 한다면 다른 스레드들이 BLOCKED 상태에서 빠져나오지 못하는지, 왜 synchronized, wait(), notify()를 사용해야 하고이것들이 가진 한계가 무엇이길래 ReentrantLock, Condition을 사용하는지 이해해야 BlockingQueue라는 이미 잘 만들어진 큐를 사용할 자격이 생기는 것이다.
이제 위 내용을 다 이해했다. 그럼 이미 만들어놓은 기능인 BlockingQueue를 한번 사용해보자!
BlockingQueue
자바는 생산자 소비자 문제를 해결하기 위해 java.util.concurrent.BlockingQueue라는 특별한 멀티스레드 자료 구조를 제공한다.
이것은 이름 그대로 스레드를 차단(Blocking)할 수 있는 큐다.
데이터 추가 차단: 큐가 가득 차면 데이터 추가 작업을 시도하는 스레드는 공간이 생길때까지 차단된다.
데이터 획득 차단: 큐가 비어 있으면 획득 작업을 시도하는 스레드는 큐에 데이터가 들어올 때까지 차단된다.
BlockingQueue는 인터페이스이고, 다음과 같은 다양한 기능을 제공한다.
package java.util.concurrent;
public interface BlockingQueue<E> extends Queue<E> {
boolean add(E e);
boolean offer(E e);
void put(E e) throws InterruptedException;
boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
E take() throws InterruptedException;
E poll(long timeout, TimeUnit unit) throws InterruptedException;
boolean remove(Object o);
...
}
주요 메서드만 정리했다.
데이터 추가 메서드: add(), offer(), put(), offer(타임아웃)
데이터 획득 메서드: take(), poll(타임아웃), remove()
Queue를 상속 받는다. 큐를 상속 받았기 때문에 추가로 큐의 기능들도 사용할 수 있다.
보면 데이터 추가와 획득에서 메서드가 굉장히 많다는 것을 알 수 있다. 이게 왜 이렇게 여러개가 있는지도 이후에 설명한다.
BlockingQueue 인터페이스의 대표적인 구현체
ArrayBlockingQueue: 배열 기반으로 구현되어 있고, 버퍼의 크기가 고정되어 있다.
LinkedBlockingQueue: 링크 기반으로 구현되어 있고, 버퍼의 크기를 고정할 수도, 무한하게 사용할 수도 있다.
이제 BlockingQueue를 사용하도록 기존 코드를 변경해보자.
BoundedQueueV6_1
package thread.bounded;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class BoundedQueueV6_1 implements BoundedQueue {
private final BlockingQueue<String> queue;
public BoundedQueueV6_1(int max) {
this.queue = new ArrayBlockingQueue<String>(max);
}
@Override
public void put(String data) {
try {
queue.put(data);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
@Override
public String take() {
try {
return queue.take();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
@Override
public String toString() {
return queue.toString();
}
}
ArrayBlockingQueue를 사용한다.
BlockingQueue.put(data): 데이터를 추가한다.
BlockingQueue.take(): 데이터를 뽑아온다.
여기서 BlockingQueue.put(data) 메서드를 한번 어떻게 구현했는지 봐보자.
다음 코드가 java.util.concurrent.BlockingQueue를 구현한 ArrayBlockingQueue의 put() 메서드이다.
public void put(E e) throws InterruptedException {
Objects.requireNonNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
lock.unlock();
}
}
앞에서 만들어본 BoundedQueueV5와 굉장히 유사하게 생긴것을 알 수 있다. 다른건 lock.lock()이 아니라 lock.lockInterruptibly()를 사용했다 정도의 차이가 있다.
그러니까 결론은 자바에서 제공하는 멀티 스레드 용 자료구조가 이제 눈에 들어온다는 것이다.
이제 이 BoundedQueueV6_1를 사용해서 실행해보면 같은 결과를 얻을것이다.
BlockingQueue 기능 설명
아까 보니까 데이터를 추가하는 것도 많은 메서드가 있던것을 알 수 있었다. 왜 그럴까? 실무에서 멀티스레드를 사용할땐 응답성이 중요하다. 예를 들어, 대기 상태에 있어도 고객이 중지 요청을 하거나, 또는 너무 오래 대기한 경우 포기하고 빠져나갈 수 있는 방법이 필요하다.
생산자가 무언가 데이터를 생산하는데, 버퍼가 빠지지 않아서 너무 오래 대기해야 한다면 무한정 기다리는 것보다는 작업을 포기하고 사용자에게는 "죄송합니다. 현재 시스템에 문제가 있습니다. 나중에 다시 시도해주세요."라고 하는 것이 더 나은 선택일 것이다.
즉, 멀티스레드 세상에서는 정말 여러 상황과 결론이 만들어질 수 있다는 얘기고 그렇기에 이 BlockingQueue는 각 상황에 맞는 다양한 메서드를 제공하고 있다.
BlockingQueue의 다양한 기능
Operation
Throws Exception
Special Value
Blocks
Times Out
Insert
add(e)
offer(e)
put(e)
offer(e, time, unit)
Remove
remove()
poll()
take()
poll(time, unit)
Throws Exception (대기 시 예외를 터트림)
add(e): 지정된 요소를 큐에 추가하며, 큐가 가득 차면 IllegalStateException 예외를 던진다.
remove(): 큐에서 요소를 제거하며 반환한다. 큐가 비어 있으면 NoSuchElementException 예외를 던진다.
Special Value (대기 시 즉시 반환)
offer(e): 지정된 요소를 큐에 추가하려고 시도하며, 큐가 가득 차면 false를 반환한다.
poll(): 큐에서 요소를 제거하고 반환한다. 큐가 비어 있으면 null을 반환한다.
Blocks (대기)
put(e): 지정된 요소를 큐에 추가할 때까지 대기한다. 큐가 가득 차면 공간이 생길 때까지 대기한다.
take(): 큐에서 요소를 제거하고 반환한다. 큐가 비어 있으면 요소가 준비될 때까지 대기한다.
Times Out (시간만큼만 대기)
offer(e, time, unit): 지정된 요소를 큐에 추가하려고 시도하며, 지정된 시간 동안 큐가 비워지기를 기다리다가 시간이 초과되면 false를 반환한다.
poll(time, unit): 큐에서 요소를 제거하고 반환한다. 큐에 요소가 없다면 지정된 시간 동안 요소가 준비되기를 기다렸다가 시간이 초과되면 null을 반환한다.
참고로, BlockingQueue의 모든 대기, 시간 대기 메서드는 인터럽트를 제공한다. 대기하는 put(e), take()는 바로 위에서 예제로 직접 만들어보았다. 나머지도 하나씩 코드로 확인해보자.
BlockingQueue 즉시 반환
BoundedQueueV6_2
package thread.bounded;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import static util.MyLogger.log;
public class BoundedQueueV6_2 implements BoundedQueue {
private final BlockingQueue<String> queue;
public BoundedQueueV6_2(int max) {
this.queue = new ArrayBlockingQueue<String>(max);
}
@Override
public void put(String data) {
boolean result = queue.offer(data);
log("저장 시도 결과 = " + result);
}
@Override
public String take() {
return queue.poll();
}
@Override
public String toString() {
return queue.toString();
}
}
저장 시도 시 기다리는 시간을 1 나노초로 했기 때문에 그냥 뭐 안 기다리는 수준으로 기다리게 설정했다. 그러니까 실행 결과를 보면 저장 시도 결과는 false가 출력되고 저장을 마지막에 하지 못했으니 소비도 마지막 스레드는 할 수 없어 null이 반환됐다. 만약 기다리는 시간을 넉넉하게 잡고 awaitNanos(timeout)으로 잠시 대기하게 한 후 소비자 스레드가 들어와서 큐에 데이터를 소비해서 공간이 생긴후에도 지정한 시간을 지나지 않았다면 아마 데이터가 추가도, 데이터 소비도 정상적으로 될 것이다.
아, 참고로 여기서 awaitNanos(timeout)은 그냥 내가 이렇게 사용하자가 아니라 실제로 BlockingQueue가 offer(e, time, unit) 메서드에서 사용하고 있는 것이다. 아래 실제 구현 코드 참고!
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
Objects.requireNonNull(e);
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length) {
if (nanos <= 0L)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
BlockingQueue예외 터트리기
BoundedQueueV6_4
package thread.bounded;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import static util.MyLogger.log;
public class BoundedQueueV6_4 implements BoundedQueue {
private final BlockingQueue<String> queue;
public BoundedQueueV6_4(int max) {
this.queue = new ArrayBlockingQueue<String>(max);
}
@Override
public void put(String data) {
queue.add(data);
}
@Override
public String take() {
return queue.remove();
}
@Override
public String toString() {
return queue.toString();
}
}
실행 결과를 보면 데이터를 추가할 때 공간이 없으면 IllegalStateException을, 데이터 소비할 때 소비할 데이터가 없으면 NoSuchElementException을 발생시키고 있음을 알 수 있다.
참고로, 지금 V6_1, V6_2, V6_3, V6_4 모두 BoundedQueue 인터페이스를 구현해서 만들고 있는데 이래야만 가능한게 아니라 기존에 작성한 코드들(producerFirst, consumerFirst, ...)이 전부 BoundedQueue를 의존하고 있기 때문에 이 인터페이스를 구현한 구현체를 만들어서 그 안에서 BlockingQueue를 사용하는 식으로 만든거고 그게 아니라면 그냥 바로 BlockingQueue를 사용해도 상관없다!
정리
이렇듯, 기존에 아주 아주 잘 만들어진 BlockingQueue를 사용하면 훨씬 더 다양한 상황을 더 유연하게 대처할 수 있음을 알게됐다.
무작정 기다릴수도, 정해진 시간만큼만 기다릴수도, 아예 바로 결과를 반환할수도, 예외를 터트릴수도 있다. 그리고 내부가 어떻게 구현됐는지도 이제 이해할 수 있는 레벨이 됐다!
생산자 소비자 문제는 멀티스레드 프로그래밍에서 자주 등장하는 동시성 문제 중 하나로, 여러 스레드가 동시에 데이터를 생산하고 소비하는 상황을 다룬다.
멀티스레드의 핵심을 제대로 이해하려면 반드시 생산자 소비자 문제를 이해하고, 올바른 해결 방안도 함께 알아두어야 한다. 생산자 소비자 문제를 제대로 이해하면 멀티스레드를 제대로 이해했다고 볼 수 있다. 그만큼 중요한 내용이다.
생산자(Producer): 데이터를 생성하는 역할을 한다. 예를 들어, 파일에서 데이터를 읽어오거나 네트워크에서 데이터를 받아오는 스레드가 생산자 역할을 할 수 있다.
위 프린터 예제에서 사용자의 입력을 프린터 큐에 전달하는 스레드가 생산자의 역할이다.
소비자(Consumer): 생성된 데이터를 사용하는 역할을 한다. 예를 들어, 데이터를 처리하거나 저장하는 스레드가 소비자 역할을 할 수 있다.
위 프린터 예제에서 프린터 큐에 전달된 데이터를 받아서 출력하는 스레드가 소비자 역할이다.
버퍼(Buffer): 생산자가 생성한 데이터를 일시적으로 저장하는 공간이다. 이 버퍼는 한정된 크기를 가지며, 생산자와 소비자가 이 버퍼를 통해 데이터를 주고 받는다.
위 프린터 예제에서 프린터 큐가 버퍼 역할이다.
그럼 이게 왜 문제가 된다는 것일까?
문제 상황
생산자가 너무 빠를 때: 버퍼가 가득 차서 더 이상 데이터를 넣을 수 없을 때까지 생산자가 데이터를 생성한다. 버퍼가 가득 찬 경우 생산자는 버퍼에 빈 공간이 생길 때까지 기다려야 한다.
소비자가 너무 빠를 때: 버퍼가 비어서 더 이상 소비할 데이터가 없을 때까지 소비자가 데이터를 처리한다. 버퍼가 비어있을 때 소비자는 버퍼에 새로운 데이터가 들어올 때까지 기다려야 한다.
예를 들면, 초밥집에 가서 주방장님이 초밥을 하나씩 만들어서 서빙 카운터에 초밥을 하나씩 올려 놓는데 더 이상 올려 놓을 곳이 없어 초밥을 만들지 못하는 경우가 생산자가 너무 빠른 경우이고, 반대로 서빙 카운터에 초밥을 내려놓자마자 손님이 초밥을 다 먹어버려서 음식이 나올때까지 기다려야 하는 경우가 소비자가 너무 빠른 경우이다. 이때 서빙 카운터는 버퍼라고 볼 수 있다.
이 문제는 다음 두 용어로도 불린다.
생산자 소비자 문제(producer-consumer problem): 생산자 소비자 문제는, 생산자 스레드와 소비자 스레드가 특정 자원을 함께 생산하고 소비하면서 발생하는 문제이다.
한정된 버퍼 문제(bounded-buffer problem): 이 문제는 결국 중간에 있는 버퍼의 크기가 한정되어 있기 때문에 발생한다. 따라서 한정된 버퍼 문제라고도 한다.
put(data): 버퍼에 데이터를 보관한다. (생산자 스레드가 호출하고, 데이터를 생산한다.)
take(): 버퍼에 보관된 값을 가져간다. (소비자 스레드가 호출하고, 데이터를 소비한다.)
BoundedQueueV1
package thread.bounded;
import java.util.ArrayDeque;
import java.util.Queue;
import static util.MyLogger.log;
public class BoundedQueueV1 implements BoundedQueue {
private final Queue<String> queue = new ArrayDeque<>();
private final int max;
public BoundedQueueV1(int max) {
this.max = max;
}
@Override
public synchronized void put(String data) {
if (queue.size() == max) {
log("[put] 큐가 가득 찼습니다. 버립니다." + data);
return;
}
queue.offer(data);
}
@Override
public synchronized String take() {
if (queue.isEmpty()) {
return null;
}
return queue.poll();
}
@Override
public String toString() {
return queue.toString();
}
}
BoundedQueueV1: 한정된 버퍼 역할을 하는 가장 단순한 구현체이다. 이후에 버전이 점점 올라가면서 코드를 개선한다.
Queue, ArrayDeque: 데이터를 중간에 보관하는 버퍼로 큐(Queue)를 사용한다. 구현체로는 ArrayDeque를 사용한다.
int max: 한정된(Bounded) 버퍼이므로, 버퍼에 저장할 수 있는 최대 크기를 지정한다.
put(): 큐에 데이터를 저장한다. 큐가 가득 찬 경우, 더는 데이터를 보관할 수 없으므로 데이터를 버린다.
take(): 큐의 데이터를 가져간다. 큐에 데이터가 없는 경우 null을 반환한다.
toString(): 버퍼 역할을 하는 queue 정보를 출력한다.
주의! 원칙적으로는 toString()에도 synchronized를 적용해야 맞다. 그래야 toString()을 통한 조회 시점에도 정확한 데이터를 조회할 수 있다. 예를 들어, toString()을 호출하는 시점에 모니터 락을 사용하지 않으면 다른 스레드가 put을 하는 동시에 이 toString()을 호출해서 실제 데이터와 다른 데이터가 출력될 수 있으니까. 그러나, 예제 코드를 단순하게 유지하고 목적에 부합한 결과를 출력하기 위해 의도적으로 synchronized를 넣지 않았다.
임계 영역
여기서 핵심 공유 자원은 바로 queue이다. 여러 스레드가 접근할 예정이므로 synchronized를 사용해서 한번에 하나의 스레드만 put() 또는 take()를 실행할 수 있도록 안전한 임계 영역을 만든다.
위 producerFirst 메서드와 완전히 동일하고 먼저 실행되는 메서드가 startProducer가 아니라 startConsumer라는 차이만 있다.
private static void startProducer(BoundedQueue queue, List<Thread> threads) {
System.out.println();
log("생산자 시작");
for (int i = 1; i <= 3; i++) {
Thread producer = new Thread(new ProducerTask(queue, "data" + i), "producer" + i);
threads.add(producer);
producer.start();
sleep(100);
}
}
3개의 생산자 스레드를 만들고 각 스레드들을 실행한다. 로그 출력을 조금 알아보기 쉽게 하기 위해 sleep(100)을 추가했다. 의도적으로 추가한 것이고 실제 업무라면 없는게 맞다. 이렇게 잠시 0.1초 동안 대기하게 만들어 두면 로그로 producer1 → producer2 → producer3 순으로 이쁘게 출력될 것이다.
만들어지는 각각의 생산자 스레드 모두 파라미터로 넘어온 threads에 추가된다.
private static void startConsumer(BoundedQueue queue, List<Thread> threads) {
System.out.println();
log("소비자 시작");
for (int i = 1; i <= 3; i++) {
Thread consumer = new Thread(new ConsumerTask(queue), "consumer" + i);
threads.add(consumer);
consumer.start();
sleep(100);
}
}
위 startProducer와 완전히 동일한 코드이지만 생산자 스레드를 만드는 게 아니라 소비자 스레드를 만들어 실행한다.
public static void main(String[] args) {
//1. BoundedQueue 선택
BoundedQueue queue = new BoundedQueueV1(2);
//2. 생산자, 소비자 실행 순서 선택, 반드시 하나만 선택!
producerFirst(queue);
//consumerFirst(queue);
}
public static void main(String[] args) {
//1. BoundedQueue 선택
BoundedQueue queue = new BoundedQueueV1(2);
//2. 생산자, 소비자 실행 순서 선택, 반드시 하나만 선택!
//producerFirst(queue);
consumerFirst(queue);
}
consumer1, consumer2 스레드가 모두 한번씩 실행되면서 큐에 있는 데이터를 소비했다. 이제 consumer3이 실행될 차례다.
consumer3이 데이터를 소비하려고 했지만 데이터가 없다. 아무것도 가져올 수 없다. 아무런 데이터도 얻지 못한 채 쓸쓸히 돌아가게 된다.
이 시점이 마지막 시점이다. 결과적으로 버퍼가 가득차서 p3 스레드가 생성한 데이터는 버려졌고, c3 스레드는 버퍼에 아무런 데이터도 없기 때문에 어떠한 데이터도 얻지 못한 채 돌아갔다. 이 부분에서 문제점이 보인다. 만약 이걸 해결하려면 가장 간단한 대안으로는 기다리는 대안이 있을 것 같다. p3 입장에서는 버퍼에 남는 공간이 생길 때 까지 기다렸다가 데이터를 넣고, c3 입장에서는 버퍼에 데이터가 생길 때 까지 기다렸다가 생기면 데이터를 가져오는 식으로 말이다.
생산자 소비자 문제 - 소비자 우선 결과 분석
이번엔 반대로 소비자 먼저 실행한 결과를 분석해보자.
c1, c2, c3 순으로 스레드가 큐에서 데이터를 가져오려고 시도하지만, 데이터가 아무것도 없다. 결국 모든 소비자 스레드는 빈털털이로 돌아오게 된다.
이제 생산자 스레드가 하나씩 생성되면서 큐에 데이터를 넣는다.
마지막 p3 스레드는 역시나 버퍼에 데이터가 꽉 차있기 때문에 데이터를 넣지 못하고 버리게 된다.
이게 마지막 시점의 모습이다. 이부분에서도 역시나 문제점이 많다. 소비자 스레드는 모든 스레드가 다 데이터를 가지지 못했고, 생산자 스레드는 결국 p3 스레드는 여전히 데이터를 추가하지 못했다.
총평을 내려보자면,
생산자 스레드 먼저 실행의 경우, p3이 보관하는 데이터는 버려지고 c3은 데이터를 받지 못한다.
소비자 스레드 먼저 실행의 경우, c1, c2, c3 모두 데이터를 받지 못하고 p3이 보관하는 데이터는 버려진다.
결국 생산자 소비자 문제는 이런 결과를 발생시킨다. 어디 한쪽이 너무 빠르면 뭐가 됐건 문제가 생기고 그 근본 원인은 사실 버퍼의 사이즈다. 버퍼가 가득 찬 경우는 생산자 입장에서 버퍼에 여유가 생길 때 까지 조금만 기다리면 되는데 기다리지 못하고 데이터를 버리는 것이 아쉽고, 버퍼가 빈 경우는 소비자 입장에서 버퍼에 데이터가 채워질 때 까지 조금만 기다리면 되는데 기다리지 못하고 데이터를 못 얻는게 아쉽다.
예제 변경 - 대기하는 코드로 바꿔보기
가장 간단한 해결 방법은 기다리는 것이다! 한번 기다리게 해서 문제를 해결해보자!
BoundedQueueV2
package thread.bounded;
import java.util.ArrayDeque;
import java.util.Queue;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class BoundedQueueV2 implements BoundedQueue {
private final Queue<String> queue = new ArrayDeque<>();
private final int max;
public BoundedQueueV2(int max) {
this.max = max;
}
@Override
public synchronized void put(String data) {
while (queue.size() == max) {
log("[put] 큐가 가득 찼습니다. 생산자는 대기합니다.");
sleep(1000);
}
queue.offer(data);
}
@Override
public synchronized String take() {
while (queue.isEmpty()) {
log("[take] 큐에 데이터가 없습니다. 소비자는 대기합니다");
sleep(1000);
}
return queue.poll();
}
@Override
public String toString() {
return queue.toString();
}
}
BoundedQueueV2를 만들고 여기서 put(data), take() 메서드에서 데이터가 꽉차거나 없다면 빠져나가는게 아니라 기다리는 것이다.
그래서, put(String data)를 보면 큐의 사이즈가 꽉 찼다면 1초 정도 대기한 후 다시 확인해보는 것이다. 데이터가 빠져나갈 때까지.
또한, take()도 큐에 아무런 데이터가 없다면 1초 정도 대기한 후 다시 확인해보는 것이다. 데이터가 들어올 때까지.
c1 스레드가 버퍼에 데이터가 아무것도 없기 때문에 데이터가 들어올때까지 기다리고 있어서 락을 놔주고 있지 않다. 그 결과 어떤 스레드도 접근이 불가능하다. 그래서 c1 스레드를 제외한 모든 스레드가 다 BLOCKED 상태가 돼버렸다.
"어? 그럼 sleep()말고 yield()를 사용해서 다른 스레드에게 양보하면 되지 않아요?"
진짜 좋은 생각이다. 한번 그렇게 해볼까? 다음 코드를 보자.
@Override
public synchronized void put(String data) {
while (queue.size() == max) {
log("[put] 큐가 가득 찼습니다. 생산자는 대기합니다.");
// sleep(1000);
Thread.yield();
}
queue.offer(data);
}
이번엔 yield()를 사용해서 욕심 부리지 말고 다른 스레드에게 양보하자! 과연 잘 될까?
생산자 스레드 먼저 실행 코드로 실행해보면 다음과 같은 결과를 얻는다.
...
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
2024-07-24 14:18:59.595 [producer3] [put] 큐가 가득 찼습니다. 생산자는 대기합니다.
...
결국 같은 현상이다. 대신 1초의 대기 시간이 사라지니까 너무 빠르게 출력이 된 것 뿐이다. 왜 같은 현상이 일어날까?
Thread.yield()는 synchronized의 락을 반납하는게 아니다.
그저 다른 스레드에게 CPU 사용을 양보하는 것 뿐이지 synchronized의 락은 여전히 본인이 가지고 있다. 언제까지? synchronized 블록이 끝날 때까지. 그래서 결국 다른 스레드들은 같은 모니터 락이 필요한 synchronized 메서드나 블록에 접근하지 못하게 된다.
그럼 이 생산자 소비자 문제는 도대체 어떻게 해결할까?
양보하는게 맞다! 근데 양보를 할 때 락도 양보하면 된다.
자바의 Object.wait(), Object.notify()를 사용하면, 락을 가지고 대기하는 스레드가 대기하는 동안 다른 스레드에게 락을 양보할 수 있다. 바로 알아보자!
Object.wait(), Object.notify()
저번에도 말했지만, 자바는 처음부터 멀티스레드를 고려하고 탄생한 언어이다. 앞서 설명한 synchronized를 사용한 임계 영역 안에서 락을 가지고 무한 대기하는 문제는 흥미롭게도 Object 클래스에 해결 방안이 있다. Object 클래스는 이런 문제를 해결할 수 있는 wait(), notify()라는 메서드를 제공한다. Object는 모든 자바 객체의 부모이기 때문에, 여기 있는 기능들은 모두 자바 언어의 기본 기능이라 생각하면 된다.
Object.wait()
현재 스레드가 가진 락을 반납하고 대기(WAITING)한다.
현재 스레드를 대기 상태로 전환한다. 이 메서드는 현재 스레드가 synchronized 블록이나 메서드에서 락을 소유하고 있을 때만 호출할 수 있다. 호출한 스레드는 락을 반납하고, 다른 스레드가 해당 락을 획득할 수 있도록 한다. 이렇게 대기 상태로 전환된 스레드는 다른 스레드가 notify() 또는 notifyAll()을 호출할 때까지 대기 상태를 유지한다.
Object.notify()
대기 중인 스레드 중 하나를 깨운다.
이 메서드는 synchronized 블록이나 메서드에서 호출되어야 한다. 깨운 스레드는 락을 다시 획득할 기회를 얻게 된다. 만약, 대기 중인 스레드가 여러개라면 그 중 하나만이 깨워지게 된다.
Object.notifyAll()
대기 중인 모든 스레드를 깨운다.
이 메서드 역시 synchronized 블록이나 메서드에서 호출되어야 한다. 모든 대기중인 스레드가 락을 획득할 수 있는 기회를 얻게 된다. 이 방법은 모든 스레드를 깨워야 할 필요가 있는 경우에 유용하다.
wait(), notify() 메서드를 적절하게 사용하면, 멀티스레드 환경에서 발생할 수 있는 문제를 효율적으로 해결할 수 있다. 이 기능을 활용해서 스레드가 락을 가지고 임계 영역안에서 무한 대기하는 문제를 해결해보자!
BoundedQueueV3
package thread.bounded;
import java.util.ArrayDeque;
import java.util.Queue;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class BoundedQueueV3 implements BoundedQueue {
private final Queue<String> queue = new ArrayDeque<>();
private final int max;
public BoundedQueueV3(int max) {
this.max = max;
}
@Override
public synchronized void put(String data) {
while (queue.size() == max) {
log("[put] 큐가 가득 찼습니다. 생산자는 대기합니다.");
try {
wait(); // RUNNABLE -> WAITING, 락 반납
log("[put] 생산자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
queue.offer(data);
log("[put] 생산자 데이터 저장, notify() 호출");
notify(); // 대기 스레드에게 WAIT -> BLOCKED
}
@Override
public synchronized String take() {
while (queue.isEmpty()) {
log("[take] 큐에 데이터가 없습니다. 소비자는 대기합니다");
try {
wait();
log("[take] 소비자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
String data = queue.poll();
log("[take] 소비자 데이터 획득, notify() 호출");
notify(); // 대기 스레드에게 WAIT -> BLOCKED
return data;
}
@Override
public String toString() {
return queue.toString();
}
}
이제 V3이다. 앞서 사용한 sleep() 코드는 제거하고, Object.wait()을 사용하자. Object는 모든 클래스의 부모이므로 자바의 모든 객체는 해당 기능을 사용할 수 있다.
@Override
public synchronized void put(String data) {
while (queue.size() == max) {
log("[put] 큐가 가득 찼습니다. 생산자는 대기합니다.");
try {
wait(); // RUNNABLE -> WAITING, 락 반납
log("[put] 생산자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
queue.offer(data);
log("[put] 생산자 데이터 저장, notify() 호출");
notify(); // 대기 스레드에게 WAIT -> BLOCKED
}
synchronized를 통해 임계 영역을 설정한다. 생산자 스레드는 락 획득을 시도한다.
락을 획득한 생산자 스레드는 반복문을 통해서 큐에 빈 공간이 생기는지 주기적으로 체크한다. 만약, 빈 공간이 없다면 Object.wait()을 사용해서 대기한다. 참고로 대기할 때 락을 반납하고 대기한다. 그리고 대기 상태에서 깨어나면 다시 반복문에서 큐의 빈 공간을 체크한다.
wait()을 호출해서 대기하는 경우 RUNNABLE → WAITING 상태가 된다.
생산자가 데이터를 큐에 저장하고 나면 notify()를 통해 대기하는 스레드에게 저장된 데이터가 있다고 알려주어야 한다. 예를 들어서 큐에 데이터가 없어서 대기하는 소비자 스레드가 있다고 가정하자. 이때 notify()를 호출하면 소비자 스레드는 깨어나서 저장된 데이터를 가져갈 수 있다.
@Override
public synchronized String take() {
while (queue.isEmpty()) {
log("[take] 큐에 데이터가 없습니다. 소비자는 대기합니다");
try {
wait();
log("[take] 소비자 깨어남");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
String data = queue.poll();
log("[take] 소비자 데이터 획득, notify() 호출");
notify(); // 대기 스레드에게 WAIT -> BLOCKED
return data;
}
synchronized를 통해 임계 영역을 설정한다. 소비자 스레드는 락 획득을 시도한다.
락을 획득한 소비자 스레드는 반복문을 사용해서 큐에 데이터가 있는지 주기적으로 체크한다. 만약, 데이터가 없다면 Object.wait()을 사용해서 대기한다. 참고로 대기할 때 락을 반납하고 대기한다. 그리고 대기 상태에서 깨어나면, 다시 반복문에서 큐에 데이터가 있는지 체크한다.
대기하는 경우 RUNNABLE → WAITING 상태가 된다.
소비자가 데이터를 획득하고 나면 notify()를 통해 대기하는 생산자 스레드에게 큐에 저장할 여유 공간이 생겼다고알려주어야 한다. 예를 들어, 큐에 데이터가 꽉차서 데이터를 넣지 못해 대기하는 생산자 스레드가 있다고 가정하자. 이때 notify()를 호출하면 생산자 스레드는 깨어나서 저장된 데이터를 획득할 수 있다.
wait()으로 대기 상태에 빠진 스레드는 notify()를 사용해야 깨울 수 있다. 생산자는 생산을 완료하면 notify()로 대기하는 스레드를 깨워서 생산된 데이터를 가져가게 하고, 소비자는 소비를 완료하면 notify()로 대기하는 스레드를 깨워서 데이터를 생산하라고 하면 된다. 여기서 중요한 핵심은 wait()을 호출해서 대기 상태에 빠질 땐 락을 반납하고 대기 상태에 빠진다는 것이다. 대기 상태에 빠지면 어차피 아무일도 하지 않으므로 락도 필요하지 않다.
V3로 변경하고 생산자 먼저 실행 코드로 변경 하기
public static void main(String[] args) {
//1. BoundedQueue 선택
BoundedQueue queue = new BoundedQueueV3(2);
//2. 생산자, 소비자 실행 순서 선택, 반드시 하나만 선택!
producerFirst(queue);
//consumerFirst(queue);
}
public static void main(String[] args) {
//1. BoundedQueue 선택
BoundedQueue queue = new BoundedQueueV3(2);
//2. 생산자, 소비자 실행 순서 선택, 반드시 하나만 선택!
//producerFirst(queue);
consumerFirst(queue);
}
로그만 보면 결국 잘 저장하고 잘 사용한것 같아 보인다. 근데 로그만으로는 이해하기 쉽지 않다. 그림으로 하나씩 분석해보자!
wait(), notify() 생산자 우선 분석
우선 못보던게 하나 생겼다. 스레드 대기 집합
스레드 대기 집합 (wait set)
synchronized 임계 영역 안에서 Object.wait()을 호출하면 스레드는 대기 상태에 들어간다. 이렇게 대기 상태에 들어간 스레드를 관리하는 것을 대기 집합이라 한다. 참고로 모든 객체는 각자 자기의 대기 집합과 모니터 락을 가지고 있다. 그리고 이 둘은 한 쌍으로 사용된다. 따라서 락을 획득한 객체의 대기 집합을 사용해야 한다. 여기서는 BoundedQueue(x001) 구현 인스턴스의 락과 대기 집합을 사용한다.
synchronized를 메서드에 적용하면 해당 인스턴스의 락을 사용한다. 여기서는 BoundedQueue(x001)의 구현체이다.
wait() 호출은 앞에 this를 생략할 수 있다. this는 해당 인스턴스를 뜻한다. 여기서는 BoundedQueue(x001)의 구현체이다.
이제 순서대로 흐름을 분석해보자!
p1이 락을 획득하고 큐에 데이터를 저장한다.
큐에 데이터가 추가 되었기 때문에 스레드 대기 집합에 이 사실을 알려야 한다. (코드 흐름이 그렇다)
notify()를 호출하면 스레드 대기 집합에서 대기하는 스레드 중 하나를 깨운다.
현재 대기 집합에 스레드가 없으므로 아무일도 발생하지 않는다. 만약 소비자 스레드가 대기 집합에 있었다면 깨어나서 큐에 들어있는 데이터를 소비했을 것이다.
p1은 할일을 다 끝내고 락을 반납한다.
p2도 큐에 데이터를 저장하고 생산을 완료했다.
p3가 데이터를 생산하려고 하는데, 큐가 가득 찼다. wait()을 호출한다.
wait()을 호출하면 락을 반납한다.
wait()을 호출하면 스레드의 상태가 RUNNABLE → WAITING으로 변경된다.
wait()을 호출하면 스레드 대기 집합에서 관리된다.
스레드 대기 집합에서 관리되는 스레드는 이후에 다른 스레드가 notify()를 통해 스레드 대기 집합에 신호를 주면 깨어날 수 있다.
이제 소비자 스레드들이 움직일 차례가 됐다.
c1이 데이터를 획득했다. 그래서 큐에 데이터를 보관할 빈자리 생겼다.
c1은 notify()를 호출한다 (코드 흐름이 그렇다)
스레드 대기 집합에 있는 p3를 깨운다.
스레드 대기 집합은 notify() 신호를 받으면 대기 집합에 있는 스레드 중 하나를 깨운다.
그런데 대기 집합에 있는 스레드가 깨어난다고 바로 작동하는 것이 아니다. 왜냐하면 이 스레드는 여전히 임계 영역 안에 있기 때문이다.
임계 영역에 있는 코드를 실행하려면 가장 먼저 락이 필요하다. p3는 대기 집합에서는 나가지만 여전히 임계 영역에 있으므로 락을 획득하기 위해 BLOCKED 상태로 대기한다. p3: WAITING → BLOCKED
참고로 이때 임계 영역의 코드를 처음으로 돌아가서 실행하는 게 아니다. 대기 집합에 들어오게 된 wait()을 호출한 부분부터 다시 실행된다. 락을 획득하면 wait() 이후의 코드를 실행한다.
c1은 데이터 소비를 완료하고 락을 반납하고 임계 영역을 빠져나간다.
p3가 락을 획득한다.
BLOCKED → RUNNABLE
wait() 코드에서 대기했기 때문에 이후의 코드를 실행한다.
data3을 큐에 저장한다.
notify()를 호출한다. 데이터를 저장했기 때문에 혹시 스레드 대기 집합에 소비자가 대기하고 있다면 소비자를 하나 깨워줘야 한다. 물론 지금은 대기 집합에 스레드가 없기 때문에 아무일도 일어나지 않는다.
p3는 락을 반납하고 임계 영역을 빠져나간다.
이제 c2, c3가 하나씩 데이터를 원래대로 가져갈 것이다.
c2, c3가 실행됐고 데이터가 있으므로 둘 다 데이터를 소비하고 완료한다.
둘 다 notify()를 호출했지만 대기 집합에 스레드가 없으므로 아무일도 일어나지 않는다.
wait(), notify() 덕분에 스레드가 락을 놓고 대기하고, 또 대기하는 스레드를 필요한 시점에 깨울 수 있었다. 생산자 스레드가 큐가 가득차서 대기해도 소비자 스레드가 큐의 데이터를 소비하고 나면 알려주기 때문에 최적의 타이밍에 깨어나서 데이터를 생산할 수 있었다.
덕분에 최종 결과를 보면 p1, p2, p3 모두 데이터를 정상 생산하고 c1, c2, c3 모두 데이터를 정상 소비할 수 있었다.
wait(), notify()소비자 우선 분석
이제 소비자 우선 코드로 시작해보자. 최초의 상태이다.
c1이 락을 얻고 임계 영역에 들어왔지만, 데이터가 없다. wait()을 호출하고 대기 집합에 대기하게 된다.
큐에 데이터가 없기 때문에 c1, c2, c3 모두 스레드 대기 집합에서 대기하게 된다.
이후에 생산자가 큐에 데이터를 생산하면 notify()를 통해 이 스레드들을 하나씩 깨워서 데이터를 소비할 수 있을것이다.
p1은 락을 획득하고 큐에 데이터를 생산한다. 큐에 데이터가 있기 때문에 소비자를 하나 깨울 수 있다. notify()를 통해 스레드 대기 집합에 이 사실을 알려준다.
notify()를 받은 스레드 대기 집합은 스레드 중에 하나를 깨운다.
여기서 c1, c2, c3 중에 어떤 스레드가 깨어날지는 알 수 없다.
어떤 스레드가 깨워질지는 JVM 스펙에 명시되어 있지 않다. 따라서 JVM 버전 및 환경에 따라 달라진다.
그런데 대기 집합에 있는 스레드가 깨어난다고 바로 작동하는 것은 아니다. 깨어난 스레드는 여전히 임계 영역 안에 있다.
임계 영역 안에 있는 코드를 실행하려면 먼저 락이 필요하다. 대기 집합에서는 나가지만 여전히 임계 영역 안에 있으므로 락을 획득하기 위해 BLOCKED 상태로 대기한다. WAITING → BLOCKED
p1이 락을 반납하고 임계 영역에서 나간다.
c1은 락을 획득한다.
c1은 락을 획득하고 임계 영역 안에서 실행되며 데이터를 획득한다.
c1이 데이터를 획득했으므로 큐에 데이터를 넣을 공간이 있다는 것을 대기 집합에 알려준다. 만약 대기 집합에 생산자 스레드가 대기하고 있다면 큐에 데이터를 넣을 수 있을 것이다.
c1이 notify()로 스레드 대기 집합에 알렸지만, 생산자 스레드가 아니라 소비자 스레드만 있다. 따라서 의도와는 다르게 소비자 스레드인 c2가 대기상태에서 깨어난다. (물론 대기 집합에 있는 어떤 스레드가 깨어날지는 알 수 없다. 여기서는 c2가 깨어난다고 가정한다. 심지어 생산자와 소비자 스레드가 함께 대기 집합에 있어도 어떤 스레드가 깨어날지는 알 수 없다.)
c1은 작업을 완료한다.
c1이 c2를 깨웠지만, 문제가 있다. 바로 큐에 데이터가 없다는 것이다.
c2는 락을 획득하고 큐에 데이터를 소비하려고 시도한다. 그런데 큐에 데이터가 없으므로 c2는 결국 wait()을 호출해서 다시 대기 상태로 변하며 대기 집합에 들어간다.
이처럼 소비자인 c1이 같은 소비자인 c2를 깨우는것은 상당히 비효율적이다.
c1 입장에서 c2를 깨우게 되면 아무 일도 하지 않고 그냥 다시 스레드 대기 집합에 들어갈 수 있다. 결과적으로 CPU만 사용하고, 아무 일도 하지 않은 상태로 다시 대기 상태가 되어버린다.
그렇다고 c1이 스레드 대기 집합에 있는 어떤 스레드를 깨울지 선택할 수는 없다. notify()는 스레드 대기 집합에 있는 스레드 중 임의의 하나를 깨울뿐이다.
물론 이게 비효율적이라는 것이지 문제가 되는 것은 아니다. 결과에는 아무런 문제가 없다. 살짝 돌아갈 뿐이다.
p2가 락을 획득하고, 데이터를 저장한 다음에 notify()를 호출한다. 데이터가 있으므로 소비자 스레드가 깨어난다면 데이터를 소비할 수 있다.
스레드 대기 집합에 있는 c3가 깨어난다. 참고로 어떤 스레드가 깨어날지는 알 수 없다.
c3는 임계 영역 안에 있으므로 락을 획득하기 위해 대기(BLOCKED) 한다.
p2가 작업을 끝마치고 락을 반납하고 나간다.
c3는 락을 획득하고 BLOCKED → RUNNABLE 상태가 된다.
c3는 데이터를 획득한 다음에 notify()를 통해 스레드 대기 집합에 알린다. 큐에 여유 공간이 생겼기 때문에 생산자 스레드가 대기 중이라면 데이터를 생산할 수 있다.
notify()를 호출했지만, 스레드 대기 집합에는 소비자인 c2만 존재한다.
c2가 깨어나지만 임계 영역 안에 있으므로 락을 기다리는 BLOCKED 상태가 된다.
c3는 락을 반납하고 임계 영역을 나간다.
c2가 락을 획득하고, 큐에서 데이터를 획득하려 하지만 데이터가 없다.
c2는 다시 wait()을 호출해서 대기(WAITING)상태에 들어가고, 다시 대기 집합에서 관리된다.
물론 c2의 지금 이 사이클은 CPU 자원만 소모하고 다시 대기 집합에 들어갔기 때문에 비효율적이다.
만약 소비자인 c3 입장에서 생산자, 소비자 스레드를 선택해서 깨울 수 있다면, 소비자인 c2를 깨우지는 않았을 것이다. 하지만 notify()는 이런 선택을 할 수 없다.
p3가 락을 얻고 데이터를 저장한다. notify()를 통해 스레드 대기 집합에 알린다.
스레드 대기 집합에는 소비자 c2가 있으므로 생산한 데이터를 잘 소비할 수 있다.
c2가 notify()를 통해 깨어나고 BLOCKED 상태로 대기하고 있다가 락을 획득하면 큐에 데이터를 잘 소비해서 임계영역을 빠져나오고 종료된다.
정리를 하자면
최종 결과를 보면 p1, p2, p3 모두 데이터를 잘 생산하고 c1, c2, c3 모두 데이터를 잘 소비했다. 하지만 소비자인 c1이 같은 소비자인 c2, c3를 깨울 수 있었다. 이 경우 큐에 데이터가 없을 가능성이 있다. 이땐 깨어난 소비자 스레드가 CPU 자원만 소모하고 다시 대기 집합에 들어갔기 때문에 비효율적이다.
만약, 소비자인 c1 입장에서 생산자, 소비자 스레드를 선택해서 깨울 수 있다면, 소비자인 c2를 깨우지는 않았을 것이다. 예를 들어 소비자는 생산자만 깨우고, 생산자는 소비자만 깨울 수 있다면 더 효율적으로 작동할 수 있을 것 같다. 하지만 notify()는 이런 선택을 할 수 없다.
비효율적이지만 문제는 없다. 좀 돌아갈 뿐이다.
Object - wait(), notify() 한계
지금까지 살펴본 Object.wait(), Object.notify() 방식은 스레드 대기 집합 하나에 생산자, 소비자 스레드를 모두 관리한다. 그리고 notify()를 호출할 때 임의의 스레드가 선택된다. 따라서 앞서 살펴본 것 처럼 큐에 데이터가 없는 상황에 소비자가 같은 소비자를 깨우는 비효율이 발생할 수 있다. 또는 큐에 데이터가 가득 차있는데 생산자가 같은 생산자를 깨우는 비효율도 발생할 수 있다.
다음과 같은 상황을 연출해보자.
큐에 dataX가 보관되어 있다.
스레드 대기 집합에는 다음 스레드가 대기하고 있다.
소비자: c1, c2, c3
생산자: p1, p2, p3
p0 스레드가 data0 생산을 시도한다.
p0 스레드가 실행되면서 data0을 큐에 저장한다. 이때 큐에 데이터가 가득찬다.
notify()를 통해 대기 집합의 스레드를 하나 깨운다.
만약, notify()의 결과로 소비자 스레드가 깨어나게 되면 소비자 스레드는 큐의 데이터를 획득하고 완료된다.
그러나 notify()의 결과로 생산자 스레드를 깨우게 되면, 이미 큐에 데이터는 가득 차 있다. 따라서 데이터를 생산하지 못하고 다시 대기 집합으로 이동하는 비효율이 발생한다.
이번엔 반대의 경우로 소비자에 대해서도 이야기 해보자. 아래와 같은 상황이 있다.
c0 스레드가 실행되고 data0을 획득한다.
이제 큐에 데이터는 비어있게 된다.
c0 스레드는 notify()를 호출한다.
스레드 대기 집합에서 소비자 스레드가 깨어나면 큐에 데이터가 없기 때문에 다시 대기 집합으로 이동하는 비효율이 발생한다.
결국, 같은 종류의 스레드를 깨울 때 비효율이 발생한다.
이 내용을 통해서 알 수 있는 사실은 생산자가 생산자를 깨우거나, 소비자가 소비자를 깨울 때 비효율이 발생한다. 생산자가 소비자를 깨우고 반대로 소비자가 생산자를 깨운다면 이런 비효율은 발생하지 않을 것이다.
또 하나의 문제가 있다. 바로 스레드 기아(thread starvation) 문제점이다.
notify()의 또 다른 문제점으로는 어떤 스레드가 깨어날 지 알 수 없기 때문에 발생할 수 있는 스레드 기아 문제가 있다.
notify()가 어떤 스레드를 깨울지는 알 수 없다. 최악의 경우 c1 - c5 스레드가 반복해서 깨어날 수 있다.
c1 - c5 스레드가 깨어나도 큐에 소비할 데이터가 없다. 따라서 다시 스레드 대기 집합에 들어간다.
notify()로 다시 깨우는데 어떤 스레드를 깨울지 알 수 없다. 따라서 c1 - c5 스레드가 반복해서 깨어날 수 있다.
이렇게 대기 상태의 스레드가 실행 순서를 계속 얻지 못해서 실행되지 않는 상황을 스레드 기아 상태라 한다.
여기서 깨어나야 할 이상적인 스레드는 바로 생산자 스레드인 p1이다.
이 스레드 기아를 해결하는 방법 중 하나인 notifyAll()이 있다.
notifyAll()
이 메서드는 스레드 대기 집합에 있는 모든 스레드를 한번에 다 깨울 수 있다.
데이터를 획득한 c0 스레드가 notifyAll()을 호출한다.
대기 집합에 있는 모든 스레드가 깨어난다.
모든 스레드는 다 임계 영역 안에 있다. 따라서 먼저 락을 획득해야 한다.
락을 획득하지 못하면 BLOCKED 상태에서 머무르게 된다.
만약, c1이 먼저 락을 획득한다면 큐에 데이터가 없으므로 다시 스레드 대기 집합에 들어간다.
c2 - c5도 마찬가지다.
따라서, p1이 가장 늦게 락 획득을 시도해도 c1 - c5 모두 스레드 대기 집합에 들어가있으므로 결과적으로 p1만 남게되고 결국 락을 획득하게 된다.
그러나, 이 경우에 스레드 기아 문제를 해결한다 하더라도 비효율은 해결하지 못한다. 결국 가장 좋은 방법은 소비자는 생산자를, 생산자는 소비자를 깨우는 방법이다.
정리
생산자 - 소비자 문제란?
생산자가 너무 빠를 경우 버퍼에 데이터가 꽉 차서 더 이상 데이터를 생산해낼 수 없다.
소비자가 너무 빠를 경우 버퍼에 남은 데이터가 없어 데이터를 소비할 수 없다.
이 문제를 해결하는 방법은 생산자의 경우 버퍼가 빈 공간이 생길때까지 기다리는 것이고, 소비자의 경우 버퍼에 데이터가 생길때까지 기다리는 것이다. 결국 기다리는 것이다.
그러나, 단순히 기다릴 순 없다. 왜냐하면 임계 영역은 딱 하나의 스레드만 작업할 수 있게 설계되었다. 아무리 특정 스레드가 하루종일 기다린다해도 본인이 락을 들고 있는 상태에서 놔주지 않으면 다른 스레드는 진입할 수 없다.
그래서 단순히 기다린 게 아니라 락을 반납하고 기다린다.
이때 사용할 수 있는 것이 Object.wait()이다. 이 wait(), notify(), notifyAll()은 synchronized와 같이 사용할 수 있다.
wait()으로 락을 반납하고 해당 스레드는 스레드 대기 집합에 들어간다. 그리고 락을 반납했으니 다른 스레드가 진입할 수 있게 된다. 다른 스레드가 작업을 다 마치고 notify()로 스레드 대기 집합에 알린다. 그럼 스레드 대기 집합에 있는 임의의 스레드 하나가 튀어나온다.
그땐 스레드는 BLOCKED 상태이다. 아직 락을 얻지 못한 상태이니까. 그리고 최종적으로 락을 반납하고 스레드가 나가면 튀어나온 스레드가 이 락을 얻어 작업을 진행할 수 있게 됐다.
근본적인 생산자 소비자 문제를 해결했지만, 여기서 파생되는 비효율이 발생했다.
생산자가 생산자를 계속 깨우거나, 소비자가 소비자를 계속 깨우면 의미없이 CPU 자원만 소모하고 아무것도 할 수 없다.
즉, 가장 좋은 방법은 깨울때 생산자는 소비자를, 소비자는 생산자를 깨우는 게 가장 좋은 방법이다. 이 방법을 다음 포스팅에서 알아보자!
이번에는 이벤트 리스너를 등록해보자. 이벤트 리스너란, 이벤트가 발생했을 때 원하는 후처리 작업을 할 수 있는 방법이다.
JavaScript의 이벤트 리스너랑 완전 똑같은 것이라고 보면 된다.
참고로, 이 포스팅은 공식 문서에서 제공하는 방식과 살짝 다르다. 스프링에서 InitializingBean, DisposableBean 인터페이스를 구현하여 빈으로 등록해서, 스프링 컨텍스트(컨테이너)가 최초로 띄워질때와 마지막에 종료될 때 호출될 메서드와 사용할 이벤트 리스너를 등록해 보았다. 왜 그러냐면, 이 플러그인 관련 포스팅을 Part.1에서 쭉 보다보면 스프링의 기술이 들어가 있는것을 알 수가 있는데 스프링의 기술을 사용중이니까 스프링과 잘 호환되는 기술을 사용해보고자 이런 방식을 구현했다
그리고 스프링 기술을 이용했기 때문에 Add-on Descriptor(atlassian-plugin.xml)에 어떠한 작업도 필요 없고 그래서 더 간결하다는 것을 캐치해서 유심히 봐보자!
IssueCreatedResolvedListener
package kr.osci.kapproval.com.jira.eventlistener;
import com.atlassian.event.api.EventListener;
import com.atlassian.event.api.EventPublisher;
import com.atlassian.jira.event.issue.IssueEvent;
import com.atlassian.jira.event.type.EventType;
import com.atlassian.jira.issue.Issue;
import com.atlassian.plugin.spring.scanner.annotation.imports.JiraImport;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
@Slf4j
@Service
@RequiredArgsConstructor
public class IssueCreatedResolvedListener implements InitializingBean, DisposableBean {
@JiraImport
private final EventPublisher eventPublisher;
/**
* Called when the plugin has been enabled.
*/
@Override
public void afterPropertiesSet() {
log.info("Enabling plugin");
eventPublisher.register(this);
}
/**
* Called when the plugin is being disabled or removed.
*/
@Override
public void destroy() {
log.info("Disabling plugin");
eventPublisher.unregister(this);
}
@EventListener
public void onIssueEvent(IssueEvent issueEvent) {
Long eventTypeId = issueEvent.getEventTypeId();
Issue issue = issueEvent.getIssue();
if (eventTypeId.equals(EventType.ISSUE_CREATED_ID)) {
log.info("Issue {} has been created at {}.", issue.getKey(), issue.getCreated());
// 이슈 Created 이벤트가 발생했을 때 실행되는 부분
} else if (eventTypeId.equals(EventType.ISSUE_RESOLVED_ID)) {
log.info("Issue {} has been resolved at {}.", issue.getKey(), issue.getResolutionDate());
// 이슈 Resolved 이벤트가 발생했을 때 실행되는 부분
} else if (eventTypeId.equals(EventType.ISSUE_CLOSED_ID)) {
log.info("Issue {} has been closed at {}.", issue.getKey(), issue.getUpdated());
// 이슈 Closed 이벤트가 발생했을 때 실행되는 부분
}
}
}
우선, InitializingBean을 구현하려면 재정의 할 메서드가 있다.
afterPropertiesSet()
이 메서드는 스프링 컨텍스트가 완전히 띄워졌을 때, 호출되는 메서드이다. 그러니까 스프링이 진짜 이제 실행될 준비가 됐을 때 자동으로 호출되는 메서드이다. 여기서 무엇을 해야 하냐면 내가 이벤트 퍼블리셔를 등록하겠다고 선언해줘야 한다. 그래야 어떤 이벤트가 발생했을 때 이벤트를 캐치할 수 있게 된다.
그래서 이 메서드안에 다음 코드 한 줄이 있다.
eventPublisher.register(this);
그 다음, DisposableBean을 구현하려면 또 한가지 재정의 할 메서드가 있다.
destroy()
이 메서드는 스프링 컨텍스트가 내려가기 바로 전에 호출되는 메서드이다. 그러니까, 스프링이 내려가기 전 마지막으로 정리할 자원들을 정리하는 메서드라고 생각하면 된다. 그래서 등록한 이벤트 퍼블리셔를 다시 등록 해제하면 된다.
eventPublisher.unregister(this);
그리고, 실제 이벤트가 발생했을 때마다 호출될 메서드가 있다. 바로 다음 메서드.
@EventListener
public void onIssueEvent(IssueEvent issueEvent) {
Long eventTypeId = issueEvent.getEventTypeId();
Issue issue = issueEvent.getIssue();
if (eventTypeId.equals(EventType.ISSUE_CREATED_ID)) {
log.info("Issue {} has been created at {}.", issue.getKey(), issue.getCreated());
// 이슈 Created 이벤트가 발생했을 때 실행되는 부분
} else if (eventTypeId.equals(EventType.ISSUE_RESOLVED_ID)) {
log.info("Issue {} has been resolved at {}.", issue.getKey(), issue.getResolutionDate());
// 이슈 Resolved 이벤트가 발생했을 때 실행되는 부분
} else if (eventTypeId.equals(EventType.ISSUE_CLOSED_ID)) {
log.info("Issue {} has been closed at {}.", issue.getKey(), issue.getUpdated());
// 이슈 Closed 이벤트가 발생했을 때 실행되는 부분
}
}
주의 깊게 볼 건 @EventListener 애노테이션이다. 이 애노테이션은 어떠한 public 메서드라도 상관없이 달 수 있는데 이 애노테이션이 달린 메서드의 파라미터 이벤트가 발생할 때마다 이 메서드가 호출된다. 여기서는, IssueEvent라는 이슈 관련 이벤트를 파라미터로 받는다. 생성, 수정, 삭제 등등의 이벤트가 다 잡히게 될 것이다.
그래서 실제로 원하는 이벤트의 후처리 코드는 이 @EventListener 애노테이션이 달린 메서드에서 작업하면 된다.
이렇게 스프링과 JIRA가 제공하는 @EventListener 애노테이션을 사용해서 스프링의 라이프 사이클을 이용해 스프링 컨테이너가 완전히 올라왔을 때(플러그인이 띄워질 때)와 스프링 컨테이너가 완전히 내려가기 바로 직전에(플러그인이 내려가기 직전에) 딱 한 번씩만 이벤트 퍼블리셔를 등록할 수 있고, 이벤트 리스너 메서드를 만들 수 있다.
package kr.osci.aijql.eventlistener;
import com.atlassian.event.api.EventListener;
import com.atlassian.event.api.EventPublisher;
import com.atlassian.jira.event.issue.link.RemoteIssueLinkCreateEvent;
import com.atlassian.jira.event.issue.link.RemoteIssueLinkUICreateEvent;
import com.atlassian.jira.issue.link.RemoteIssueLinkManager;
import com.atlassian.plugin.spring.scanner.annotation.imports.JiraImport;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
@Slf4j
@Service
@RequiredArgsConstructor
public class RemoteIssueLinkListener implements InitializingBean, DisposableBean {
@JiraImport
private final EventPublisher eventPublisher;
@JiraImport
private final RemoteIssueLinkManager remoteIssueLinkManager;
/**
* Called when the plugin has been enabled.
*/
@Override
public void afterPropertiesSet() {
log.debug("[afterPropertiesSet]: RemoteIssueLinkListener initialized.");
eventPublisher.register(this);
}
/**
* REST API 또는 애플리케이션에서 직접 Remote Issue Link 추가하는 경우 호출
* @param remoteIssueLinkCreateEvent remoteIssueLinkCreateEvent
*/
@EventListener
public void onCreateRemoteIssueLinkEvent(RemoteIssueLinkCreateEvent remoteIssueLinkCreateEvent) {
log.info("[onCreateRemoteIssueLinkEvent] called");
log.info("[onCreateRemoteIssueLinkEvent] remote issue link id = {}", remoteIssueLinkCreateEvent.getRemoteIssueLinkId());
log.info("[onCreateRemoteIssueLinkEvent] global id = {}", remoteIssueLinkCreateEvent.getGlobalId());
}
/**
* 오직 애플리케이션에서 사용자가 Remote Issue Link 추가하는 경우 호출
* @param remoteIssueLinkUiCreateEvent remoteIssueLinkUiCreateEvent
*/
@EventListener
public void onCreateUiRemoteIssueLinkEvent(RemoteIssueLinkUICreateEvent remoteIssueLinkUiCreateEvent) {
log.info("[onCreateUiRemoteIssueLinkEvent] called");
log.info("[onCreateUiRemoteIssueLinkEvent] remote issue link id = {}", remoteIssueLinkUiCreateEvent.getRemoteIssueLinkId());
log.info("[onCreateUiRemoteIssueLinkEvent] global id = {}", remoteIssueLinkUiCreateEvent.getGlobalId());
}
/**
* Called when the plugin is being disabled or removed.
*/
@Override
public void destroy() {
log.info("[destroy]: RemoteIssueLinkListener destroyed.");
eventPublisher.unregister(this);
}
}
이번에는 서블릿 필터를 만들어보자. 서블릿 필터는 사실 그냥 Java로 서블릿을 사용하면 거의 무조건 사용하는 컴포넌트이다.
그래서 이건 뭐 JIRA 플러그인을 개발하기 위해 따로 알아야 하는 개념이 아니라 아마 익숙할 것 같다.
우선 서블릿 필터를 만드려면 당연히 dependencies로 서블릿이 있어야 할 것이고, 이건 이미 이 전 포스팅에서 다뤘다.
그리고 필터를 구현하는 클래스를 만들면 된다.
CustomServletFilter
package kr.osci.kapproval.admin.servlet.filter;
import lombok.RequiredArgsConstructor;
import javax.servlet.*;
import java.io.IOException;
@RequiredArgsConstructor
public class CustomServletFilter implements Filter {
@Override
public void init(FilterConfig filterConfig) {
}
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
}
@Override
public void destroy() {
}
}
void init(): 필터 초기화 작업이 필요하다면 이 메서드에 작성하면 된다.
doFilter(): 각 필터마다 이 메서드에서 필터 처리를 한다. 필터는 요청을 다음 필터 또는 서블릿으로 전달할지 여부를 결정할 수 있습니다.
destroy(): 필터 종료 작업이 필요하다면 이 메서드에 작성하면 된다.
필터의 주요 역할
요청(Request)에 대한 전처리: 클라이언트 요청이 서블릿이나 JSP로 전달되기 전에 요청을 수정하거나, 인증/인가 검사를 수행하거나, 로깅 등을 할 수 있다.
응답(Response)에 대한 후처리: 서블릿이나 JSP가 응답을 만들고, 클라이언트에게 전달되기 전에 응답을 수정하거나, 로깅 등을 할 수 있다.
필터의 동작 과정
클라이언트의 요청 수신: 클라이언트로부터 HTTP 요청이 들어오면 웹 서버는 이를 필터 체인(Filter Chain)에 전달한다.
필터 체인 통과: 요청은 필터 체인을 따라 이동하며, 각 필터는 요청을 처리할 기회를 가진다.
각 필터는 `doFilter` 메서드를 통해 요청을 처리한다.
필터는 요청을 다음 필터 또는 서블릿으로 전달할지 여부를 결정할 수 있다.
서블릿 또는 JSP로 전달: 필터 체인을 모두 통과한 요청은 최종적으로 서블릿이나 JSP에 도달하여 본래의 비즈니스 로직을 수행한다.
응답 생성: 서블릿이나 JSP가 응답을 생성하면, 응답은 다시 필터 체인을 따라 클라이언트로 돌아간다.
필터 체인 역순 통과: 응답은 필터 체인을 역순으로 통과하며 각 필터는 응답을 처리할 기회를 가진다.
클라이언트 응답 전달: 최종적으로 처리된 응답이 클라이언트에게 전달된다.
다음과 같은 필터를 만들어보자!
CustomServletFilter
public class CustomServletFilter implements Filter {
@Override
public void init(FilterConfig filterConfig) throws ServletException {
// 필터 초기화 작업 (필요한 경우)
}
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
// 요청 전처리 작업
System.out.println("Request received at MyFilter");
// 필터 체인의 다음 요소로 요청을 전달
chain.doFilter(request, response);
// 응답 후처리 작업
System.out.println("Response leaving MyFilter");
}
@Override
public void destroy() {
// 필터 종료 작업 (필요한 경우)
}
}
이 필터는 다음과 같은 동작을 한다.
클라이언트로부터 요청이 들어오면 `Request received at MyFilter`를 출력한다.
요청을 다음 필터 또는 서블릿으로 넘긴다.
요청에 대한 응답을 생성한 서블릿 또는 JSP가 응답을 다시 필터로 넘기고 그 응답이 여러 필터를 거쳐 이 필터로 도착한다.
`Response leaving MyFilter`를 출력한다.
응답을 클라이언트에게 최종적으로 전달한다.
이렇게 만든 필터를 결국 등록을 해야 사용할 수 있는데, 이 JIRA DC 플러그인을 개발할땐 언제나 리소스는? Add-on Descriptor(atlassian-plugin.xml)에 등록한다. 참고로 JIRA DC 플러그인을 개발하는게 아니면 개발 방식에 따라 필터 등록하는 방법은 다 가지각색이라 목적에 맞는 방법을 찾으면 된다.
서블릿 필터를 등록하고, url-pattern 태그로 어떤 URL의 요청이 이 필터를 거칠지를 결정하면 된다. 이렇게 설정하면 끝이다.
여기서 location 이라는 attribute가 있다. 이건 이 필터가 어디쯤에 위치할지를 정하는 것이다.
나는 `before-dispatch` 라는 값을 주었다. 이게 기본값이고 이건 서블릿 필터 체인의 가장 마지막에 이 필터를 추가하는 것이다. 그러니까 이 요청을 처리하는 서블릿이나 JSP에 도달하기 바로 직전에. 그리고 이러한 옵션들에 대한 내용, 또한 서블릿 필터에 대한 자세한 내용은 아래 공식 문서를 참고하자.
만약, 요청을 가로채서 하는 작업이 인증/인가를 확인하는 처리라면 인증이 되지 않은 경우 다음 필터 또는 서블릿으로 넘기기 전에 그냥 바로 사용자에게 응답을 돌려줄 수 있다. 예를 들면 이런 코드를 작성할 수 있다.
public class CustomServletFilter implements Filter {
@Override
public void init(FilterConfig filterConfig) throws ServletException {
// 필터 초기화 작업 (필요한 경우)
}
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
HttpServletRequest httpRequest = (HttpServletRequest) request;
HttpServletResponse httpResponse = (HttpServletResponse) response;
// 인증 여부 확인 (여기서는 간단하게 세션에 "authenticated" 속성이 있는지 확인)
Boolean isAuthenticated = (Boolean) httpRequest.getSession().getAttribute("authenticated");
if (isAuthenticated == null || !isAuthenticated) {
// 인증이 안된 경우, 바로 응답 생성
httpResponse.setContentType("text/html");
PrintWriter out = httpResponse.getWriter();
out.println("<html><body>");
out.println("<h3>Authentication Required</h3>");
out.println("<p>You are not authenticated. Please <a href=\"login.html\">login</a>.</p>");
out.println("</body></html>");
out.close();
} else {
// 인증이 된 경우, 다음 필터 또는 서블릿으로 요청 전달
chain.doFilter(request, response);
}
}
@Override
public void destroy() {
// 필터 종료 작업 (필요한 경우)
}
}
이거는 단순 예시일 뿐, 구현은 원하는대로 어떻게든 할 수 있다!
결론
서블릿 필터를 통해 클라이언트의 요청을 가로채서 추가 작업을 할 수 있고 응답을 내보내기 전 마지막 작업을 할 수 있다. 그러려면 Filter를 구현한 서블릿 필터 클래스가 필요하고, 이 클래스는 init, doFilter, destroy 라는 메서드를 재정의해야 하는데 여기서 가장 중요한 건 doFilter 메서드이다. 이 doFilter 메서드에 chain.doFilter()를 호출하기 전에 작성한 코드가 클라이언트의 요청이 서블릿으로 넘어가기 전 작업하는 부분이고 chain.doFilter()를 호출한 이후 코드가 생성된 응답을 클라이언트에게 최종적으로 전달하기 전 작업하는 부분이 된다.
이 LockSupport 라는 것을 배우기 전에 먼저 다시 복습을 해보자. synchronized는 자바 1.0부터 제공되는 매우 편리한 기능이지만, 다음과 같은 한계가 있다.
무한대기: BLOCKED 상태의 스레드는 락이 풀릴 때 까지 무한 대기한다.
특정 시간까지만 대기하는 타임아웃도 없다.
중간에 인터럽트도 못한다.
공정성: 락이 돌아왔을 때 BLOCKED 상태의 여러 스레드 중에 어떤 스레드가 락을 획득할 지 알 수 없다. 최악의 경우 특정 스레드가 너무 오랜기간 락을 획득하지 못할 수 있다.
결국, 더 유연하고 더 세밀한 제어가 가능한 방법들이 필요하게 되었다. 이런 문제를 해결하기 위해 자바 1.5부터 java.util.concurrent 라는 동시성 문제 해결을 위한 라이브러리 패키지가 추가된다.
이 라이브러리에는 수 많은 클래스가 있지만, 가장 기본이 되는 LockSupport에 대해서 먼저 알아보자. LockSupport를 사용하면 synchronized의 가장 큰 단점인 무한 대기 문제를 해결할 수 있다.
LockSupport 기능
LockSupport는 스레드를 WAITING 상태로 변경한다. WAITING 상태는 누가 깨워주기 전까지는 계속 대기한다. 그리고 CPU 스케쥴링에 들어가지 않는다. LockSupport의 대표적인 기능은 다음과 같다.
park(): 스레드를 WAITING 상태로 변경한다.
parkNanos(nanos): 스레드를 나노초 동안만 TIMED_WAITING 상태로 변경한다. 지정한 나노초가 지나면 TIMED_WAITING 상태에서 빠져나오고 RUNNABLE 상태로 변경된다.
unpark(thread): WAITING 상태의 대상 스레드를 RUNNABLE 상태로 변경한다.
LockSupportMainV1
package thread.sync.lock;
import java.util.concurrent.locks.LockSupport;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class LockSupportMainV1 {
public static void main(String[] args) {
Thread thread1 = new Thread(new ParkTest(), "Thread-1");
thread1.start();
sleep(100);
log("Thread-1 state: " + thread1.getState());
log("main -> unpark(Thread-1)");
LockSupport.unpark(thread1);
// thread1.interrupt();
}
static class ParkTest implements Runnable {
@Override
public void run() {
log("park 시작");
LockSupport.park();
log("park 종료, state " + Thread.currentThread().getState());
log("인터럽트 상태: " + Thread.currentThread().isInterrupted());
}
}
}
실행결과
2024-07-22 20:57:28.683 [ Thread-1] park 시작
2024-07-22 20:57:28.765 [ main] Thread-1 state: WAITING
2024-07-22 20:57:28.766 [ main] main -> unpark(Thread-1)
2024-07-22 20:57:28.766 [ Thread-1] park 종료, state RUNNABLE
2024-07-22 20:57:28.772 [ Thread-1] 인터럽트 상태: false
main 스레드가 Thread-1을 start()하면 Thread-1은 RUNNABLE 상태가 된다.
Thread-1은 LockSupport.park()를 호출한다. Thread-1은 RUNNABLE → WAITING 상태가 되면서 대기한다.
main 스레드가 Thread-1을 unpark()로 깨운다. Thread-1은 대기 상태에서 실행 가능 상태로 변한다.
WAITING → RUNNABLE 상태로 변한다.
이처럼, LockSupport는 특정 스레드를 WAITING 상태로 또 RUNNABLE 상태로 변경할 수 있다.
그런데, 대기 상태로 바꾸는 LockSupport.park()는 매개변수가 없는데, 실행 가능 상태로 바꾸는 LockSupport.unpark(thread1)은 왜 특정 스레드를 지정하는 매개변수가 있을까? 왜냐하면 실행 중인 스레드는 LockSupport.park()를 호출해서 스스로 대기 상태에 빠질 수 있지만, 대기 상태의 스레드는 자신의 코드를 실행할 수 없기 때문이다. 따라서 외부 스레드의 도움을 받아야 깨어날 수 있다.
인터럽트를 사용해서 WAITING → RUNNABLE로 바꾸기
WAITING 상태의 스레드에 인터럽트가 발생하면 WAITING 상태에서 RUNNABLE 상태로 변하면서 깨어난다.
2024-07-22 21:04:04.277 [ Thread-1] park 시작
2024-07-22 21:04:04.357 [ main] Thread-1 state: WAITING
2024-07-22 21:04:04.357 [ main] main -> unpark(Thread-1)
2024-07-22 21:04:04.357 [ Thread-1] park 종료, state RUNNABLE
2024-07-22 21:04:04.362 [ Thread-1] 인터럽트 상태: true
실행 결과를 보면 스레드가 RUNNABLE 상태로 깨어난 것을 확인할 수 있다. 그리고 해당 스레드의 인터럽트의 상태도 true인 것을 확인할 수 있다. 이처럼 WAITING 상태의 스레드는 인터럽트를 걸어서 중간에 깨울 수 있다.
시간 대기
이번에는 스레드를 특정 시간 동안만 대기하는 parkNanos(nanos)를 호출해보자.
parkNanos(nanos): 스레드를 나노초 동안만 TIMED_WAITING 상태로 변경한다. 지정한 나노초가 지나면 TIMED_WAITING 상태에서 빠져나와서 RUNNABLE 상태로 변경된다.
참고로 밀리초 동안만 대기하는 메서드는 없다. parkUntil(밀리초)라는 메서드가 있는데, 이 메서드는 특정 에포크(Epoch) 시간에 맞추어 깨어나는 메서드이다. 정확한 미래의 에포크 시점을 지정해야 한다.
LockSupportMainV2
package thread.sync.lock;
import java.util.concurrent.locks.LockSupport;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class LockSupportMainV2 {
public static void main(String[] args) {
Thread thread1 = new Thread(new ParkTest(), "Thread-1");
thread1.start();
sleep(100);
log("Thread-1 state: " + thread1.getState());
}
static class ParkTest implements Runnable {
@Override
public void run() {
log("park 시작");
LockSupport.parkNanos(2_000_000_000); // 2초
log("park 종료, state " + Thread.currentThread().getState());
log("인터럽트 상태: " + Thread.currentThread().isInterrupted());
}
}
}
여기서 스레드를 깨우기 위한 unpark()를 사용하지 않는다.
parkNanos(나노초)를 사용하면 지정한 시간 이후에 스레드가 알아서 일어난다.
1초 = 1000밀리초(ms)
1밀리초 = 1,000,000 나노초(ns)
2초 = 2,000,000,000 나노초(ns)
BLOCKED vs WAITING
WAITING 상태에 특정 시간까지만 대기하는 기능이 포함된 것이 TIMED_WAITING이다. 여기서는 둘을 묶어서 WAITING 상태라 표현하겠다.
인터럽트
BLOCKED 상태는 인터럽트가 걸려도 대기 상태를 빠져나오지 못한다. 여전히 BLOCKED 상태이다.
WAITING, TIMED_WAITING 상태는 인터럽트가 걸리면 대기상태를 빠져나온다. 그래서 RUNNABLE 상태로 변한다.
용도
BLOCKED 상태는 자바의 synchronized에서 락을 획득하기 위해 대기할 때 사용된다.
WAITING, TIMED_WAITING 상태는 스레드가 특정 조건이나 시간 동안 대기할 때 발생하는 상태이다.
WAITING 상태는 다양한 상황에서 사용된다. 예를 들어, Thread.join(), LockSupport.park(), Object.wait()과 같은 메서드 호출 시 WAITING 상태가 된다.
TIMED_WAITING 상태는 Thread.sleep(ms), Object.wait(long timeout), Thread.join(long millis), LockSupport.parkNanos(nanos) 등과 같은 시간 제한이 있는 대기 메서드를 호출할 때 발생한다.
BLOCKED, WAITING, TIMED_WAITING 상태 모두 스레드가 대기하며, 실행 스케쥴링에 들어가지 않기 때문에 CPU 입장에서 보면 실행하지 않는 비슷한 상태이다.
BLOCKED 상태는 synchronized에서만 사용하는 특별한 대기 상태라고 이해하면 된다.
WAITING, TIMED_WAITING 상태는 범용적으로 활용할 수 있는 대기 상태라고 이해하면 된다.
LockSupport 정리
LockSupport를 사용하면 스레드를 WAITING, TIMED_WAITING 상태로 변경할 수 있고, 또 인터럽트를 받아서 스레드를 깨울 수도 있다. 이런 기능들을 잘 활용하면 synchronized의 단점인 무한 대기 문제를 해결할 수 있을 것 같다.
synchronized의 단점
무한 대기: BLOCKED 상태의 스레드는 락이 풀릴 때 까지 무한으로 대기한다.
특정 시간까지만 대기하는 타임아웃 X →parkNanos(nanos)를 사용하면 특정 시간까지만 대기할 수 있게 된다.
중간에 인터럽트 X →park(), parkNanos(nanos)는 인터럽트를 걸 수 있다.
이처럼 LockSupport를 활용하면, 무한 대기하지 않는 락 기능을 만들 수 있다. 물론 그냥 되는게 아니라 LockSupport를 활용해서 안전한 임계 영역을 만드는 어떤 기능을 개발해야 한다. 그렇지만! 이런 기능을 직접 만드는 건 너무 어렵다. 예를 들어 스레드 10개를 동시에 실행했는데, 그 중에 딱 한 개의 스레드만 락을 가질 수 있도록 락 기능을 만들어야 한다. 그리고 나머지 9개의 스레드가 대기해야 하는데 어떤 스레드가 대기하고 있는지 알 수 있는 자료구조도 필요하다. 그래야 이후에 대기 중인 스레드를 찾아서 깨울 수 있다. 여기서 끝이 아니고 대기 중인 스레드 중에 어떤 스레드를 깨울지에 대한 우선순위 결정도 필요하다.
즉, LockSupport는 사용할만 한 게 못된다. synchronized를 사용하는 게 날 것 같다. 하지만! 자바는 Lock 인터페이스와 ReentrantLock 이라는 구현체로 이런 기능들을 이미 다 구현해두었다. ReentrantLock은 LockSupport를 활용해서 synchronized의 단점을 극복하면서도 매우 편리하게 임계 영역을 다룰 수 있는 다양한 기능을 제공한다. 즉, 이 녀석을 사용하면 된다.
ReentrantLock
자바는 1.0부터 존재한 synchronized와 BLOCKED 상태를 통한 임계 영역 관리의 한계를 극복하기 위해 자바 1.5부터 Lock 인터페이스와 ReentrantLock 구현체를 제공한다.
synchronized 단점
무한대기: BLOCKED 상태의 스레드는 락이 풀릴 때 까지 무한 대기한다.
특정 시간까지만 대기하는 타임아웃 X
중간에 인터럽트 X
공정성: 락이 돌아왔을 때 BLOCKED 상태의 여러 스레드 중에 어떤 스레드가 락을 획득할 지 알 수 없다. 최악의 경우 특정 스레드가 너무 오랜기간 락을 획득하지 못할 수 있다.
Lock 인터페이스는 동시성 프로그래밍에서 쓰이는 안전한 임계 영역을 위한 락을 구현하는데 사용된다.
Lock 인터페이스는 다음과 같은 메서드들을 제공하는데 대표적인 구현체는 ReentrantLock이 있다.
void lock()
락을 획득한다. 만약 다른 스레드가 이미 락을 획득했다면, 락이 풀릴 때 까지 현재 스레드는 대기(WAITING)한다. 이 메서드는 인터럽트에 응답하지 않는다.
주의! 여기서 사용하는 락은 객체 내부에 있는 모니터 락이 아니다! Lock 인터페이스와 ReentrantLock이 제공하는 기능이다. 모니터 락과 BLOCKED 상태는 synchronized에서만 사용된다.
void lockInterruptibly()
락 획득을 시도하되, 다른 스레드가 인터럽트 할 수 있도록 한다. 만약, 다른 스레드가 이미 락을 획득했다면, 현재 스레드는 락을 획득할 때 까지 대기한다. 대기 중에 인터럽트가 발생하면 InterruptedException이 발생하며 락 획득을 포기한다.
boolean tryLock()
락 획득을 시도하고, 즉시 성공 여부를 반환한다. 만약 다른 스레드가 이미 락을 획득했다면, false를 반환하고 그렇지 않으면 락을 획득하고 true를 반환한다.
boolean tryLock(long time, TimeUnit unit)
주어진 시간동안 락 획득을 시도한다. 주어진 시간 안에 락을 획득하면 true를 반환한다. 주어진 시간이 지나도 락을 획득하지 못한 경우 false를 반환한다. 이 메서드는 대기 중 인터럽트가 발생하면 InterruptedException이 발생하며 락 획득을 포기한다.
void unlock()
락을 해제한다. 락을 해제하면 락 획득을 대기 중인 스레드 중 하나가 락을 획득할 수 있게 된다.
락을 획득한 스레드가 호출해야 하며, 그렇지 않으면 IllegalMonitorStateException이 발생할 수 있다.
Condition newCondition()
Condition 객체를 생성하여 반환한다. Condition 객체는 락과 결합되어 사용되며, 스레드가 특정 조건을 기다리거나 신호를 받을 수 있도록 한다. 이는 Object 클래스의 wait, notify, notifyAll 메서드와 유사한 역할을 한다. 참고로 이 부분은 뒤에서 자세히 다루겠다.
이 메서드들을 사용하면 고수준의 동기화 기법을 구현할 수 있다. Lock 인터페이스는 synchronized 블록보다 더 많은 유연성을 제공하며, 특히 락을 특정 시간 만큼만 시도하거나 인터럽트 가능한 락을 사용할 때 유용하다. 이 메서드들을 보면 알겠지만 다양한 메서드들을 통해 synchronized의 단점인 무한 대기 문제도 깔끔하게 해결할 수 있다.
참고로, lock() 메서드는 인터럽트에 응하지 않는다고 되어있다. 이 메서드의 의도는 인터럽트가 발생해도 무시하고 락을 기다리는 것이다. 앞서 대기(WAITING) 상태의 스레드에 인터럽트가 발생하면 대기 상태를 빠져나온다고 했다. 그런데 lock() 메서드의 설명을 보면 WAITING 상태인데 인터럽트에 응하지 않는다고 되어 있다. 어떻게 된 것일까? lock()을 호출해서 락을 얻기 위해 대기중인 스레드에 인터럽트가 발생하면 순간 대기 상태를 빠져나오는 것은 맞다. 그래서 아주 짧지만 WAITING → RUNNABLE이 된다. 그런데 lock()메서드 안에서 해당 스레드를 다시 WAITING 상태로 강제로 변경해버린다. 이런 원리로 인터럽트를 무시하는 것이다. 참고로 인터럽트가 필요하면 lockInterruptibly()를 사용하면 된다. 새로운 Lock은 개발자에게 다양한 선택권을 제공한다.
공정성
Lock 인터페이스가 제공하는 다양한 기능 덕분에 synchronized의 단점인 무한 대기 문제가 해결되었다. 그런데 공정성에 대한 문제가 남아 있다.
synchronized의 단점
공정성: 락이 돌아왔을 때 BLOCKED 상태의 여러 스레드 중에 어떤 스레드가 락을 획득할 지 알 수 없다. 최악의 경우 특정 스레드가 너무 오랜기간 락을 획득하지 못할 수 있다.
Lock 인터페이스의 대표적인 구현체로 ReentrantLock이 있는데, 이 클래스는 스레드가 공정하게 락을 얻을 수 있는 모드를 제공한다.
private final Lock unFairLock = new ReentrantLock(); // 비공정 모드 락
private final Lock fairLock = new ReentrantLock(true); // 공정 모드 락
ReentrantLock 락은 공정성 모드와 비공정 모드로 설정할 수 있으며, 이 두 모드는 락을 획득하는 방식에서 차이가 있다.
비공정 모드
비공정 모드는 ReentrantLock의 기본 모드이다. 이 모드에서는 락을 먼저 요청한 스레드가 락을 먼저 획득한다는 보장이 없다. 락을 풀었을 때 대기 중인 스레드 중 아무나 락을 획득할 수 있다. 이는 락을 빨리 획득할 수 있지만, 특정 스레드가 장기간 락을 획득하지 못할수도 있다.
비공정 모드 특징
성능 우선: 락을 획득하는 속도가 빠르다.
선점 가능: 새로운 스레드가 기존 대기 스레드보다 먼저 락을 획득할 수 있다.
기아 현상 가능성: 특정 스레드가 계속해서 락을 획득하지 못할 수 있다.
공정 모드
생성자에서 true를 전달하면 된다. 공정 모드는 락을 요청한 순서대로 스레드가 락을 획득할 수 있게 한다. 이는 먼저 대기한 스레드가 먼저 락을 획득하게 되어 스레드 간 공정성을 보장한다. 그러나 이로 인해 성능이 저하될 수 있다.
공정 모드 특징
공정성 보장: 대기 큐에서 먼저 대기한 스레드가 락을 먼저 획득한다.
기아 현상 방지: 모든 스레드가 언젠가 락을 획득할 수 있게 보장한다.
성능 저하: 락을 획득하는 속도가 느려질 수 있다.
정리를 하자면, Lock 인터페이스와 ReentrantLock 구현체를 이용하면 synchronized 단점인 무한 대기와 공정성 문제를 모두 해결할 수 있다!
ReentrantLock 활용
이제 이 ReentrantLock을 직접 이용해보자! 앞에 사용했던 예제 BankAccountV3를 가지고 해보자.
정확한 계산 결과가 반영됐다. 중간에 t1 스레드는 WAITING이고 t2 스레드는 TIMED_WAITING이다. 이 t1이 WAITING인 이유는 t2 스레드가 먼저 임계 영역에 진입했기 때문에 락을 얻지 못해 t1은 기다리는 상태이고, t2 스레드는 로직중에 sleep()에 걸려있을 때 찍힌 TIMED_WAITING이다.
실행 결과를 자세히 분석해보자!
t1, t2가 출금을 시작한다. 여기서는 t1이 약간 먼저 실행된다고 가정하겠다.
ReentrantLock 내부에는 락과 락을 얻지 못해 대기하는 스레드를 관리하는 대기 큐가 존재한다.
여기서 이야기하는 락은 객체 내부에 있는 모니터 락이 아니다! ReentrantLock이 제공하는 기능이다.
t1: ReentrantLock에 있는 락을 획득한다.
락을 획득하는 경우 RUNNABLE 상태가 유지되고, 임계 영역의 코드를 실행할 수 있다.
t1: 임계 영역의 코드를 실행한다.
t2: ReentrantLock에 있는 락 획득을 시도한다. 하지만 락이 없다.
t2: 락을 획득하지 못하면 WAITING 상태가 되고, 대기 큐에서 관리된다.
LockSupport.park()가 내부에서 호출된다.
참고로 tryLock(long time, TimeUnit unit)과 같은 시간 대기 기능을 사용하면 TIMED_WAITING 상태가 되고 대기 큐에서 관리된다.
t1: 임계 영역의 수행을 완료했다. 이 때 잔액은 balance=200이 된다.
t1: 임계 영역을 수행하고 나면 lock.unlock()을 호출한다.
t1: 락을 반납한다.
t1: 대기 큐의 스레드를 하나 깨운다. LockSupport.unpark(thread)가 내부에서 호출된다.
t2: RUNNABLE 상태가 되면서 깨어난 스레드는 락 획득을 시도한다.
이때 락을 획득하면 lock.lock()을 빠져나오면서 대기 큐에서도 제거된다.
이때 락을 획득하지 못하면 다시 대기 상태가 되면서 대기 큐에 유지된다.
참고로 락 획득을 시도하는 잠깐 사이에 새로운 스레드가 먼저 락을 가져갈 수 있다.
공정 모드의 경우 대기 큐에 먼저 대기한 스레드가 먼저 락을 가져간다.
t2: 락을 획득한 t2 스레드는 RUNNABLE 상태로 임계 영역을 수행한다.
t2: 잔액(200)이 출금액(800)보다 적으므로 검증에 실패한다. 따라서 return false가 호출된다.
이때 finally 구문이 있으므로 finally 구문으로 이동한다.
t2: lock.unlock()을 호출해서 락을 반납하고, 대기 큐의 스레드를 하나 깨우려고 시도한다. 대기 큐에 스레드가 없으므로 이때는 깨우지 않는다.
종료된다.
참고로, volatile을 사용하지 않아도 Lock을 사용할 때 접근하는 변수의 메모리 가시성 문제는 해결된다. 모든 동기화를 돕는 기술을 사용하면 알아서 메모리 가시성 문제는 해결된다. (synchronized, LockSupport, ...)
ReentrantLock 활용 2 - 대기 중단
바로 위에서 해봤던건 락을 얻을때까지 대기하는 lock()을 사용했다. 시간도 정해지지 않았고 인터럽트도 먹지 않는 방법이다.
이번에는 tryLock(), tryLock(long time, TimeUnit unit) 메서드들을 사용해서 한번만 시도해보고 얻을 수 있으면 얻고 그렇지 않으면 얻는걸 포기하는 방법과 주어진 시간까지만 시도해보는 방법을 알아보자.
boolean tryLock()
락 획득을 시도하고, 즉시 성공 여부를 반환한다. 만약 다른 스레드가 이미 락을 획득했다면 false를 반환하고, 그렇지 않으면 락을 획득하고 true를 반환한다.
boolean tryLock(long time, TimeUnit unit)
주어진 시간동안 락 획득을 시도한다. 주어진 시간안에 락을 획득하면 true를 반환한다. 주어진 시간이 지나도 락을 획득하지 못한 경우 false를 반환한다. 이 메서드는 대기 중 인터럽트가 발생하면 InterruptedException이 발생하며 락 획득을 포기한다.
아무런 파라미터도 받지 않는 tryLock()은 인터럽트 예외가 발생할 수가 없다. 즉시 결과를 반환하기 때문에 기다리는 시간이 존재하지 않으니까!
t2: lock.tryLock(500, TimeUnit.MILLISECONDS)를 호출하고 락 획득을 시도한다. 락이 없으므로 0.5초 대기한다.
이때 t2는 TIMED_WAITING이 된다.
내부에서는 LockSupport.parkNanos(nanos)가 호출된다.
t2: 대기 시간인 0.5초간 락을 획득하지 못했다. lock.tryLock(500, TimeUnit.MILLISECONDS)에서 즉시 빠져나온다. 이때 false가 반환된다.
스레드는 TIMED_WAITING → RUNNABLE이 된다.
t2: "[진입 실패] 이미 처리중인 작업이 있습니다."를 출력하고 false를 반환하면서 메서드를 종료한다.
t1: 임계 영역의 수행을 완료하고 거래를 종료한다. 마지막으로 락을 반납한다.
정리
자바 1.5에서 등장한 Lock 인터페이스와 ReentrantLock 구현체 덕분에 synchronized의 단점인 무한 대기와 공정성 문제를 극복하고, 또 더욱 유연하고 세밀한 스레드 제어가 가능하게 되었다. 다음 포스팅엔 정말 중요한 생산자 소비자 문제에 대해 알아보자! 어떻게 보면 이 부분이 멀티스레드의 핵심이라고 볼 수 있다!
이제 계좌의 잔액은 200원이다. t2 스레드가 800원을 출금하면 잔액보다 더 많은 돈을 출금하게 되므로 출금에 실패해야 한다.
그런데 실행 결과를 보면 기대와는 다르게 t1, t2 각각 800원씩 총 1600원 출금에 성공한다. 계좌의 잔액은 -600원이 되어있고, 계좌는 예상치 못하게 마이너스 금액이 되어버렸다. 악의적인 사용자는 2대의 PC를 통해 자신의 계좌에 있는 1000원보다 더 많은 금액인 1600원 출금에 성공한다. 분명히 계좌를 출금할 때 잔고를 체크하는 로직이 있는데도 불구하고 왜 이런 문제가 발생했을까?
참고로, balance 값에 volatile을 도입하면 문제가 해결되지 않을까? 하겠지만 그렇지 않다. volatile은 한 스레드가 값을 변경했을 때 다른 스레드에서 변경된 값을 즉시 볼 수 있게 하는 메모리 가시성의 문제를 해결할 뿐이다. 예를 들어, t1 스레드가 balance의 값을 변경했을 때 t2 스레드에서 balance의 변경된 값을 즉시 확인해도 여전히 같은 문제가 발생한다. 이미 t2가 실행할 검증 로직은 지나갔으므로. 물론, 위 예제는 메모리 가시성 문제도 있지만 그것을 해결한다고 해서 동시성 문제가 해결되는것이 아니다.
그럼 이 동시성 문제는 어떤 흐름으로 발생했을까? 이 문제를 어떻게 해결할까를 알기 전에 어떻게 이런 일이 일어났는지를 먼저 확인해보자.
위 동시성 문제의 흐름
심지어 이 예제에서 두가지 케이스의 문제가 있다.
t1 → t2 순서로 실행된 케이스
t1 → t2 순서로 실행됐다고 가정한다. 즉, t1이 아주 약간 빠르게 실행되는 경우를 먼저 알아보자.
t1이 약간 먼저 실행되면서, 출금을 시도한다.
t1이 출금 코드에 있는 검증 로직을 실행한다. 이때 잔액이 출금 액수보다 많은지 확인한다.
잔액(1000)이 출금액(800)보다 많으므로 검증 로직을 통과한다.
t1: 출금 검증 로직을 통과해서 출금을 위해 잠시 대기중이다. 출금에 걸리는 시간이나 실제 출금하기까지 필요한 작업들을 처리하는 단계라고 생각하자.
t2: 검증 로직을 실행한다. 잔액이 출금 금액보다 많은지 확인한다.
잔액(1000)이 출금액(800)보다 많으므로 통과한다.
바로 이 부분이 문제다! t1이 아직 잔액(balance)을 줄이지 못했기 때문에 t2는 검증 로직에서 현재 잔액을 1000원으로 확인한다.
t1이 검증 로직을 통과하고 바로 잔액을 줄였다면 이런 문제가 발생하지 않겠지만, t1이 검증 로직을 통과하고 잔액을 줄이기도 전에 먼저 t2가 검증 로직을 확인한 것이다.
"어? 그럼 sleep(1000)을 빼버리면 되지 않나요?"
→ t1이 검증 로직을 통과하고 balance = balance - amount를 계산하기 직전에 t2가 실행되면서 검증 로직을 통과 할수도 있다. sleep(1000)은 동시성 문제를 해결하는데 아무런 도움도 되지 않는다.
결과적으로 t1, t2 모두 검증 로직을 통과하고, 출금을 위해 잠시 대기중이다. 출금에 걸리는 시간으로 생각하자.
결과적으로, t1, t2 모두 검증 로직을 통과하고, 출금을 위해 잠시 대기중이다. 출금에 걸리는 시간으로 생각하자.
t1은 800원을 출금하면서, 잔액을 1000원에서 출금 액수인 800원 만큼 차감한다. 이제 잔액은 200원이 된다.
t2는 800원을 출금하면서, 잔액을 1000원에서 출금 액수인 800원 만큼 차감한다. 이제 잔액은 200원이 된다.
t1, t2 모두 동시에 실행되기 때문에 둘 다 잔액(balance)를 확인하는 시점에 잔액은 1000원이다.
t1, t2 모두 동시에 계산된 결과를 잔액에 반영하는데, 둘 다 계산 결과인 200원을 balance에 반영하므로 최종 잔액이 200원이 된다.
balance = balance - amount;
이 한줄의 코드도 사실 3단계로 나뉘어진다.
계산을 위해 오른쪽에 있는 balance 값과 amount 값을 조회한다.
두 값을 계산한다.
계산 결과를 왼쪽의 balance 변수에 저장한다
여기서, 1번 단계의 balance 값을 조회할 때 t1, t2 두 스레드가 동시에 x001.balance의 필드값을 읽는다. 이때 값은 1000이다. 따라서 두 스레드는 모두 잔액을 1000원으로 인식한다. 2번 단계에서 두 스레드 모두 1000 - 800을 계산해서 200이라는 결과를 얻는다. 3번 단계에서 두 스레드 모두 balance = 200을 대입한다.
결과적으로,
t1: 800원 출금 완료
t2: 800원 출금 완료
원래 원금이 1000원이었는데, 최종 잔액은 200원이 된다.
은행 입장에서 보면 1600원이 빠져나갔는데, 잔액은 800원만 줄었다. 800원이 감쪽같이 어디론가 사라진 것이다.
실행 결과에서 시간이 완전히 동일하다는 사실을 통해 두 스레드가 같이 실행된 것을 대략 확인할 수 있다.
결국 이런 흐름으로 동시성 문제가 발생했다. 이 문제가 그럼 왜 발생했고 어떻게 해결할 수 있을지 알아보자!
임계 영역
이런 문제가 발생한 근본 원인은 여러 스레드가 함께 사용하는 공유 자원을 여러 단계로 나누어 사용하기 때문이다.
1. 검증 단계: 잔액(balance)이 출금액(amount)보다 많은지 확인하는 단계
2. 출금 단계: 잔액(balance)을 출금액(amount)만큼 줄이는 단계
출금() {
1. 검증 단계
2. 출금 단계
}
이 로직에는 하나의 큰 가정이 있다.
스레드 하나의 관점에서 출금()을 보면, 1. 검증 단계에서 확인한 잔액(balance) 1000원은 2. 출금 단계에서 계산을 끝마칠 때 까지 같은 1000원으로 유지되어야 한다. 그래야 검증 단계에서 확인한 금액으로, 출금 단계에서 정확한 잔액을 계산을 할 수 있다. 결국 여기서는 내가 사용하는 값이 중간에 변경되지 않을 것이라는 가정이 있다.
그런데, 만약 중간에 다른 스레드가 잔액의 값을 변경한다면? 큰 혼란이 발생한다. 1000원이라 생각한 잔액이 다른 값으로 변경되면 잔액이 전혀 다른 값으로 계산될 수 있다.
공유 자원
잔액(balance)은 여러 스레드가 함께 사용하는 공유 자원이다. 따라서 출금 로직을 수행하는 중간에 다른 스레드에서 이 값을 얼마든지 변경할 수 있다. 참고로 여기서는 출금() 메서드를 호출할 때만 잔액(balance)의 값이 변경된다. 따라서 다른 스레드가 출금 메서드를 호출하면서, 사용중인 balance 값을 중간에 변경해 버릴 수 있다.
한 번에 하나의 스레드만 실행
만약, 출금()이라는 메서드를 한 번에 하나의 스레드만 실행할 수 있게 제한한다면 어떻게 될까?
예를 들어, t1, t2 스레드가 함께 출금()을 호출하면 t1 스레드가 먼저 처음부터 끝까지 출금() 메서드를 완료하고, 그 다음에 t2 스레드가 처음부터 끝까지 출금() 메서드를 완료하는 것이다. 이렇게 하면 공유 자원인 balance를 한 번에 하나의 스레드만 변경할 수 있다. 따라서 계산 중간에 다른 스레드가 balance의 값을 변경하는 부분을 걱정하지 않아도 된다. (이 예제에선 출금() 메서드를 호출할때만 잔액(balance)의 값이 변경되므로)
더 자세히는 출금을 진행할 때 잔액(balance)을 검증하는 단계부터 잔액의 계산을 완료할 때 까지 잔액의 값은 중간에 변하면 안된다.
이 검증과 계산 두 단계는 한 번에 하나의 스레드만 실행해야 한다. 그래야 잔액(balance)이 중간에 변하지 않고, 안전하게 계산을 수행할 수 있다.
임계 영역 (critical section)
영어로 크리티컬 섹션이라고 한다.
여러 스레드가 동시에 접근하면 데이터 불일치나 예상치 못한 동작이 발생할 수 있는 위험하고 또 중요한 코드 부분을 뜻한다.
여러 스레드가 동시에 접근해서는 안되는 공유 자원을 접근하거나 수정하는 부분을 의미한다.
예) 공유 변수나 공유 객체를 수정
앞서, 살펴본 출금()로직이 바로 임계 영역이다.
더 자세히는 출금을 진행할 때 잔액(balance)을 검증하는 단계부터 잔액의 계산을 완료할 때 까지가 임계 영역이다. 여기서 balance는 여러 스레드가 동시에 접근해서는 안되는 공유 자원이다.
이런 임계 영역은 한 번에 하나의 스레드만 접근할 수 있도록 안전하게 보호해야 한다. 그럼 어떻게 한 번에 하나의 스레드만 접근할 수 있도록 임계 영역을 안전하게 보호할 수 있을까? 여러가지 방법이 있는데 자바는 synchronized 키워드를 통해 아주 간단하게 임계 영역을 보호할 수 있다.
Synchorinzed
자바의 synchronized 키워드를 사용하면 한 번에 하나의 스레드만 실행할 수 있는 코드 구간을 만들 수 있다.
BankAccountV1을 복사해서 BankAccountV2 클래스를 만들고 synchronized를 도입해보자.
실행 결과를 보면 t2가 withdraw() 메서드를 시작부터 완료까지 모두 끝내고 나서 그 다음에 t1이 withdraw() 메서드를 수행한 것을 확인할 수 있다. 물론 환경에 따라 t1이 먼저 할수도 있다.
synchronized 분석
지금부터 자바의 synchronized가 어떻게 작동하는지 그림으로 분석해보자.
참고로 실행 결과를 보면 t1이 BLOCKED 상태인데 이 상태도 확인해보자.
여기서부턴 t1이 먼저 실행됐다고 가정한다.
스레드 t1이 먼저 synchronized 키워드가 있는 withdraw() 메서드를 호출한다.
synchronized 메서드를 호출하려면 먼저 해당 인스턴스의 락이 필요하다.
모든 인스턴스는 인스턴스마다 락이 있다. 이건 자바의 표준 규칙이다.
같은 클래스의 서로 다른 인스턴스도 다 인스턴스별로 락이 있다.
락이 있으므로 스레드 t1은 BankAccount(x001)인스턴스에 있는 락을 획득한다.
스레드 t1은 해당 인스턴스의 락을 획득했기 때문에 withdraw() 메서드에 진입할 수 있다.
스레드 t2도 withdraw() 메서드 호출을 시도한다. synchronized 메서드를 호출하려면 먼저 해당 인스턴스의 락이 필요하다.
스레드 t2는 BankAccount(x001) 인스턴스에 있는 락 획득을 시도한다. 하지만 락이 없다. 이렇게 락이 없으면 t2 스레드는 락을 획득할 때 까지 BLOCKED 상태로 대기한다.
t2 스레드의 상태는 RUNNABLE → BLOCKED 상태로 변하고, 락을 획득할 때 까지 무한정 대기한다.
참고로, BLOCKED 상태가 되면 락을 다시 획득하기 전까지는 계속 대기하고, CPU 실행 스케쥴링에 들어가지 않는다
t1: 출금을 위한 검증 로직을 수행한다. 조건을 만족하므로 검증 로직을 통과한다.
잔액(1000)이 출금액(800)보다 많으므로 통과한다.
t1: 잔액 1000원에서 800원을 출금하고 계산 결과인 200원을 잔액(balance)에 반영한다.
t1: 메서드 호출이 끝나면 락을 반납한다.
t2: 인스턴스에 락이 반납되면 락 획득을 대기하는 스레드는 자동으로 락을 획득한다.
이때 락을 획득한 스레드는 BLOCKED → RUNNABLE 상태가 되고, 다시 코드를 실행한다.
스레드 t2는 해당 인스턴스의 락을 획득했기 때문에 withdraw() 메서드에 진입할 수 있다.
t2: 출금을 위한 검증 로직을 수행한다. 조건을 만족하지 않으므로 false를 반환한다.
이때 잔액(balance)은 200원이다. 800원을 출금해야 하므로 조건을 만족하지 않는다.
t2: 락을 반납하면서 return한다.
결과
t1: 800원 출금 완료
t2: 잔액 부족으로 출금 실패
원금 1000원, 최종 잔액 200원
t1은 800원 출금에 성공하지만, t2는 잔액 부족으로 출금에 실패한다. 그리고 최종 잔액은 1000원에서 200원이 되므로 정확하게 맞다. 이렇게 자바의 synchronized를 사용하면 한 번에 하나의 스레드만 실행하는 안전한 임계 영역 구간을 편리하게 만들 수 있다.
참고로, 락을 획득하는 순서는 보장되지 않는다. 위 실행결과에서도 봤듯 t2가 먼저 실행됐고 t1이 BLOCKED 상태로 변경됐다면 이 그림에서는 그 반대였다. 즉, BankAccount(x001) 인스턴스의 withdraw()를 수많은 스레드가 동시에 호출한다면, 1개의 스레드만 락을 획득하고 나머지 모두 BLOCKED 상태가 된다. 그리고 이후에 BankAccount(x001) 인스턴스에 락을 반납하면, 해당 인스턴스의 락을 기다리는 수많은 스레드 중에 하나의 스레드만 락을 획득하고, 락을 획득한 스레드만 BLOCKED → RUNNABLE 상태가 된다. 이때, 어떤 순서로 락을 획득하는지는 자바 표준에 정의되어 있지 않다. 따라서 순서를 보장하지 않고 환경에 따라 달라질 수 있다.
참고로, volatile을 사용하지 않아도, synchronized 안에서 접근하는 변수의 메모리 가시성 문제는 해결된다. happens-before를 생각해보자.
synchronized 코드 블럭
synchronized의 가장 큰 장점이자 단점은 한 번에 하나의 스레드만 실행할 수 있다는 점이다. 여러 스레드가 동시에 실행하지 못하기 때문에, 전체로 보면 성능이 떨어질 수 있다. 따라서 synchronized를 통해 여러 스레드를 동시에 실행할 수 없는 코드 구간은 꼭! 필요한 곳으로 한정해서 설정해야 한다.
다시 말해, 메서드 전체로 synchronized를 걸어버리면 그 메서드 안에서 굳이 임계 영역이 아닌 부분까지도 한 스레드만 접근이 가능하니까 성능 저하가 생길 수 있다는 얘기다. 이전에 작성한 코드를 보자.
근데, 메서드 전체에 synchronized를 걸어버리면 굳이 임계 영역이 아닌 부분까지도 한 스레드만 접근이 가능하니까 상대적으로 당연히 성능 저하가 생길 수 밖에 없다. 이 문제를 자바는 synchronized를 메서드 단위가 아니라 특정 코드 블록에 최적화해서 적용할 수 있게 제공한다. 다음 BankAccountV3를 보자.
synchronized 블록 기능을 사용한 덕분에 딱 필요한 부분에 임계 영역을 지정할 수 있었다. 덕분에 아주 약간이지만 여러 스레드가 동시에 수행되는 부분을 더 늘려서, 전체적으로 성능을 더 향상할 수 있었다. 지금의 예는 단순 로그 몇 줄이지만, 실제 업무에서는 더 많은 코드가 존재할 거고 그럴땐 큰 성능 차이가 발생할 것이다.
이 예제에서는 처음 거래 시작 로그를 찍는 부분이 t1, t2 동시에 실행된 것을 확인할 수 있다.
2024-07-21 16:58:41.150 [ T2] 거래 시작: BankAccountV3
2024-07-21 16:58:41.150 [ T1] 거래 시작: BankAccountV3
synchronized 동기화 정리
자바에서 동기화는 여러 스레드가 동시에 접근할 수 있는 자원(예: 객체, 메서드)에 대해 일관성 있고 안전한 접근을 보장하기 위한 메커니즘이다. 동기화는 주로 멀티스레드 환경에서 발생할 수 있는 문제, 예를 들어 데이터 손상이나 예기치 않은 결과를 방지하기 위해 사용된다.
메서드 동기화: 메서드를 synchronized로 선언해서, 메서드에 접근하는 스레드가 하나뿐이도록 보장한다. 이 과정에서 인스턴스의 락을 사용하게 된다. 모든 인스턴스는 다 자기의 락을 가지고 있고, synchronized 키워드가 붙은 곳에 접근하려면 이 락을 스레드는 획득해야 한다!
블록 동기화: 코드 블록을 synchronized로 감싸서 동기화를 구현한다. 위에서 설명한 것과 마찬가지로 인스턴스 락을 사용한다.
이런 동기화를 사용하면 다음 문제들을 해결할 수 있다.
경합 조건(Race condition): 두 개 이상의 스레드가 경쟁적으로 동일한 자원을 수정할 때 발생하는 문제
데이터 일관성: 여러 스레드가 동시에 읽고 쓰는 데이터의 일관성을 유지
동기화는 멀티 스레드 환경에서 필수적인 기능이지만, 과도하게 사용할 경우 성능 저하를 초래할 수 있으므로 꼭 필요한 곳에 적절히 사용해야 한다!
자바는, 처음부터 멀티스레드를 고려하고 나온 언어이다. 그래서 자바 1.0부터 synchronized 동기화 방법을 프로그래밍 언어의 문법에 포함해서 제공한다.
synchronized장점
프로그래밍 언어에 문법으로 제공
아주 편리함
자동 잠금 해제: synchronized 메서드나 블록이 완료되면, 자동으로 락을 대기중인 다른 스레드에게 전달하고 기다리고 있던 스레드의 BLOCKED 상태가 해제된다. 개발자가 직접 특정 스레드를 깨우도록 관리해야 한다면, 매우 어렵고 번거로울 것이다.
synchronized는 매우 편리하지만, 제공하는 기능이 너무 단순하다는 단점이 있다. 시간이 점점 지나면서 멀티 스레드가 더 중요해지고 점점 더 복잡한 동시성 개발 방법들이 필요해졌다.
synchronized단점
무한 대기: BLOCKED 상태의 스레드는 락이 풀릴 때까지 무한 대기한다.
특정 시간까지만 대기하는 타임아웃 X
중간에 인터럽트 X
공정성: 락이 돌아왔을 때 BLOCKED 상태의 여러 스레드 중에 어떤 스레드가 락을 획득할지 알 수 없다. 최악의 경우 특정 스레드가 너무 오랜기간 락을 획득하지 못할 수 있다.
synchronized의 가장 치명적인 단점은 락을 얻기 위해 BLOCKED 상태가 되면 락을 얻을 때까지 무한 대기한다는 점이다. 예를 들어 웹 애플리케이션의 경우 고객이 어떤 요청을 했는데, 화면에 계속 요청 중만 뜨고 응답을 못 받는 것이다. 차라리 너무 오랜 시간이 지나면 시스템에 사용자가 너무 많아서 다음에 다시 시도해달라고 하는 식의 응답을 주는 것이 더 나은 선택일 것이다.
결국 더 유연하고, 더 세밀한 제어가 가능한 방법들이 필요하게 됐다. 이런 문제를 해결하기 위해 자바 1.5부터 java.util.concurrent 라는 동시성 문제 해결을 위한 패키지가 추가된다.
이제 synchronized 까지 배워봤다. 멀티 스레드의 기본적인 것들은 얼추 배웠다고 볼 수 있다. 이제 이 기본기를 가지고 더 좋은 더 여러 기능에 대해 하나씩 알아보자!
참고로, 단순하고 편리하게 사용하기에는 synchronized가 최고이므로, 목적에 부합만 한다면 사용하는데 아무런 문제는 없다!
package thread.volatilestudy;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class VolatileFlagMain {
public static void main(String[] args) {
MyTask task = new MyTask();
Thread t = new Thread(task, "work");
log("runFlag = " + task.runFlag);
t.start();
sleep(1000);
log("runFlag false로 변경");
task.runFlag = false;
log("runFlag = " + task.runFlag);
log("main 종료");
}
static class MyTask implements Runnable {
boolean runFlag = true;
@Override
public void run() {
log("task 시작");
while (runFlag) {
// ?
}
log("task 종료");
}
}
}
프로그램은 아주 간단하다. runFlag를 사용해서 스레드의 작업을 종료한다.
work 스레드는 MyTask를 실행한다. 여기에는 runFlag를 체크하는 무한 루프가 있다.
runFlag 값이 false가 되면 무한 루프를 탈출하여 작업을 종료한다.
이후에 main 스레드가 runFlag의 값을 false로 변경한다.
runFlag의 값이 false가 되었으므로 work 스레드는 무한 루프를 탈출하며, 작업을 종료한다.
참고로, 여기서 runFlag에 volatile 키워드가 안 붙었다. 주의하자!
main 스레드, work 스레드 모두 MyTask 인스턴스(x001)에 있는 runFlag를 사용한다.
이 값을 false로 변경하면 work 스레드의 작업을 종료할 수 있다.
기대하는 실행 결과
2024-07-19 18:26:30.057 [ main] runFlag = true
2024-07-19 18:26:30.061 [ work] task 시작
2024-07-19 18:26:31.063 [ main] runFlag false로 변경
2024-07-19 18:26:31.064 [ main] runFlag = false
2024-07-19 18:26:31.064 [ work] task 종료
2024-07-19 18:26:31.064 [ main] main 종료
실제 실행 결과
2024-07-19 18:26:30.057 [ main] runFlag = true
2024-07-19 18:26:30.061 [ work] task 시작
2024-07-19 18:26:31.063 [ main] runFlag false로 변경
2024-07-19 18:26:31.064 [ main] runFlag = false
2024-07-19 18:26:31.064 [ main] main 종료
실제 실행 결과를 보면 task 종료가 출력되지 않고, 자바 프로그램도 멈추지 않고 계속 실행된다. 정확히는 work 스레드가 while문에서 빠져나오지 못하고 있는 것이다. 분명히 runFlag는 false로 변경됐고, 그럼 while문을 빠져나와야 맞는데 그렇게 동작하지 않는다. 무슨 일일까?
메모리 가시성 문제
멀티 스레드는 메모리 가시성 문제라는 것이 있다. 이게 어떤 문제이고 왜 이런 문제가 발생하는지 그리고 어떻게 해결하는지 차근차근 알아보자.
먼저 일반적으로 생각하는 메모리 접근 방식을 보자.
main 스레드와 work 스레드는 각각의 CPU 코어에 할당되어서 실행된다.
물론 CPU 코어가 1개라면 빠르게 번갈아 가면서 실행할거다. 지금은 2개라고 생각해보자.
실선 위쪽은 스레드의 실행 흐름을 나타내고 실선 아래쪽은 하드웨어를 나타낸다.
자바 프로그램을 실행하고 main 스레드와 work 스레드는 모두 메인 메모리의 runFlag 값을 읽는다.
프로그램의 시작 지점에는 runFlag의 값을 변경하지 않기 때문에 모든 스레드에서 true의 값을 읽는다.
runFlag의 초기값이 true니까 그렇다.
work 스레드의 경우 while(runFlag)가 만족하기 때문에 while문을 계속 반복해서 수행한다.
0. main 스레드는 runFlag의 값을 false로 설정한다.
1. 이때 메인 메모리의 runFlag의 값이 false로 변경된다.
2. work 스레드는 while(runFlag)를 실행할 때 runFlag의 데이터를 메인 메모리에서 확인한다.
3. runFlag의 값이 false이므로 while문을 탈출하고, "task 종료"를 출력한다.
아마도 이런 시나리오를 생각했을 것이다. 그런데 실제로는 이렇게 동작하지 않는다.
실제 메모리 접근 방식
CPU는 처리 성능을 개선하기 위해 중간에 캐시 메모리라는 것을 사용한다.
메인 메모리는 CPU 입장에서 보면 거리도 멀고, 속도도 상대적으로 느리다. 대신에 상대적으로 가격이 저렴해서 큰 용량을 쉽게 구성할 수 있다.
CPU 연산은 매우 빠르기 때문에 CPU 연산의 빠른 성능을 따라 가려면 CPU 가까이에 매우 빠른 메모리가 필요한데, 이것이 바로 캐시 메모리이다. 캐시 메모리는 CPU와 가까이 붙어있고, 속도도 매우 빠른 메모리이다. 하지만 상대적으로 가격이 비싸기 때문에 큰 용량을 구성하기는 어렵다.
현대의 CPU 대부분은 코어 단위로 캐시 메모리를 각각 보유하고 있다.
참고로 여러 코어가 공유하는 캐시 메모리도 있다.
각 스레드가 runFlag의 값을 사용하면, CPU는 이 값을 효율적으로 처리하기 위해 먼저 runFlag를 캐시 메모리에 불러온다.
그리고 이후에는 캐시 메모리에 있는 runFlag를 사용하게 된다.
실선 위쪽은 스레드의 실행 흐름을 나타내고, 점선 아래쪽은 하드웨어를 나타낸다.
자바 프로그램을 실행하고 main 스레드와 work 스레드 모두 runFlag 값을 읽는다.
CPU는 이 값을 효율적으로 처리하기 위해 먼저 캐시 메모리에 불러온다.
main 스레드와 work 스레드가 사용하는 runFlag가 각각의 캐시 메모리에 보관된다.
프로그램의 시작 지점에는 runFlag의 값을 변경하지 않기 때문에 모든 스레드에서 true의 값을 읽는다.
runFlag의 초기값이 true이다.
work 스레드의 경우 while(runFlag)가 만족하기 때문에 while문을 계속 반복해서 수행한다.
0. main 스레드는 runFlag의 값을 false로 변경한다.
1. 이때 캐시 메모리의 runFlag가 false로 설정된다.
여기서 핵심은 캐시 메모리의 runFlag 값만 변한다는 것이다! 메인 메모리에 이 값이 즉시 반영되지 않는다.
main 스레드가 runFlag의 값을 변경해도 CPU 코어1이 사용하는 캐시 메모리의 runFlag 값만 false로 변경된다.
work 스레드가 사용하는 CPU 코어2의 캐시 메모리의 runFlag 값은 여전히 true이다.
work 스레드의 경우 while(runFlag)가 만족하기 때문에 while문을 계속 반복해서 수행한다.
캐시 메모리에 있는 runFlag의 값이 언제 메인 메모리에 반영될까?
이 부분에 대한 정답은 "알 수 없다"이다. CPU 설계 방식과 종류에 따라 다르다. 극단적으로 보면 평생 반영되지 않을수도 있다!
메인 메모리에 반영을 한다고 해도 문제는 여기서 끝나지 않는다.
메인 메모리에 반영이 된 runFlag의 값을 work 스레드가 사용하는 캐시 메모리에 다시 불러와야 한다.
메인 메모리에 반영된 runFlag 값이 언제 CPU 코어2의 캐시 메모리에 반영될까?
이 부분에 대한 정답도 "알 수 없다"이다. CPU 설계 방식과 종류에 따라 다르다. 극단적으로 보면 평생 반영되지 않을수도 있다!
언젠가 CPU 코어2의 캐시 메모리에 runFlag 값을 불러오게 되면 work 스레드가 확인하는 runFlag의 값이 false가 되므로 while문을 탈출하고 "task 종료"를 출력한다.
캐시 메모리를 메인 메모리에 반영하거나, 메인 메모리의 변경 내역을 캐시 메모리에 다시 불러오는 것은 언제 발생할까?
이 부분은 CPU 설계 방식과 실행 환경에 따라 다를 수 있다. 즉시 반영될 수도 있고, 몇 밀리초 후에 될 수도 있고, 몇 초 후에 될 수도 있고, 평생 반영되지 않을 수도 있다. 주로 컨텍스트 스위칭이 될 때, 캐시 메모리도 함께 갱신되는데, 이 부분도 환경에 따라 달라질 수 있다.
예를 들어, Thread.sleep()이나 콘솔에 내용을 출력할 때 스레드가 잠시 쉬는데, 이럴때 컨텍스트 스위칭이 되면서 주로 갱신된다. 하지만 이것이 갱신을 보장하는 것이 아니다.
메모리 가시성 (memory visibility)
이처럼, 멀티 스레드 환경에서 한 스레드가 변경한 값이 다른 스레드에서 언제 보이는지에 대한 문제를 메모리 가시성이라 한다. 이름 그대로 메모리에 변경한 값이 보이는가, 보이지 않는가의 문제이다.
그렇다면 한 스레드에서 변경한 값이 다른 스레드에서 즉시 보이게 하려면 어떻게 해야 할까?
volatile 키워드 사용
캐시 메모리를 사용하면 CPU 처리 성능을 개선할 수 있다. 하지만 때로는 이런 성능 향상보다는, 여러 스레드에서 같은 시점에 정확히 같은 데이터를 보는 것이 더 중요할 수 있다. 해결방안은 아주 단순하다. 성능을 약간 포기하는 대신에, 값을 읽을 때, 값을 쓸 때 모두 메인 메모리에 직접 접근하면 된다. 자바에서는 volatile 이라는 키워드로 이런 기능을 제공한다.
VolatileFlagMain
package thread.volatilestudy;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class VolatileFlagMain {
public static void main(String[] args) {
MyTask task = new MyTask();
Thread t = new Thread(task, "work");
log("runFlag = " + task.runFlag);
t.start();
sleep(1000);
log("runFlag false로 변경");
task.runFlag = false;
log("runFlag = " + task.runFlag);
log("main 종료");
}
static class MyTask implements Runnable {
volatile boolean runFlag = true;
@Override
public void run() {
log("task 시작");
while (runFlag) {
// ?
}
log("task 종료");
}
}
}
기존 코드에서 boolean runFlag 앞에 volatile 키워드만 추가했다.
이렇게 하면 runFlag에 대해서는 캐시 메모리를 사용하지 않고, 값을 읽거나 쓸 때 항상 메인 메모리에 직접 접근한다.
실행결과
2024-07-19 19:11:10.542 [ main] runFlag = true
2024-07-19 19:11:10.545 [ work] task 시작
2024-07-19 19:11:11.551 [ main] runFlag false로 변경
2024-07-19 19:11:11.551 [ work] task 종료
2024-07-19 19:11:11.551 [ main] runFlag = false
2024-07-19 19:11:11.551 [ main] main 종료
여러 스레드에서 같은 값을 읽고 써야 한다면, volatile 키워드를 사용하면 된다. 단 캐시 메모리를 사용할 때 보다 성능이 느려지는 단점이 있기 때문에 꼭! 필요한 곳에만 사용하는 것이 좋다.
자바 메모리 모델 (Java Memory Model)
메모리 가시성(memory visibility)
멀티 스레드 환경에서 한 스레드가 변경한 값이 다른 스레드에 언제 보이는지에 대한 것을 메모리 가시성이라 한다. 이름 그대로 메모리에 변경한 값이 보이는가, 보이지 않는가의 문제이다.
위에서 봤지만 캐시 메모리를 사용한다면 스레드 간 공유하는 값이 서로 다를 수 있고 그 경우 메모리 가시성 문제가 있다고 말한다. (예: main에서 runFlag를 false로 변경했지만 work 스레드는 계속해서 while문을 돌고 있는 경우)
그래서 이 메모리 가시성 문제를 해결하기 위해, volatile 키워드를 사용해서 캐시 메모리를 사용하지 않고 메인 메모리에 직접 접근하게 하도록 했다. 이런 메모리에 대한 접근과 수정이 어떻게 이루어지는가에 대한 것을 정의한게 있는데 이게 자바 메모리 모델이다.
Java Memory Model
Java Memory Model(JMM)은 자바 프로그램이 어떻게 메모리에 접근하고 수정할 수 있는지를 규정하며, 특히 멀티 스레드 환경에서 스레드 간 상호작용을 정의한다. JMM에 대한 여러가지 내용이 있지만, 핵심은 여러 스레드들의 작업 순서를 보장하는 happens-before 관계에 대한 정의다.
happens-before
happens-before 관계는 자바 메모리 모델에서 스레드 간의 작업 순서를 정의하는 개념이다. 만약 A 작업이 B 작업보다 happens-before관계에 있다면 A 작업에서의 모든 메모리 변경 사항은 B 작업에서 볼 수 있다. 즉, A 작업에서 변경된 내용은 B 작업이 시작되기 전에 모두 메모리에 반영된다.
happens-before 관계는 이름 그대로, 한 동작이 다른 동작보다 먼저 발생함을 보장한다.
happens-before 관계는 스레드 간의 메모리 가시성을 보장하는 규칙이다.
happens-before 관계가 성립하면, 한 스레드의 작업을 다른 스레드에서 볼 수 있게 된다.
즉, 한 스레드에서 수행한 작업을 다른 스레드가 참조할 때 최신 상태가 보장되는 것이다.
이 규칙을 따르면, 프로그래머가 멀티 스레드 프로그램을 작성할 때 예상치 못한 동작을 피할 수 있다.
happens-before 관계가 발생하는 경우
이건 외우는 게 아니다! 이런것이 있구나를 알고 넘어가는 것이다.
프로그램 순서 규칙
단일 스레드 내에서 프로그램의 순서대로 작성된 모든 명령문은 happens-before 순서로 실행된다. 예를 들어,
int a = 1;
int b = 2;
이런 코드가 있을 때 a = 1 이 b = 2 보다 먼저 실행된다는 게 보장된다.
volatile 변수 규칙
한 스레드에서 volatile 변수에 대한 쓰기 작업은 해당 변수를 읽는 모든 스레드에 보이도록 한다. 즉, volatile 변수에 대한 쓰기 작업은 그 변수를 읽는 작업보다 happens-before 관계를 형성한다.
그래서, 위에서 volatile로 선언한 runFlag를 false로 변경하고, 그 어떤 다른 동작없이 work 스레드가 while문을 빠져나가고 종료된 것이다.
스레드 시작 규칙
한 스레드에서 Thread.start()를 호출하면, start()를 호출 하기 전에 수행된 모든 작업이 새로운 스레드가 시작된 후의 작업보다 happens-before 관계를 가진다.
스레드 종료 규칙
한 스레드에서 Thread.join()을 호출하면, join 대상 스레드의 모든 작업은 join()이 반환된 후의 작업보다 happens-before 관계를 가진다. 그러니까 Thread.join() 이후에 호출된 코드는 저 join() 대상 스레드에서 한 모든 작업에 대해 알고 있는 상태라는 얘기다.
인터럽트 규칙
한 스레드에서 Thread.interrupt()를 호출하는 작업이 인터럽트된 스레드가 인터럽트를 감지하는 시점의 작업보다 happens-before 관계가 성립한다.
객체 생성 규칙
객체의 생성자는 객체가 완전히 생성된 이후에만 다른 스레드에 의해 참조될 수 있도록 보장한다. 즉, 객체의 생성자에서 초기화 된 필드는 생성자가 완료된 후 다른 스레드에서 참조될 때 happens-before 관계가 성립한다.
모니터 락 규칙
한 스레드에서 synchronized 블록을 종료한 후, 그 모니터 락을 얻는 모든 스레드는 해당 블록 내의 모든 작업을 볼 수 있다. 예를 들어, synchronized(lock) { ... } 블록 내에서의 모든 작업은 블록을 나가는 시점에 happens-before 관계가 형성된다. 뿐만 아니라, ReentrantLock과 같이 락을 사용하는 경우에도 happens-before 관계가 성립한다.
전이 규칙
만약, A가 B보다 happens-before 관계에 있고, B가 C보다 happens-before 관계에 있다면 A는 C보다 happens-before 관계에 있다.
정리
메모리 가시성이란, 멀티 스레드 환경에서 한 스레드의 작업이 다른 스레드에 언제 보이는가에 대한 이야기이다. CPU 코어는 성능의 효율성을 위해 메인 메모리가 아닌 각각 캐시 메모리를 사용하기 때문에 각 스레드가 같은 참조 필드를 알고 있을지라도 서로 다른 값을 가지고 있을 수 있다. 이럴때 메모리 가시성 문제가 발생하고 이런 문제를 해결하기 위해 volatile 키워드를 사용해서, 이 키워드가 붙은 필드는 캐시 메모리를 사용하지 않고 메인 메모리에 직접 접근하게 했다.
즉, volatile 또는 스레드 동기화 기법(synchronized, ReentrantLock)을 사용하면 메모리 가시성 문제가 발생하지 않는다.
package thread.control.interrupt;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class ThreadStopMainV1 {
public static void main(String[] args) {
MyTask task = new MyTask();
Thread thread = new Thread(task, "work");
thread.start();
sleep(4000);
task.runFlag = false;
log("runFlag = false로 변경");
}
static class MyTask implements Runnable {
volatile boolean runFlag = true;
@Override
public void run() {
while (runFlag) {
log("작업 중");
sleep(3000);
}
log("자원 정리");
log("작업 종료");
}
}
}
특정 스레드의 작업을 중단하는 가장 쉬운 방법은 변수를 사용하는 것이다.
여기서는 runFlag를 사용해서 work 스레드에 작업 중단을 지시할 수 있다.
작업 하나에 3초가 걸린다고 가정하고, sleep(3000)을 사용하자.
main 스레드는 4초 뒤에 작업 중단을 지시한다.
volatile 키워드는 뒤에서 자세히 설명한다. 지금은 단순히 여러 스레드에서 공유하는 값에 사용하는 키워드라고 알아두자.
실행결과
2024-07-19 10:08:20.946 [ work] 작업 중
2024-07-19 10:08:23.950 [ work] 작업 중
2024-07-19 10:08:24.920 [ main] runFlag = false로 변경
2024-07-19 10:08:26.952 [ work] 자원 정리
2024-07-19 10:08:26.954 [ work] 작업 종료
문제점
실행해보면 바로 느끼겠지만 main 스레드가 runFlag=false를 통해 작업 중단을 지시해도, work 스레드가 즉각 반응하지 않는다. 왜냐하면 MyTask의 run()에는 sleep(3000)가 있기 때문이다. 3초간 잠들어 있는 상태에서는 runFlag 값이 false여도 잠이 깬 다음 while(runFlag) 코드를 실행해야 작업을 중단할 수 있다. 어떻게 하면 sleep()처럼 스레드가 대기하는 상태에서 스레드를 깨우고, 작업도 빨리 종료할 수 있을까?
인터럽트 코드 추가
예를 들어, 특정 스레드가 Thread.sleep()을 통해 쉬고 있는데 처리해야 하는 작업이 들어와서 해당 스레드를 급하게 깨워야 할 수 있다. 또는 sleep()으로 쉬고 있는 스레드에게 더는 일이 없으니 작업 종료를 지시할 수도 있다.
인터럽트를 사용하면 WAITING, TIMED_WAITING 같은 대기 상태의 스레드를 직접 깨워서 작동하는 RUNNABLE 상태로 만들 수 있다. 아래 예제를 보자.
예제의 run()에서 Thread.sleep(3000)를 사용했다. 이 sleep()은 InterruptedException을 처리해야 한다.
특정 스레드의 인스턴스에 interrupt() 메서드를 호출하면, 해당 스레드에 인터럽트가 발생한다.
인터럽트가 발생하면 해당 스레드에 InterruptedException이 발생한다.
이때 인터럽트를 받은 스레드는 대기 상태에서 깨어나 RUNNABLE 상태가 되고, 코드를 정상 수행한다.
이때 InterruptedException을 catch로 잡아서 정상 흐름으로 변경하면 된다.
참고로, interrupt()를 호출했다고 해서 즉각 InterruptedException이 발생하는 것은 아니다. 오직 sleep()처럼 InterruptedException을 던지는 메서드를 호출하거나 호출 중일 때 예외가 발생한다.
예를 들어, 위 코드에서 while(true), log("작업 중") 에서는 InterruptedException이 발생하지 않는다.
Thread.sleep()처럼 InterruptedException을 던지는 메서드를 호출하거나 또는 호출하며 대기중일 때 예외가 발생한다.
실행결과
2024-07-19 10:57:48.092 [ work] 작업 중
2024-07-19 10:57:51.096 [ work] 작업 중
2024-07-19 10:57:52.070 [ main] 작업중단 지시: thread.interrupt()
2024-07-19 10:57:52.093 [ main] work 스레드 인터럽트 상태1 = true
2024-07-19 10:57:52.093 [ work] work 스레드 인터럽트 상태2 = false
2024-07-19 10:57:52.094 [ work] interrupt message = null
2024-07-19 10:57:52.094 [ work] state = RUNNABLE
2024-07-19 10:57:52.095 [ work] 자원 정리
2024-07-19 10:57:52.095 [ work] 작업 종료
thread.interrupt()를 통해 작업 중단을 지시하고, 거의 즉각적으로 인터럽트가 발생한 것을 확인할 수 있다.
이때 work 스레드는 TIMED_WAITING 에서 RUNNABLE 상태로 변경되면서 InterruptedException 예외가 발생한다.
참고로, 스레드가 RUNNABLE 상태여야 catch의 예외 코드도 실행될 수 있다.
실행 결과를 보면 work 스레드가 catch 블록 안에서 RUNNABLE 상태로 바뀐 것을 알 수 있다.
흐름을 자세히 뜯어보자.
main 스레드가 4초 뒤에 work 스레드에 interrupt()를 호출한다.
work 스레드는 인터럽트 상태(true)가 된다.
스레드가 인터럽트 상태일 땐, sleep()처럼 InterruptedException이 발생하는 메서드를 호출하거나 또는 이미 호출해서 대기 중이라면 InterruptedException이 발생한다. 그게 아니라 log("작업 중") 이런 단계에서는 InterruptedException은 발생하지 않는다.
그러니까 만약, interrupt()가 호출되서 인터럽트 상태가 true가 된 work 스레드가 현재 진행중인 코드가 log("작업 중")이면 InterruptedException이 발생하지 않고 계속 다음 코드를 진행하고 그렇게 진행하다가 InterruptedException을 발생시키는 코드인 Thread.sleep()을 만나면 InterruptedException이 터진다.
이때 두가지 일이 발생한다.
work 스레드는 TIMED_WAITING 에서 RUNNABLE 상태로 변경되고, InterruptedException 예외를 처리하면서 반복문을 탈출한다.
work 스레드는 인터럽트 상태가 된 상태고 인터럽트 예외가 발생한다. 인터럽트 예외가 발생하면 다시 이 work 스레드는 인터럽트 상태가 false가 된다. InterruptedException이 터졌으니까. 그리고 터짐과 동시에 work 스레드는 다시 작동하는 RUNNABLE 상태가 되는 것이다.
주요 로그
2024-07-19 10:57:52.093 [ main] work 스레드 인터럽트 상태1 = true //여기서 인터럽트 발생
2024-07-19 10:57:52.093 [ work] work 스레드 인터럽트 상태2 = false //이 지점은 인터럽트 예외가 터지고 다시 인터럽트 상태가 false가 된, 즉 RUNNABLE 상태가 된 지점
인터럽트가 적용되고, 인터럽트 예외가 발생하면, 해당 스레드는 실행 가능 상태가 되고 인터럽트 발생 상태도 정상으로 돌아온다.
인터럽트를 사용하면 대기중인 스레드를 바로 깨워서 실행 가능한 상태로 바꿀 수 있다. 덕분에 단순히 runFlag를 사용하는 이전 방식보다 반응성이 더 좋아진 것을 알 수 있다.
여기서 while(true) 부분은 인터럽트 체크를 하지 않는다는 점이다. 인터럽트가 저 시점에 이미 발생했다고 해도 체크하지 않기 때문에 다음 코드로 넘어가고 또 다음 코드로 넘어가서 Thread.sleep(3000); 이라는 인터럽트 예외가 발생할 수 있는 지점에 도착해서야 비로소 인터럽트가 발생한다.
만약 다음과 같이 인터럽트 상태를 확인하면 더 빨리 반응할 수 있을것이다.
while (인터럽트 상태 확인) {
log("작업 중");
Thread.sleep(3000);
}
추가적으로 인터럽트의 상태를 직접 확인하면, 인터럽트를 발생시키는 sleep()과 같은 코드가 없어도 인터럽트 상태를 직접 확인하기 때문에 while 문을 빠져나갈 수 있다.
while (인터럽트 상태 확인) {
log("작업 중");
}
그래서 다음과 같이 코드를 바꿔보자.
ThreadStopMainV3
package thread.control.interrupt;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class ThreadStopMainV3 {
public static void main(String[] args) {
MyTask task = new MyTask();
Thread thread = new Thread(task, "work");
thread.start();
sleep(100);
log("작업중단 지시: thread.interrupt()");
thread.interrupt();
log("work 스레드 인터럽트 상태1 = " + thread.isInterrupted());
}
static class MyTask implements Runnable {
@Override
public void run() {
while (!Thread.currentThread().isInterrupted()) {
log("작업 중");
}
log("work 스레드 인터럽트 상태2 = " + Thread.currentThread().isInterrupted());
log("자원 정리");
log("작업 종료");
}
}
}
Thread.currentThread().isInterrupted() 로 현재 스레드가 인터럽트 상태인지 확인할 수 있다.
물론 이 Thread.currentThread().isInterrupted() 코드는 상태를 확인만 하지 인터럽트 상태를 변경하지는 않는다.
실행결과
...
2024-07-19 12:34:19.008 [ work] 작업 중
2024-07-19 12:34:19.008 [ work] 작업 중
2024-07-19 12:34:19.008 [ work] 작업 중
2024-07-19 12:34:19.008 [ main] 작업중단 지시: thread.interrupt()
2024-07-19 12:34:19.008 [ work] 작업 중
2024-07-19 12:34:19.016 [ work] work 스레드 인터럽트 상태2 = true
2024-07-19 12:34:19.016 [ main] work 스레드 인터럽트 상태1 = true
2024-07-19 12:34:19.016 [ work] 자원 정리
2024-07-19 12:34:19.016 [ work] 작업 종료
main 스레드는 interrupt() 메서드를 사용해서 work 스레드에 인터럽트를 건다.
work 스레드는 인터럽트 상태가 됐다. isInterrupted()는 true가 된다.
이때 다음과 같이 while 조건이 인터럽트 상태가 아닌 경우에만 돌기 때문에 인터럽트 상태가 되면 이 while 문을 빠져나온다.
문제가 없어보이지만 이 코드에는 심각한 문제가 있다.
바로 work 스레드의 인터럽트 상태가 보는것과 같이 계속 true로 유지된다는 점이다.
앞서, 인터럽트 예외가 터진 경우 스레드의 인터럽트 상태는 false가 된다.
반면에 isInterrupted() 메서드는 인터럽트의 상태를 변경하지는 않고 상태를 확인만 한다.
그럼 원초적인 질문으로 돌아와서 왜 InterruptedException이 발생하면 다시 인터럽트 상태가 false로 변경될까?
다음과 같은 코드가 있다고 생각해보자.
@Override
public void run() {
while (!Thread.currentThread().isInterrupted()) {
log("작업 중");
}
log("work 스레드 인터럽트 상태2 = " + Thread.currentThread().isInterrupted());
try {
log("자원 정리");
Thread.sleep(1000);
} catch (InterruptedException e) {
log("인터럽트 예외 발생 - 자원 정리 실패");
}
log("작업 종료");
}
인터럽트 상태라면 while 문을 빠져나오는 run() 메서드이다.
main 스레드가 work 스레드에 인터럽트를 발생시키고 이제 work 스레드는 인터럽트 상태가 된다.
인터럽트 상태이니까 while 문을 빠져나온다.
그러나 여전히 인터럽트 상태는 true이다.
자원 정리를 시작한다. 자원 정리 시에는 상당시간 소요가 걸리는 작업이라고 가정하며 중간 중간에 sleep()을 통해 잠시 쉬어가야 하는 부분도 있다고 해보자.
그러나 자원 정리 중 만나는 Thread.sleep(1000); 에서 인터럽트 예외가 발생한다. 왜? 인터럽트 상태가 계속 true 였으니까.
이걸 원하는 개발자는 없을것이다. 자원 정리시에는 어떤 상황에도 정리가 깔끔히 되기를 원하지만 인터럽트 예외가 터지지 않고서는 인터럽트 상태가 false로 변경되지 않으니까 자원 정리시 만나는 sleep()에 인터럽트 예외가 발생해서 자원 정리를 마저 할 수 없게 됐다.
이런 상황을 겪지 않게 하고자 InterruptedException이 발생하면 자동적으로 인터럽트 상태가 false로 변경되는 것이다. 즉, 인터럽트의 목적을 달성하면 인터럽트 상태를 다시 정상으로 돌려두어야 한다. 그리고 자원 정리 시 만나는 인터럽트 예외 시에 인터럽트 상태는 당연히 false로 변경된다.
그럼 어떻게 할까? 인터럽트를 확인한 다음에 인터럽트 상태를 확인해서 while 문을 빠져나오면 다시 인터럽트 상태를 정상으로 돌려두어야 한다.
Thread.interrupted()
스레드의 인터럽트 상태를 단순히 확인만 하는 용도라면 isInterrupted()를 사용하면 된다.
하지만 직접 체크해서 사용할땐 interrupted()를 사용해야 한다.
이 메서드는 굉장히 재밌다.
우선 스레드가 인터럽트 상태인지 확인한다. 맞다면 true를 아니라면 false를 반환한다.
그리고 true를 반환할 땐, 다른 말로 인터럽트 상태라면 인터럽트 상태를 원래대로(false) 돌려놓는다.
ThreadStopMainV4
package thread.control.interrupt;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class ThreadStopMainV4 {
public static void main(String[] args) {
MyTask task = new MyTask();
Thread thread = new Thread(task, "work");
thread.start();
sleep(100);
log("작업중단 지시: thread.interrupt()");
thread.interrupt();
log("work 스레드 인터럽트 상태1 = " + thread.isInterrupted());
}
static class MyTask implements Runnable {
@Override
public void run() {
while (!Thread.interrupted()) {
log("작업 중");
}
log("work 스레드 인터럽트 상태2 = " + Thread.currentThread().isInterrupted());
try {
log("자원 정리");
Thread.sleep(1000);
} catch (InterruptedException e) {
log("인터럽트 예외 발생 - 자원 정리 실패");
}
log("작업 종료");
}
}
}
실행결과
...
2024-07-19 12:56:21.495 [ work] 작업 중
2024-07-19 12:56:21.495 [ work] 작업 중
2024-07-19 12:56:21.495 [ work] 작업 중
2024-07-19 12:56:21.495 [ work] 작업 중
2024-07-19 12:56:21.494 [ main] 작업중단 지시: thread.interrupt()
2024-07-19 12:56:21.495 [ work] 작업 중
2024-07-19 12:56:21.504 [ work] work 스레드 인터럽트 상태2 = false
2024-07-19 12:56:21.504 [ work] 자원 정리
2024-07-19 12:56:21.504 [ main] work 스레드 인터럽트 상태1 = true
2024-07-19 12:56:22.505 [ work] 작업 종료
실행 결과처럼, Thread.interrupted()로 인터럽트 상태를 체크하고, 맞다면 다시 원래대로 돌려 놓기 때문에 2번째 상태 체크에서 false를 반환했다. 마치 인터럽트 예외를 만났을 때와 같이 동작한다.
그래서 마침내 원하는 목적을 최종적으로 모두 달성할 수 있게 됐다. 물론, 무조건 인터럽트 상태를 체크했다면 다시 돌려놔야 하는게 정답은 아니다. 모든 세상사가 다 그렇듯 정해져있는 정답이란게 없기 때문에. 만약, 자원 정리고 뭐고 당장 이 스레드를 꺼야한다면 그냥 인터럽트 만나자마자 바로 끝내버려도 된다. 그래서 정답은 없지만 일반적인 흐름에서는 이렇게 인터럽트의 목적을 달성하면, 인터럽트를 원래 상태로 돌려놓아야 한다.
참고로, 예전에 Thread.stop()이라는 메서드가 있었다. Java 1.2 이후에 Deprecated된 메서드이고 이제는 사용하면 안된다. 원하는대로 동작도 안할것이고 사용하면 안된다 그냥. 왜 그러냐면 뭐 밑도 끝도 없이 그냥 바로 스레드를 중지 시키면 예측 할 수 없는 상황이 발생할 가능성이 너무 높기 때문에 이제는 사용하지 않고 Thread.interrupt()를 사용한다고 알아두자.
인터럽트의 실제 사용 예제
인터럽트를 그럼 실제로 어떤식으로 사용할 수 있는지 실용적인 예제를 만들어보자.
프린터가 있다고 생각해보자. 조금 특별한 프린터인데 사용자의 입력을 출력하는 프린터.
MyPrinterV1
package thread.control.printer;
import java.util.Queue;
import java.util.Scanner;
import java.util.concurrent.ConcurrentLinkedQueue;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;
public class MyPrinterV1 {
public static void main(String[] args) {
Printer printer = new Printer();
Thread printerThread = new Thread(printer, "printer");
printerThread.start();
Scanner userInput = new Scanner(System.in);
while (true) {
log("프린터할 문서를 입력하세요. 종료(q): ");
String input = userInput.nextLine();
if (input.equals("q")) {
printer.work = false;
break;
}
printer.addJob(input);
}
}
static class Printer implements Runnable {
volatile boolean work = true;
Queue<String> q = new ConcurrentLinkedQueue<>();
@Override
public void run() {
while (work) {
if (q.isEmpty()) {
continue;
}
String job = q.poll();
log("출력 시작: " + job + ", 대기 문서: " + q);
sleep(3000);
log("출력 완료");
}
log("프린터 종료");
}
public void addJob(String job) {
q.offer(job);
}
}
}
프린트 기능을 실질적으로 담당하는 스레드를 하나 만들었다. Printer
이 스레드는 두 개의 필드를 가진다. work, q
work는 프린트 기능을 종료하는데 판단되는 필드이고 q는 사용자의 입력을 순서대로 받는 Queue이다.
work는 volatile 키워드를 붙였다. 여러 스레드에서 접근하는 변수인 경우에 사용한다고 생각하면 된다. 이후에 더 자세히 배우게 된다.
q는 ConcurrentLinkedQueue 인스턴스이다. 이 Concurrent 역시 여러 스레드에서 접근하는 경우에 사용하는 자료구조는 일반적인 자료구조를 사용하면 안전하지 않기 때문에 일단은 동시성을 지원하는 동시성 컬렉션을 사용한다고 생각하자.
run() 에서는 큐에 들어간 데이터가 없으면 continue;를 만나서 계속해서 while문을 다시 돌고 있다면 하나씩 앞에서부터 꺼내서 출력하고 큐에 남아있는 값들을 출력한다.
만약, work가 false가 된다면 while 문을 빠져나오고 스레드는 종료된다.
addJob(String job) 메서드는 main 스레드에서 사용자의 입력을 받을때마다 그 값을 이 메서드를 호출해서 큐에 넣는다.
실행하면 다음과 유사한 결과를 얻는다.
2024-07-19 13:17:19.845 [ main] 프린터할 문서를 입력하세요. 종료(q):
a
2024-07-19 13:17:21.631 [ main] 프린터할 문서를 입력하세요. 종료(q):
2024-07-19 13:17:21.639 [ printer] 출력 시작: a, 대기 문서: []
b
2024-07-19 13:17:21.901 [ main] 프린터할 문서를 입력하세요. 종료(q):
c
2024-07-19 13:17:22.228 [ main] 프린터할 문서를 입력하세요. 종료(q):
d
2024-07-19 13:17:22.594 [ main] 프린터할 문서를 입력하세요. 종료(q):
e
2024-07-19 13:17:22.923 [ main] 프린터할 문서를 입력하세요. 종료(q):
f
2024-07-19 13:17:23.340 [ main] 프린터할 문서를 입력하세요. 종료(q):
2024-07-19 13:17:24.640 [ printer] 출력 완료
2024-07-19 13:17:24.642 [ printer] 출력 시작: b, 대기 문서: [c, d, e, f]
2024-07-19 13:17:27.644 [ printer] 출력 완료
2024-07-19 13:17:27.644 [ printer] 출력 시작: c, 대기 문서: [d, e, f]
2024-07-19 13:17:30.646 [ printer] 출력 완료
2024-07-19 13:17:30.647 [ printer] 출력 시작: d, 대기 문서: [e, f]
2024-07-19 13:17:33.649 [ printer] 출력 완료
2024-07-19 13:17:33.650 [ printer] 출력 시작: e, 대기 문서: [f]
2024-07-19 13:17:36.651 [ printer] 출력 완료
2024-07-19 13:17:36.652 [ printer] 출력 시작: f, 대기 문서: []
2024-07-19 13:17:39.653 [ printer] 출력 완료
앞서 살펴보았듯, 이 방식의 문제는 종료(q)를 입력했을 때 바로 반응하지 않을것이라는 점이다. 왜냐하면 printer 스레드가 반복문을 빠져나오려면 while문을 체크해야 하는데 printer 스레드가 sleep(3000)을 통해 대기 상태에 빠져서 작동하지 않기 때문이다. 따라서 최악의 경우 q를 입력하고 3초 이후에 프린터가 종료된다. 이제 인터럽트를 사용해서 반응성이 느린 문제를 해결해보자.
인터럽트 도입
MyPrinterV1
package thread.control.printer;
import java.util.Queue;
import java.util.Scanner;
import java.util.concurrent.ConcurrentLinkedQueue;
import static util.MyLogger.log;
public class MyPrinterV1 {
public static void main(String[] args) {
Printer printer = new Printer();
Thread printerThread = new Thread(printer, "printer");
printerThread.start();
Scanner userInput = new Scanner(System.in);
while (true) {
log("프린터할 문서를 입력하세요. 종료(q): ");
String input = userInput.nextLine();
if (input.equals("q")) {
printerThread.interrupt();
break;
}
printer.addJob(input);
}
}
static class Printer implements Runnable {
Queue<String> q = new ConcurrentLinkedQueue<>();
@Override
public void run() {
try {
while (!Thread.interrupted()) {
if (q.isEmpty()) {
continue;
}
String job = q.poll();
log("출력 시작: " + job + ", 대기 문서: " + q);
Thread.sleep(3000);
log("출력 완료");
}
} catch (InterruptedException e) {
log("프린터 종료");
}
}
public void addJob(String job) {
q.offer(job);
}
}
}
인터럽트를 도입한 코드이다. work 라는 스레드의 필드를 없애고 단순히 Thread.interrupted()로 while 문에서 체크한다.
인터럽트를 main 스레드에서 발생시키면 while문에서도, Thread.sleep(3000); 에서도 인터럽트가 발생시 인터럽트 예외가 발생하고 곧바로 빠져나오게 된다.
yield
어떤 스레드를 얼마나 실행할지는 운영체제가 스케쥴링을 통해 결정한다. 그런데 특정 스레드가 크게 바쁘지 않은 상황이어서 다른 스레드에게 CPU 실행 기회를 양보하고 싶을 수 있다. 이렇게 양보하면 스케쥴링 큐에 대기중인 다른 스레드가 CPU 실행 기회를 더 빨리 얻을 수 있다.
YieldMain
package thread.control.yield;
import static util.ThreadUtils.sleep;
public class YieldMain {
static final int THREAD_COUNT = 1000;
public static void main(String[] args) {
for (int i = 0; i < THREAD_COUNT; i++) {
Thread thread = new Thread(new MyRunnable());
thread.start();
}
}
static class MyRunnable implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
System.out.println(Thread.currentThread().getName() + " - " + i);
// sleep(1);
Thread.yield();
}
}
}
}
sleep(1)을 사용해서 스레드의 상태를 1밀리초 동안 아주 잠깐 RUNNABLE → TIMED_WAITING 상태로 변경한다. 이렇게 되면 스레드는 CPU 자원을 사용하지 않고 실행 스케쥴러에서 잠시 제외된다. 1밀리초의 대기 시간 이후 다시 TIMED_WAITING → RUNNABLE 상태가 되면서 실행 스케쥴링에 포함된다.
결과적으로 TIMED_WAITING 상태가 되면서 다른 스레드에 실행을 확실하게 양보하게 된다. 그리고 스케쥴링 큐에 대기중인 다른 스레드가 CPU 실행 기회를 빨리 얻을 수 있다.
하지만 이 방식은 RUNNABLE → TIMED_WAITING → RUNNABLE 로 변경되는 복잡한 과정을 거치고 또 특정 시간 만큼 스레드가 실행되지 않는다는 단점이 있다. 예를 들어, 양보할 스레드가 없다면? 차라리 내 스레드를 계속 실행하는게 더 나은 선택일 것이다. 그러나 이 방법은 다른 스레드 모두 대기 상태로 쉬고 있어도 내 스레드까지 잠깐 실행되지 않는 경우가 생길 수 있다.
자바의 스레드가 RUNNABLE 상태일 때, 운영체제의 스케쥴링은 다음과 같은 상태들을 가질 수 있다.
실행 상태: 스레드가 CPU에서 실제로 실행중이다.
실행 대기 상태: 스레드가 실행될 준비가 되었지만, CPU가 바빠서 스케쥴링 큐에서 대기중이다.
운영체제는 실행 상태의 스레드들을 잠깐만 실행하고 실행 대기 상태로 만든다. 그리고 실행 대기 상태의 스레드들을 잠깐만 실행 상태로 변경해서 실행한다. 이 과정을 계속해서 반복한다. 참고로 자바에서는 두 상태를 구분할 수 없다. 둘 다 RUNNABLE이다.
Thread.yield()메서드는 현재 실행 중인 스레드가 자발적으로 CPU를 양보하여 다른 스레드가 실행될 수 있도록 한다.
Thread.yield() 메서드를 호출한 스레드는 RUNNABLE 상태를 유지하면서 CPU를 양보한다. 이 스레드는 다시 스케쥴링 큐에 들어가면서 다른 스레드에게 CPU 사용 기회를 넘긴다.
자바에서 Thread.yield()메서드를 호출하면 현재 실행 중인 스레드가 CPU를 양보하고 스케쥴링 큐에 들어갈 뿐 RUNNABLE 상태를 유지하기 때문에 쉽게 이야기해서 양보할 사람이 없다면 본인 스레드가 계속 실행될 수 있다.
yield의 실제 사용 예제
난 처음에 이걸 왜 쓰나? 싶은 생각이 제일 먼저 들었다. 와닿지 않았으니까. 와닿는 예제가 바로 위에서 한 프린터 예제에 있다!
바로 이 부분이다.
while (!Thread.interrupted()) {
if (q.isEmpty()) {
continue;
}
...
}
프린터 예제를 실행하면 생성된 스레드가 이 while문을 계속해서 돌면서 두 가지를 체크한다.
스레드가 인터럽트가 걸렸는지 확인
안 걸렸다면 자신이 가지고 있는 큐에 데이터가 있는지 확인
없다면 continue를 만나 다시 while문으로 돌아와서 위 과정을 반복
이 과정이 눈에 안 보일뿐이지 쉴 틈 없이 CPU에서 이 로직이 수행되고 있다. 1초에 while문을 수억번 반복할 수도 있다! 운영체제가 컨텍스트 스위칭을 발생시켜 다른 스레드의 작업을 하기 전까지.
여기서 만약, 큐에 아무것도 없다면 continue를 만나서 다시 while문으로 돌아가 또 체크를 하는 과정을 몇번(어떻게 보면 몇억번)더 반복하게 CPU를 태우는것보다 다른 스레드에게 바로 양보하면 좋지 않을까? 그러니까 좀 더 크게 생각해서 이 프린터라는 스레드도 돌고 사용자 입력을 받는 스레드가 한 200개 있다고 가정하면 빨리 사용자의 입력을 받는 스레드에게 양보해서 큐에 채워넣는게 훨씬 더 이득일 것 같다.
이렇게 코드를 수정해보자.
while (!Thread.interrupted()) {
if (q.isEmpty()) {
Thread.yield();
continue;
}
...
}
위에서 말했지만 yield()는 양보를 할 수 있게 다시 스케쥴링 큐에 들어가는 거지 RUNNABLE 상태임에는 변함이 없다. 또한, 양보할 스레드가 없다면 그냥 자기가 계속 CPU를 사용할수도 있다. 그래서 몇번 더 반복해서 사용자의 입력이 들어오거나 인터럽트가 발생하는걸 체크하는걸 먼저 하는 욕심을 부리지 않고 다른 스레드에게 양보하면 CPU를 효율적으로 사용할 수 있을 것 같다.
정리
interrupt와 yield에 대해 알아보았다. 둘 다 왜 필요한지에 대해 깊이 있게 알아보았고 실전 예제로도 적용시켜 보면서 좀 더 와닿게 느껴졌다. 인터럽트는 인터럽트를 건다고 바로 인터럽트가 터지는게 아니라 인터럽트 예외를 터트리는 메서드를 만나거나 현재 인터럽트 상태인지 확인하고 맞다면 인터럽트 상태를 변경하는 Thread.interrupted() 같은 메서드를 만나야 인터럽트가 터진다. 그리고 이 경우 인터럽트 상태에서 원래대로 돌아온다.
yield()는 양보하는 방법이다. sleep()은 양보를 확실하게 주어진 시간만큼 하지만 양보하지 않아도 될 때에도 양보를 하게 되는 아이러니한 상황도 발생한다. 그리고 우선 RUNNABLE → TIMED_WAITING으로 스레드의 상태가 변경되는것도 비용이 발생하는 문제가 있다. 그래서 yield()를 사용해서 계속 RUNNABLE 상태를 유지하지만 스케쥴링 큐로만 돌아가는 그래서 다른 스레드에게 CPU를 사용하도록 양보하게 했다. 만약, 양보할 스레드가 없다면 그냥 계속 내가 사용하게 되는 좋은 이점도 가지고 있었다.
다음 포스팅에선 지금까지 volatile 이라는 키워드를 사용하면서 이게 뭔지 몰랐다. 이 녀석에 대해 알아보자.
