Java Queue vs BlockingQueue & Leetcode 215. Kth Largest Element in an Array

https://leetcode.com/problems/kth-largest-element-in-an-array/description/?envType=study-plan-v2&envId=top-interview-150 

Kth Largest Element in an Array - LeetCode

https://github.com/fineman999/algorithm_java/issues/13

215. Kth Largest Element in an Array · Issue #13 · fineman999/algorithm_java

---

목차

서론

알고리즘 문제를 우선순위 큐를 이용해 해결할 수 있다. 우선순위 큐는 요서의 우선순위에 따라 요소를 저장하고 처리하는 자료구조이다. 이러한 우선순위 큐를 구현하는 데 사용되는 자료구조로 힙을 사용한다.

자바에서는 어떤 구현체를 사용하는지 조사해 보자.

 

알고리즘 내용

정수 배열 번호와 정수 k가 주어지면 배열에서 k번째로 큰 원소를 반환한다.

k번째로 구별되는 원소가 아니라 정렬된 순서에서 k번째로 큰 원소임에 유의한다.

정렬을 사용하지 않고 해결해야 하는 문제다.

 

Example 1:

Input: nums = [3,2,1,5,6,4], k = 2
Output: 5

Example 2:

Input: nums = [3,2,3,1,2,4,5,5,6], k = 4
Output: 4

 

테스트 케이스 작성

class SolutionTest {

    private final Solution solution = new Solution();
    
    static Stream<Arguments> generateData() {
        return Stream.of(
                Arguments.of(new int[]{3,2,1,5,6,4}, 2, 5),
                Arguments.of(new int[]{3,2,3,1,2,4,5,5,6}, 4, 4)
        );
    }

    @ParameterizedTest
    @MethodSource("generateData")
    @DisplayName("findKthLargest 메서드는 int[] nums, int k 를 받아 k번째로 큰 수를 반환한다.")
    void findKthLargest(int[] nums, int k, int expected) {

        int kthLargest = solution.findKthLargest(nums, k);

        assertThat(kthLargest).isEqualTo(expected);
    }
}

 

테스트 케이스 실패

이제 프로덕션 코드를 작성해 보자.

알고리즘 풀이

public class Solution {
    public int findKthLargest(int[] nums, int k) {

        PriorityQueue<Integer> queue = new PriorityQueue<>(Comparator.reverseOrder());

        for (int num : nums) {
            queue.add(num);
        }

        while (k != 1 && !queue.isEmpty()) {
            queue.poll();
            k -= 1;
        }

        if (queue.isEmpty()) {
            return -1;
        }
        return queue.peek();
    }
}

알고리즘 문제를 풀기 위해 자바에서 제공해 주는 PriorityQueue를 사용하였으며, Comparator.reverseOrder()를 사용하여 우선순위 큐를 최대 힙으로 설정했다.(내림차순으로 저장하게 함)

 

while문을 통해 k를 1씩 감소시킴과 동시에 queue.poll()를 하여 k번째 값을 찾는다. 그리고 queue.peek()를 통해 우선순위 큐의 맨 위에 있는 요소를 반환한다.

 

이러한 방법으로 우선순위 큐를 사용하면 시간 복잡도가 O(NlogK)로 유지된다.

테스트 결과 성공

Leetcode 테스트 성공

 

우선순위 큐를 자바에서 사용하려면 PriorityQueue를 사용해야 한다. 그럼 자바에는 어떠한 큐를 구현되어 있을까?

 

Interface Queue<E>

PriorityQueue<E>

PriorityQueue<E>는 AbstractQueue<E>를 받고 Queue<E>를 implements 하고 있다.

 

public interface Queue<E> {
    boolean offer(E e);
    E poll();
    E peek();
    int size();
    boolean isEmpty();
}

 

큐는 추가 삽입, 추출 및 검사 작업을 제공한다.

이러한 메서드는 각각 실패하면 예외를 throw하는 형태와 특별한 값을 반환하는 형태로 존재한다.

	Throws exception	Returns special value
삽입	add(e)	offer(e)
제거	remove()	poll()
검사	element()	peek()

 

add(e) vs offer(e)

둘 다 3가지의 예외를 던진다. 

• ClassCastException(클래스 형변환 예외): 특정 요소의 클래스가 해당 큐에 추가할 수 없을 때 발생
• NullPointerException(널 포인터 예외): null 요소를 추가하려고 할 때 발생
• IllegalArgumentException(부적절한 인수 예외): 특정 요소의 속성이 해당 큐에 추가될 수 없을 때 발생한다.

