PriorityQueue — Top-K, scheduler, Dijkstra cần queue thứ tự

Task scheduler nhận 1 triệu task với priority khác nhau từ 1 đến 100. Mỗi khi một worker rảnh, scheduler phải chọn task có priority cao nhất đang chờ. Cách đơn giản nhất: duyệt toàn bộ ArrayList, tìm max — O(n) mỗi lookup. Với 1 triệu task và 1000 lookup mỗi giây, mỗi lookup tốn 1ms — 1000 lookup chiếm hết một giây CPU chỉ để tìm task tiếp theo. Worker chờ, throughput sụt.

Cần cấu trúc dữ liệu hỗ trợ O(log n) insert và O(log n) extract theo thứ tự priority — không cần scan toàn bộ. PriorityQueue trong Java làm đúng điều này, dùng binary heap bên dưới.

Bài này giới thiệu PriorityQueue qua API và use case (top-K, Dijkstra, scheduler) — chi tiết binary heap structure để Module 4 lesson 05 dive vào.

1. Analogy — Phòng khám ưu tiên

Phòng khám có nhiều bệnh nhân đang chờ, mỗi người có mức độ ưu tiên khác nhau: cấp cứu ưu tiên cao nhất, tái khám thông thường ưu tiên thấp. Y tá không phục vụ theo thứ tự đến — y tá luôn gọi bệnh nhân có mức ưu tiên cao nhất tiếp theo.

Không cần scan toàn bộ phòng chờ mỗi lần gọi — danh sách đã được tổ chức sao cho bệnh nhân ưu tiên cao nhất luôn ở vị trí đầu. Khi có bệnh nhân mới đến, họ được chèn vào đúng vị trí theo priority.

2. API qua góc người dùng

Để dùng PriorityQueue hiệu quả, không cần biết binary heap là gì — chỉ cần hiểu bốn operation và complexity của chúng.

Demo cơ bản:

PriorityQueue<Integer> pq = new PriorityQueue<>();
pq.offer(3);
pq.offer(1);
pq.offer(2);

// poll() extracts minimum each time -- default is min-heap
while (!pq.isEmpty()) {
    System.out.println(pq.poll()); // prints: 1, 2, 3 (sorted ascending)
}

3. Min-heap vs max-heap và custom comparator

Java PriorityQueue mặc định là min-heap: phần tử nhỏ nhất theo natural ordering sẽ được poll() trước.

// Default: min-heap -- natural ordering ascending
PriorityQueue<Integer> minHeap = new PriorityQueue<>();
minHeap.offer(5); minHeap.offer(1); minHeap.offer(3);
minHeap.poll(); // returns 1 (minimum)

// Max-heap: reverse the comparator
PriorityQueue<Integer> maxHeap = new PriorityQueue<>(Comparator.reverseOrder());
maxHeap.offer(5); maxHeap.offer(1); maxHeap.offer(3);
maxHeap.poll(); // returns 5 (maximum)

Với custom object, dùng comparator để định nghĩa thứ tự:

record Task(String name, int priority) {}

// Lower priority number = higher urgency (e.g. priority 1 runs before priority 10)
PriorityQueue<Task> scheduler = new PriorityQueue<>(
    Comparator.comparingInt(Task::priority)
);
scheduler.offer(new Task("backup",   10));
scheduler.offer(new Task("alert",     1));
scheduler.offer(new Task("report",    5));
scheduler.poll(); // returns Task("alert", 1) -- lowest number = highest urgency

4. Cấu trúc bên dưới — binary heap (preview)

PriorityQueue lưu dữ liệu trong một array thông thường, nhưng tổ chức theo cây nhị phân ngầm định:

Min-heap với các phần tử [1, 3, 2, 7, 5, 4]:

         1                   array representation:
        / \                  index: [0, 1, 2, 3, 4, 5]
       3   2                 value: [1, 3, 2, 7, 5, 4]
      / \ /
     7  5 4                  parent of i: (i-1)/2
                             children of i: 2*i+1 (left), 2*i+2 (right)

Tính chất heap: mỗi node nhỏ hơn hoặc bằng cả hai con — đảm bảo root luôn là phần tử nhỏ nhất. Insert và extract-min dùng thao tác "sift-up" và "sift-down" trên array này để khôi phục tính chất heap sau mỗi thay đổi.

Cấu trúc này là lưu trong array, không dùng pointer — cache-friendly và không tốn overhead per-node như linked tree. Chi tiết sift-up, sift-down, và buildHeap O(n) để Module 4 lesson 05 phân tích.

5. Top-K pattern

Tìm K phần tử nhỏ nhất trong stream 1 triệu số nguyên. Đây là bài toán cổ điển xuất hiện trong interview và production analytics.

Naive approach: sort toàn bộ array → O(n log n) time, O(n) space. Không thể dùng với streaming data vì phải load toàn bộ vào RAM.

PriorityQueue approach: dùng max-heap kích thước K.

Ý tưởng: heap luôn giữ K phần tử nhỏ nhất đã thấy cho đến nay. Top của max-heap là phần tử lớn nhất trong nhóm K đó — gọi là "threshold". Khi có phần tử mới nhỏ hơn threshold, nó đủ điều kiện vào nhóm K: insert vào heap và pop threshold ra.

