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

TL;DR: 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 lấy ra trước dù insert theo thứ tự nào. Bên dưới là binary heap: cây nhị phân cân bằng hoàn toàn lưu trong array, mỗi node nhỏ hơn hoặc bằng hai con (min-heap). Hai thao tác sift-up (khi insert) và sift-down (khi extract-min) khôi phục tính chất heap trong O(log n) bằng cách đi dọc chiều cao cây. Iteration không đảm bảo sorted — phải dùng poll() liên tiếp. Top-K pattern: max-heap kích thước K, O(n log k) time, O(k) space.

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ộ danh sách, 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 — một cây nhị phân cân bằng hoàn toàn, lưu trong array, với tính chất "mỗi node nhỏ hơn hoặc bằng các con" (min-heap).

Bài này giới thiệu PriorityQueue qua cấu trúc binary heap, sift-up/sift-down, và use case (top-K, Dijkstra, scheduler) — chứng minh buildHeap O(n) và heapsort sẽ dive vào ở Thuật toán Cốt lõi — Module 2: Heap & Heapsort.

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 và complexity

Để 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 dùng pseudocode:

-- Min-heap mặc định: poll() trả phần tử nhỏ nhất
H <- MinHeap rỗng
H.offer(3)
H.offer(1)
H.offer(2)

-- poll() lấy minimum mỗi lần:
H.poll()  -- trả 1
H.poll()  -- trả 2
H.poll()  -- trả 3

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

MinHeap mặc định: phần tử nhỏ nhất theo natural ordering sẽ được poll() trước.

-- Min-heap mặc định (natural ordering tăng dần):
minHeap <- MinHeap rỗng
minHeap.offer(5); minHeap.offer(1); minHeap.offer(3)
minHeap.poll()  -- trả 1 (nhỏ nhất)

-- Max-heap: đảo comparator
maxHeap <- MaxHeap rỗng  -- comparator đảo ngược
maxHeap.offer(5); maxHeap.offer(1); maxHeap.offer(3)
maxHeap.poll()  -- trả 5 (lớn nhất)

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

-- Task với priority: số nhỏ hơn = cấp bách hơn
Task = { name, priority }

-- Min-heap theo priority (priority 1 chạy trước priority 10):
scheduler <- MinHeap theo Task.priority

scheduler.offer(Task("backup",  priority=10))
scheduler.offer(Task("alert",   priority=1))
scheduler.offer(Task("report",  priority=5))

scheduler.poll()  -- trả Task("alert", 1) -- priority thấp nhất = cấp bách nhất

4. Cấu trúc bên dưới — binary heap và sift mechanism

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. Với n phần tử, phần tử tại index i có:

• parent tại index (i - 1) / 2
• left child tại index 2*i + 1
• right child tại index 2*i + 2

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

Cây nhị phân:            Array tương ứng:
        1                index: [0, 1, 2, 3, 4, 5]
       / \               value: [1, 3, 2, 7, 5, 4]
      3   2
     / \ /
    7  5 4

Tính chất heap: mỗi node <= cả hai con
=> root luôn là phần tử nhỏ nhất

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 hai thao tác khôi phục tính chất này sau mỗi thay đổi:

4.1 Sift-up — khi insert

-- offer(x): chèn x vào heap
offer(x):
    A.append(x)           -- đặt x ở cuối array (lá mới)
    i <- A.length - 1     -- index của x

    -- Sift-up: so sánh với parent, swap nếu nhỏ hơn
    while i > 0:
        parent <- (i - 1) / 2
        if A[i] < A[parent]:
            swap(A[i], A[parent])
            i <- parent
        else:
            break         -- tính chất heap đã thỏa, dừng

Time: O(log n) — số swap tối đa bằng chiều cao cây Space: O(1)

Ví dụ sift-up khi offer(0) vào heap [1, 3, 2, 7, 5, 4]:

• Đặt 0 ở index 6 (con của index 2, giá trị 2) → 0 nhỏ hơn 2, swap → 0 lên index 2
• 0 là con của root (index 0, giá trị 1) → 0 nhỏ hơn 1, swap → 0 thành root
• Đúng 2 lần swap.

4.2 Sift-down — khi extract-min

-- poll(): lấy và xóa phần tử nhỏ nhất (root)
poll():
    if A rỗng: return NIL
    result <- A[0]            -- lưu kết quả (root = minimum)
    A[0] <- A[A.length - 1]  -- đưa phần tử cuối lên root
    A.removeLast()

    -- Sift-down: so sánh với con nhỏ hơn, swap nếu lớn hơn
    i <- 0
    while true:
        left  <- 2*i + 1
        right <- 2*i + 2
        smallest <- i

        if left < A.length and A[left] < A[smallest]:
            smallest <- left
        if right < A.length and A[right] < A[smallest]:
            smallest <- right

        if smallest = i:
            break             -- không cần swap thêm, dừng
        swap(A[i], A[smallest])
        i <- smallest

    return result

