Cách triển khai Priority Queue (Hàng đợi ưu tiên) trong Python

Giới thiệu

Module queue cung cấp cấu trúc dữ liệu FIFO (First-In-First-Out), LIFO (Last-In-First-Out) phù hợp cho lập trình đa luồng. Hàng đợi có thể truyền dữ liệu hoặc thông tin (phiên làm việc, đường dẫn, biến...) giữa luồng tạo và luồng tiêu dùng an toàn. Khóa (locking) thường được xử lý tự động cho người gọi.

Lưu ý: Bài viết giả định bạn đã hiểu bản chất của hàng đợi. Nếu chưa, hãy tham khảo tài liệu trước.

1. Hàng đợi FIFO cơ bản

import queue

fifo = queue.Queue()
# Đưa số vào hàng đợi
for i in range(5):
    fifo.put(i)

# Lấy số ra khỏi hàng đợi
print("Output")
while not fifo.empty():
    print(f" {fifo.get()} ")

Kết quả:

0 1 2 3 4

2. Hàng đợi LIFO cơ bản

import queue

lifo = queue.LifoQueue()
for i in range(5):
    lifo.put(i)

print("Output")
while not lifo.empty():
    print(f" {lifo.get()} ")

Kết quả:

4 3 2 1 0

Phần tử đưa vào cuối cùng được lấy ra đầu tiên.

3. Priority Queue (Hàng đợi ưu tiên)

Thứ tự xử lý đôi khi cần dựa trên độ ưu tiên của mục, không phải thứ tự tạo. Ví dụ: tác vụ kinh doanh quan trọng cần CPU ưu tiên cao hơn tác vụ in ấn của lập trình viên. PriorityQueue dùng thứ tự sắp xếp của nội dung để quyết định mục nào được lấy ra.

import queue
import threading

# Lớp đại diện cho công việc với độ ưu tiên
class Job:
    def __init__(self, priority, description):
        self.priority = priority
        self.description = description
        print('New job:', description)
    
    def __eq__(self, other):
        try:
            return self.priority == other.priority
        except AttributeError:
            return NotImplemented
    
    def __lt__(self, other):
        try:
            return self.priority < other.priority
        except AttributeError:
            return NotImplemented

# Tạo hàng đợi ưu tiên
q = queue.PriorityQueue()
q.put(Job(90, 'Developer-Print job'))      # Ưu tiên thấp
q.put(Job(2, 'Business-Report job'))       # Ưu tiên cao
q.put(Job(1, 'Business-Critical Job'))     # Ưu tiên cao nhất

# Hàm xử lý công việc
def process_job(q):
    while True:
        next_job = q.get()
        print(f" *** Đang xử lý công việc - {next_job.description}")
        q.task_done()

# Tạo 2 luồng worker
workers = [
    threading.Thread(target=process_job, args=(q,)),
    threading.Thread(target=process_job, args=(q,)),
]

# Khởi động và chờ hoàn thành
for w in workers:
    w.setDaemon(True)
    w.start()

q.join()

Kết quả:

New job: Developer-Print job
New job: Business-Report job
New job: Business-Critical Job
*** Đang xử lý công việc - Business-Critical Job
*** Đang xử lý công việc - Business-Report job
*** Đang xử lý công việc - Developer-Print job

4. Giải thích

Ví dụ có nhiều luồng tiêu dùng công việc, được xử lý dựa trên ưu tiên tại thời điểm get() được gọi. Thứ tự xử lý dựa trên tính quan trọng của nghiệp vụ, bất kể thứ tự đưa vào hàng đợi.