public int[] topKSmallest(int[] nums, int k) {
    // Max-heap of size k: top = largest among the k smallest seen so far
    PriorityQueue<Integer> maxHeap = new PriorityQueue<>(Comparator.reverseOrder());

    for (int n : nums) {
        maxHeap.offer(n);
        if (maxHeap.size() > k) {
            maxHeap.poll(); // evict the largest -- it does not belong in top-k
        }
    }

    // Drain heap into result array in ascending order
    int[] result = new int[k];
    for (int i = k - 1; i >= 0; i--) {
        result[i] = maxHeap.poll();
    }
    return result;
}

Complexity: O(n log k) time — mỗi trong n phần tử tốn O(log k) cho insert/poll vào heap kích thước K. O(k) space — heap không bao giờ vượt kích thước K, không cần load toàn bộ n vào RAM. Với n = 1 triệu và k = 10, thực tế gần O(n) vì log 10 là hằng số.

LeetCode 215 (Kth Largest Element in an Array) dùng cùng pattern với min-heap thay max-heap.

6. Pitfall tổng hợp

Pitfall 1 — Iteration không cho sorted output

PriorityQueue<Integer> pq = new PriorityQueue<>();
pq.offer(3); pq.offer(1); pq.offer(2);

// WRONG: for-each iterates in heap order, NOT sorted order
for (int x : pq) {
    System.out.print(x + " "); // may print: 1 3 2 (heap order, unpredictable)
}

// CORRECT: use poll() repeatedly
while (!pq.isEmpty()) {
    System.out.print(pq.poll() + " "); // prints: 1 2 3 (sorted ascending)
}

Lý do: iterator() của PriorityQueue duyệt backing array theo thứ tự index — đây là heap order, không phải sorted order. Chỉ poll() mới đảm bảo thứ tự đúng.

Pitfall 2 — Comparator trả a - b overflow

// WRONG: subtraction can overflow with extreme int values
PriorityQueue<Integer> pq1 = new PriorityQueue<>((a, b) -> a - b);
// If a = Integer.MAX_VALUE (2147483647) and b = -1:
// a - b = 2147483647 - (-1) = 2147483648 -- overflows to -2147483648
// Comparator returns negative -> wrong ordering

// CORRECT: use Integer.compare or Comparator.comparingInt
PriorityQueue<Integer> pq2 = new PriorityQueue<>(Integer::compare);
PriorityQueue<Integer> pq3 = new PriorityQueue<>(Comparator.comparingInt(x -> x));

Pitfall 3 — PriorityQueue không thread-safe

// WRONG: concurrent offer/poll corrupts the heap
PriorityQueue<Task> shared = new PriorityQueue<>();
// Thread 1: shared.offer(new Task(...));
// Thread 2: shared.poll();
// Race condition: internal array can be in inconsistent state

// CORRECT option 1: PriorityBlockingQueue for concurrent producer-consumer
PriorityBlockingQueue<Task> concurrent = new PriorityBlockingQueue<>();

// CORRECT option 2: external synchronization (if access pattern is simple)
// synchronized(lock) { shared.offer(task); }

PriorityBlockingQueue là unbounded blocking queue với ordering đảm bảo thread-safe. poll() block khi empty (khác PriorityQueue.poll() trả null). Phù hợp cho concurrent task scheduler pattern.

7. Applied to tech

Dijkstra shortest path (Module 5): mỗi vertex có distance estimate ban đầu là vô cực, trừ source = 0. PriorityQueue chứa các (distance, vertex) pair, poll ra vertex có distance nhỏ nhất để relax tiếp. Không dùng PQ thì phải scan toàn bộ V vertices mỗi vòng lặp — O(V²). Dùng PQ giảm xuống O((V+E) log V) — với graph thưa (sparse), cải thiện đáng kể.

Task scheduler / cron daemon — danh sách job được đưa vào min-heap theo trigger time. Top của heap luôn là job cần chạy sớm nhất tiếp theo. Scheduler peek() để biết thời gian chờ đến job kế, poll() khi đến giờ. Không cần scan toàn bộ danh sách job mỗi tick.

Kafka producer batch — messages được nhóm theo partition, flush theo priority dựa trên oldest unflushed timestamp (TTL pattern). Min-heap theo timestamp đảm bảo partition chờ lâu nhất được flush trước — tránh starvation khi throughput cao.

Event-driven simulation (discrete event simulation) — tất cả future events được đặt vào min-heap theo timestamp. Mỗi bước: poll() event sớm nhất, xử lý, và các event mới phát sinh từ đó được offer() vào heap. PQ là "clock" của toàn bộ simulation.

8. Deep Dive

9. Tóm tắt

• PriorityQueue không phải FIFO — phần tử có priority cao nhất (mặc định: giá trị nhỏ nhất) luôn được poll() trước. O(log n) insert và extract, O(1) peek.
• Mặc định min-heap: poll() trả minimum. Max-heap dùng Comparator.reverseOrder(). Custom object dùng Comparator.comparingInt(...).
• Iteration không sorted: for-each duyệt heap order — dùng poll() liên tiếp để lấy thứ tự sorted.
• Comparator a - b overflow: dùng Integer.compare(a, b) hoặc Comparator.comparingInt(...) để tránh integer overflow với giá trị cực đại.
• Không thread-safe: dùng PriorityBlockingQueue cho concurrent producer-consumer.
• Top-K pattern: max-heap kích thước K, O(n log k) time, O(k) space — không cần load toàn bộ n vào RAM.
• Use case thực tế: Dijkstra shortest path, task scheduler, event-driven simulation, producer batch flush theo TTL.

10. Tự kiểm tra

Bài tiếp theo: Mini-challenge: LRU Cache thủ công