[pj2] Java Concurrency Utilities

1. BlockingQueue

BlockingQueue

• java.util.concurrent.BlockingQueue에 있는 interface
• 큐로 elements를 넣거나 뺄 때 스레드들을 blocking할 수 있음
  • 큐에 자리가 없는데 값을 넣으려고 할 때 자리가 생길 때까지 그 스레드를 blocking할 수 있음

 

• 보통 producer/consumer를 다루는데 사용됨
  • producer 스레드는 새로운 객체를 만들고, BlockingQueue에다 넣음.
    • 큐가 가득차면, producer 스레드는 block됨.
    • consumer 스레드에서 객체를 빼가서 큐에 자리가 생길때까지 block! 
  • consumer 스레드는 BlockingQueue에서 객체를 빼감
    • 큐가 비었는데 객체를 빼려고 하면, consumer 스레드는 block됨
    • porducer 스레드에서 객체를 넣을 때까지 block

thread1: producer, thread2:consumer

 

•  Method
  • null을 Queue에 넣을 수 없음. 넣으면 NullPointerException 발생
  • Queue 내의 모든 요소에 access 가능.
    • 'remove(o)'로 queue 내의 객체 삭제 가능. 근데 비효율적이니까 반드시 써야하는 경우 아니면 쓰지 마라
  • 4가지 type의 method가 있는데, 요청된 작업을 즉시 수행할 수 없는 경우 각각 다르게 동작함

	Throws Exception	Special Value	Blocks	Times Out
Insert	add(o)	offer(o)	put(o)	offer(o, timeout, timeunit)
Remove	remove(o)	poll()	take()	poll(timeout, timeunit)
Examine	element()	peek()

1. Throws Exception: exception 발생
2. Special Value: special value (보통 T/F) 리턴됨
3. Blocks: 메소드 call이 block됨
4. Time Out: 메소드 call이 block되는데, timeout 시간만큼만 대기. 작업 성공 여부를 나타내는 special value (보통 T/F) 리턴

 

• BlockingQueue Example
  • add(): 매개변수로 전달된 값을 BlockingQueue에 추가. 공간이 없으면 IlallegalStateException throws
  • offer(): 매개변수로 전달된 값을 BlockingQueue에 추가. 공간이 없으면 false 반환
  • offer(long millis, TImeUnit timeUnit): 매개변수로 전달된 값을 BlockingQueue에 추가. 공간이 제한 시간내에 없는 경우 false 반환
  • put(): 매개변수로 전달된 값을 BlockingQueue에 추가. 공간이 없으면, 공간이 생길 때까지 put()을 호출하는 스레드를 block
  • take(): BlockingQueue에서 첫번째 요소를 제거. 공간이 비어있으면, 삽입될 때까지 take()를 호출하는 스레드를 lbock
  • poll(): BlockingQueue에서 첫번째 요소를 제거. 공간이 비어있으면, null()을 반환
  • poll(long timeMillis, TimeUnit timeUnit): BlockingQueue에서 첫번째 요소를 제거. 공간이 비어있는데, 지정된 제한 시간 내에 사용할 수 있는 값이 생기지 않으면 null 반환
  • remove(Object o): o가 BlockingQueue에 있으면 제거. o가 여러개가 있으면 하나만 제거. 값이 제거되면 true, 아니면 false반환 
  • peek(): 값 제거 없이 BlockingQueue의 첫 번째 값 반환. 값이 없으면 null 반환.
  • element(): 값 제거 없이 BlockingQueue의 첫 번째 값 반환. 값이 없으면 NoSuchElementException throws
  • contains(Object o): BlockingQueue에 o가 있으면 true 반환. 
  • drainTo(Collection dest): BlockingQueue 내의 값들을 모두 destination Collection으로 내보냄
  • drainTo(Collection dest, int maxElements): BlockingQueue 내의 maxElements개의 값들을 destination Collection으로 내보냄 
  • size(): BlockingQueue 내의 값들의 수를 리턴
  • remainingCapacity: BlockingQueue의 free space 리턴

 

• BlockingQueue를 이용해 구현한 여러 Queue들이 존재함
  • ArrayBlockingQueue
  • DelayQueue
  • LinkedBlockingQueue
  • LinkedBlockingDeque
  • LinkedTransferQueue
  • PriorityBlockingQueue
  • SynchrounousQueue

 

ArrayBlockingQueue