add(e)는 추가로 주어진 시점에서 요소를 큐에 추가할 수 없는 경우 발생하는 IllegalStateException을 던진다. 

반면 offer(e)는 false를 반환하게 된다.

 

remove() vs poll()

remove()는 큐가 비어있을 때 remove()를 실행하면 값이 없다는 NoSuchElementException을 던진다.

poll()는 큐가 비어있을 때 poll()를 실행하면 null을 반환한다.

element()와 peek()도 위와 같은 식으로 진행된다.

 

자 이제 implements 하고 있는 AbstractQueue<E>를 살펴보자

 

abstrac class AbstractQueue<E>

// AbstractCollection 을 상속받는다.(생략)
public abstract class AbstractQueue<E> implements Queue<E> {
    protected AbstractQueue() {
    }

    public boolean add(E e) {
        if (offer(e)) {
            return true;
        } else {
            throw new IllegalStateException("Queue full");
        }
    }

    public E remove() {
        E x = poll();
        if (x != null) {
            return x;
        } else {
            throw new NoSuchElementException();
        }
    }

    public E element() {
        E x = peek();
        if (x != null) {
            return x;
        } else {
            throw new NoSuchElementException();
        }
    }

    public void clear() {
        while (poll() != null) {
            ;
        }
    }

    public boolean addAll(Collection<? extends E> c) {
        if (c == null) {
            throw new NullPointerException();
        } else if (c == this) {
            throw new IllegalArgumentException();
        } else {
            boolean modified = false;

            for (E e : c) {
                if (add(e)) {
                    modified = true;
                }
            }

            return modified;
        }
    }
}

이 클래스는 일부 Queue의 뼈대(skeletal)를 제공한다.

앞서 말한 add, remove, element에 대한 예외 처리를 이곳에서 한다.

 

class PriortyQueue<E>

자바 문서에서는 요약하자면 아래와 같이 PriorityQueue를 설명한다.

 

PriorityQueue는 우선순위 힙을 기반으로 한 무제한 크기의 큐이다.

PriorityQueue는 balanced binary heap으로 표현된다.

PriorityQueue는 무제한이지만 내부 용량이 큐에 저장된 요소를 저장하는 데 사용되는 배열의 크기를 조절한다.

이 크기는 항상 큐 크기 이상이어야 한다. 요소가 우선순위 큐에 추가됨에 따라 용량은 자동으로 증가한다.

그리고 이 구현은 동기화되지 않았음을 주의해야 한다.

여러 스레드가 큐를 수정하는 경우 그중 하나의 스레드라도 큐를 수정하면 PriorityQueue 인스턴스에 동시에 액세스 해서는 안된다. 대신 스레드 안전한 'java.util.concurrent.PriorityBlockingQueue' 클래스를 사용하면 된다.

 

그럼 PriorityBlockingQueue는 뭐길래 스레드 안전하다고 할까?

어떤 인터페이스와 클래스를 상속받을까?

 

PriorityBlockingQueue<E>

PriorityQueue<E>와 비슷하지만 추가로 BlockingQueue<E>를 상속받는다.

BlockingQueue<E>를 알아볼 필요가 있다.

 

interface BlockingQueue<E>

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;
    int remainingCapacity();
    boolean remove(Object o);
    boolean contains(Object o);
    int drainTo(Collection<? super E> c);
    int drainTo(Collection<? super E> c, int maxElements);

}

BlockingQueue는 요소를 검색할 때 큐가 비어 있을 때 대기하고, 요소를 저장할 때 큐에 공간이 생길 때까지 대기하는 작업을 지원하는 큐이다.

 

	Thorws exception	Returns special value	Blocks	Times out
삽입	add(e)	offer(e)	put(e)	offer(e, time, unit)
제거	remove()	poll()	take()	poll(time, unit)
검사	element()	peek()	사용 불가	사용 불가

interface Queue에서 2가지 형태가 더 추가되었다.

BlockingQueue의 특징으로 다음과 같다.

• BlockingQueue는 null 요소를 허용하지 않는다.
  • 구현체는 null을 추가하거나 넣거나 제공하려고 시도할 때 NullPointerException을 던진다.
• BlockingQueue는 용량 제한이 있는 경우도 있을 수 있다.
  • 내재적인 용량 제한이 없는 BlockingQueue는 항상 Integer.MAX_VALUE의 남은 용량을 보고한다.
  • 남은 용량을 확인하는 remainingCapacity가 있다.
    • 하지만 remainingCapacity를 검사하여 요소를 삽입할 시도가 성공할 것인지는 항상 알 수 없다. 다른 스레드가 요소를 삽입하거나 제거할 예정일 수 있기 때문이다.