Time: O(log n) Space: O(1)

Vì sao O(log n)? Cả hai thao tác chỉ đi dọc theo chiều cao cây. Heap luôn là cây nhị phân cân bằng hoàn toàn (lấp đầy trái sang phải) nên chiều cao bằng log₂(n). Số swap tối đa = số tầng = O(log n) — đây là lý do cốt lõi MinHeap đạt O(log n) mà không cần scan.

Cấu trúc này lưu trong array, không dùng pointer — cache-friendly và không tốn overhead per-node như linked tree. Chứng minh buildHeap O(n) (xây heap từ collection nhanh hơn n lần insert) được phân tích đầy đủ ở Thuật toán Cốt lõi — Module 2: Heap & Heapsort.

5. Sơ đồ sift-up và sift-down

graph TD
    subgraph BEFORE["Trước offer(0)"]
        R1(["1\n(root)"]) --> L1(["3"])
        R1 --> RR1(["2"])
        L1 --> LL1(["7"])
        L1 --> LR1(["5"])
        RR1 --> RL1(["4"])
        RR1 --> NEW(["0\n(mới thêm)"])
    end

    subgraph AFTER["Sau sift-up: 0 là root"]
        R2(["0\n(root)"]) --> L2(["3"])
        R2 --> RR2(["1"])
        L2 --> LL2(["7"])
        L2 --> LR2(["5"])
        RR2 --> RL2(["4"])
        RR2 --> OLD(["2"])
    end

Sift-up: 0 nhỏ hơn parent (2) → swap; 0 nhỏ hơn parent mới (1) → swap lên root. Đúng 2 bước, chiều cao cây là 3.

6. 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ộ → 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.

MinHeap 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.

-- topKSmallest(nums[], k): trả k phần tử nhỏ nhất
topKSmallest(nums, k):
    H <- MaxHeap rỗng   -- top của H là phần tử lớn nhất trong k nhỏ nhất đã thấy

    for each n trong nums:
        H.offer(n)
        if H.size() > k:
            H.poll()    -- loại phần tử lớn nhất — nó không thuộc top-k

    -- H chứa đúng k phần tử nhỏ nhất
    result <- array[k]
    for i <- k-1 downto 0:
        result[i] <- H.poll()   -- drain theo thứ tự tăng dần
    return result

Time: O(n log k) Space: O(k)

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ố.

7. Pitfall tổng hợp

Pitfall 1 — Iteration không cho sorted output

-- SAI: for-each duyệt theo heap order, KHÔNG phải sorted
H <- MinHeap {3, 1, 2}
for each x trong H:
    print(x)    -- có thể in: 1 3 2 (heap order — không đoán được)

-- ĐÚNG: dùng poll() liên tiếp
while H không rỗng:
    print(H.poll())   -- in: 1 2 3 (sorted ascending)

Lý do: iterator 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

-- SAI: phép trừ có thể overflow với giá trị cực đại
compare(a, b): return a - b
-- Nếu a = MAX_INT và b = -1:
-- a - b = MAX_INT + 1 -- overflow, trở thành giá trị âm!
-- Comparator trả âm sai, thứ tự heap bị lệch

-- ĐÚNG: dùng Integer.compare hoặc Comparator.comparingInt
compare(a, b): return Integer.compare(a, b)

Pitfall 3 — Heap không thread-safe

-- SAI: nhiều thread cùng offer/poll làm hỏng heap
-- Thread 1: H.offer(task1)
-- Thread 2: H.poll()
-- Race condition: array nội bộ ở trạng thái không nhất quán

-- ĐÚNG option 1: PriorityBlockingQueue (blocking, thread-safe)
H <- PriorityBlockingQueue rỗng

-- ĐÚNG option 2: external lock bao quanh mọi thao tác
synchronized(lock):
    H.offer(task)

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

8. Ứng dụng thực tế

Dijkstra shortest path (Thuật toán Cốt lõi — Module 3): mỗi vertex có distance estimate ban đầu là vô cực, trừ source = 0. MinHeap chứa các (distance, vertex) pair, poll ra vertex có distance nhỏ nhất để relax tiếp. Không dùng heap thì phải scan toàn bộ V vertices mỗi vòng lặp — O(V²). Dùng heap giảm xuống O((V+E) log V) — với graph thưa, 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. Heap là "clock" của toàn bộ simulation.

9. Deep Dive

10. Tóm tắt

• MinHeap (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 đảo ngược.
• Binary heap lưu trong array: parent tại (i-1)/2, left child tại 2i+1, right child tại 2i+2. Cache-friendly, không pointer overhead.
• Sift-up (insert): đặt phần tử mới ở cuối, swap với parent cho đến khi tính chất heap thỏa — O(log n).
• Sift-down (extract-min): đặt phần tử cuối lên root, swap với con nhỏ hơn cho đến khi tính chất heap thỏa — O(log n).
• Iteration không sorted: 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.
• 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.

11. Tự kiểm tra

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