• BlockingQueue interface를 implement한 class
• array 내에 element를 저장하는 유한한 크기의 Queue
  • 동시에 저장할 수 있는 element 수의 상한이 정해져있음.
    • 초기화할 때 상한의 크기를 정할 수 있는데 한번 정하면 수정 불가
  • Queue니까 FIFO 따름

 

2. ReadWriteLock

• java.util.concurrent.locks.ReadWriteLock에 있는 interface임
• 스레드 lock 메커니즘
  • 한번에 여러 메소드들이 특정 리소스를 read하고,
  • 한번에 하나의 메소드가 특정 리소스를 write할 수 있게 만듦
  • 특정 리소스에 대한 read, write가 동시에 일어날 수 없음

 

• ReadWriteLock Locking Rules
  • Read Lock
    • write lock을 가진 스레드나 write lock을 요청한 스레드가 아닌 스레드
    • 여러 스레드가 read lock을 가질 수 있음
  • Write Lock
    • read lock이나 write lock을 가지지 않은 스레드
    • 한번에 한 스레드만 write lock을 가질 수 있음

 

• Implementations
  • ReentrantReadWriteLock

 

3. AtomicInteger

• java.util.concurrent.atomic에 있는 class
• atomic하게 읽거나 쓸 수 있는, atomic operation을 포함하는 int 변수 제공

 

• AtomicInteger 만들기
  • 매개변수로 초기값 설정 가능
  • 매개변수로 설정 안하면 0으로 초기화

AtomicInteger atomicInteger1 = new AtomicInteger(); // 0으로 초기화
AtomicInteger atomicInteger2 = new AtomicInteger(123); // 123으로 초기화

 

• AtomicInteger 값 설정
  • set() 메소드로 값 바꾸기 가능

atomicInteger2.set(234); // 123-->234

 

• Compare and Set the AtomicInteger Value
  • AtomicInteger 인스턴스의 현재 값과 expectedValue를 비교해,
  • 두 값이 같으면 새로운 값으로 설정

AtomicInteger atomicInteger = new AtomicInteger(123);

int expectedValue = 123;
int newValue      = 234;
atomicInteger.compareAndSet(expectedValue, newValue); // 123-->234

 

• Adding to the AtomicInteger Value
  • addAndGet(): AtomicInteger와 매개변수로 넘어간 수를 더한 값을 반환
  • getAndAdd(): AtomicInteger와 매개변수로 넘어간 수를 더하는데, 더하기 전 원래 AtomicInteger 값 반환
  • getAndIncrement(): getAndAdd()에서 매개변수로 1이 넘어간 것 
  • incrementAndGet(): addAndGet()에서 매개변수로 1이 넘어간 것

AtomicInteger atomicInteger = new AtomicInteger(); // 0

System.out.println(atomicInteger.getAndAdd(10)); // 0
System.out.println(atomicInteger.addAndGet(10)); // 20

 

• Subtracting From the AtomicInteger Value
  • decrementAndGet(): incrementAndGet()의 빼기 버전, 1 뺌
  • getAndDecrement(): getAndIncrement()의 빼기 버전, 1 뺌

 

4. CyclicBarrier

• java.util.concurrent.CyclicBarrier에 있는 class
• synchronization 메커니즘
• 일종의 벽인데, 스레드가 다시 실행되려면 모든 스레드들이 벽에 도달할 때까지 기다려야 함

 

• CyclicBarrier 생성

CyclicBarrier barrier = new CyclicBarrier(2); // 2개의 thread를 기다림

 

• Waiting at a CyclicBarrier
  • 스레드가 CyclicBarrier에서 기다림
  • timeout을 설정하면, 스레드가 전부 CyclicBarrier에 도달하지 않아도 제한된 시간이 지나면 스레드가 released될 수 있음
  • CyclicBarrier에서 스레드 A가 탈출하는 법
    • 마지막 스레드가 도착 (calls await())
    • 다른 스레드가 스레드 A를 interrupt함
    • 기다리고 있는 다른 스레드가 interrupt됨
    • 기다리고 있는 다른 스레드가 timeout됨
    • 외부에서 CyclicBarrier.reset()이 호출됨

barrier.await();
//barrier.await(10, TimeUnit.SECONDS);

 

• CyclicBarrier Action
  • 마지막 스레드가 도착했을 때 실행되는 동작을 지정할 수 있음

Runnable barrierAction = ...;
CyclicBarrier barrier = new CyclicBarrier(2, barrierAction);