• BlockingQueue 구현체는 스레드 안전하다.
  • 모든 큐 메서드는 내부 락 또는 다른 형태의 동시성 제어를 사용하여 원자적으로 효과를 낼 수 있다.
  • 그러나 addAll 같은 대량 Collection 작업은 구현에 따라 원자적으로 수행되지 않을 수 있다.(부분적으로 추가 후 실패 할 수 있음)

그럼 Block이란건 뭘 의미할까?

블록 된다라는 뜻은 해당 메서드가 실행되는 동안 특정 조건을 충족하지 않으면 메소드가 일시 중지되고 다른 조건이나 자원을 기다린다는 것을 의미한다. 이러한 블록은 프로그램의 정상적인 흐름을 일시적으로 중단시키는 것을 나타내며, 대개는 어떤 조건이나 자원이 사용 가능해질 때까지 기다리는 동안 다른 작업을 수행할 수 있도록 한다.

 

이제 메서드 형태 중 Blocks와 timeout에 대해 알아보자.

 

put(e) vs offer(e, time, unit)

• put(e): 해당 큐에 지정된 요소를 삽입하고 필요한 경우 공간이 사용 가능할 때까지 대기한다.
  • 기존의 IllegalArgumentException, NullPointerException, ClassCastException 예외를 가지고 추가로 스레드가 대기상태에 있을 때, 다른 스레드가 그 스레드를 인터럽트 하려고 할 때 InterruptedException을 던지는 예외를 가지고 있다.
• offer(e, time, unit): 해당 큐에 지정된 요소를 삽입하고, 필요한 경우 공간이 사용 가능해질 때까지 지정된 대기 시간 동안 대기한다. 성공하면 true를 반환하고, 지정된 대기 시간이 경과되어 공간이 사용 가능해지기 전에 실패하면 false를 반환한다.
  • 매개변수: 
    • e: 추가할 요소
    • timeout: 포기하기 전 대기할 시간
    • unit: timeout 매개변수를 해석하는 데 사용되는 시간 단위 (예: 초, 분 등)
  • 예외는 put(e)와 같다.

take() vs poll(time, unit)

• take(): 해당 큐의 맨 앞에 있는 요소를 가져와 제거하고 필요한 경우 해당 요소가 사용 가능할 때까지 대기한다.
  • 예외는 InterruptedException을 던진다.
• poll(time, unit): 해당 큐의 맨 앞에 있는 요소를 가져와 제거하고, 필요한 경우 해당 요소가 사용 가능해질 때까지 지정된 대기 시간 동안 대기한다. 만약 지정된 대기 시간이 경과되어 요소가 사용가능하지 않으면 null을 반환한다.

 

이전 알고리즘 풀이에서는 PriorityQueue를 이용하였다. 이번에 PriorityBlockingQueue를 이용하여 알고리즘을 실행해 보자.

 

알고리즘 실행 - PriorityBlockingQueue

public class BlockingQueueSolution implements Solution {

    @Override
    public int findKthLargest(int[] nums, int k) {

        PriorityBlockingQueue<Integer> blockingQueue = new PriorityBlockingQueue<>(nums.length, Comparator.reverseOrder());

        for (int num : nums) {
            blockingQueue.put(num);
        }

        while (k != 1 && !blockingQueue.isEmpty()) {
            try {
                blockingQueue.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            k -= 1;
        }

        if (blockingQueue.isEmpty()) {
            return -1;
        }

        return blockingQueue.peek();
    }
}

PriorityBlockingQueue 실행 결과

메모리는 거의 변화가 없지만 런타임 실행 속도는 11ms 정도 차이가 발생했다. 

단일 스레드 환경이라도 블록을 이용하는 put과 take를 사용하면 10ms 정도 느린 것을 확인할 수 있었다.

 

Note that this implementation is not synchronized. Multiple threads should not access a PriorityQueue instance concurrently if any of the threads modifies the queue. Instead, use the thread-safe java.util.concurrent.PriorityBlockingQueue class.

 

자바 docs에서 말한 것처럼 멀티 스레드 환경에서는 PriorityBlockingQueue를 이용하고 단일 스레드 환경에서는 좀 더 빠른 PriorityQueue를 이용하자.

 

---

참고

https://docs.oracle.com/javase/8/docs/api/java/util/Queue.html

Queue (Java Platform SE 8 )

https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/BlockingQueue.html

BlockingQueue (Java Platform SE 8 )