วิธีนำ Priority Queue มาใช้ใน Python

บทนำ

โมดูล queue มีโครงสร้างข้อมูล FIFO (First-in, First-out) และ LIFO (Last-in, First-out) เหมาะสำหรับการเขียนโปรแกรมมัลติเธรด คิวใช้ส่งข้อมูลหรือข้อมูลหลากหลาย (session details, paths, variables) ระหว่าง creator และ consumer threads อย่างปลอดภัย โดย Locking จะถูกจัดการให้โดยอัตโนมัติ

หมายเหตุ: บทความนี้สมมติว่าคุณเข้าใจลักษณะทั่วไปของคิวแล้ว หากไม่ เข้าอ่านอ้างอิงก่อนต่อไป

1. คิว FIFO แบบพื้นฐาน

import queue
fifo = queue.Queue()
# put numbers into queue
for i in range(5):
    fifo.put(i)
# if not empty get the numbers from queue
print("Ouput")
while not fifo.empty():
    print(f" {fifo.get()} ")

ผลลัพธ์: 0 1 2 3 4 (ลำดับเดียวกับที่ใส่เข้าไป)

2. คิว LIFO แบบพื้นฐาน

import queue
lifo = queue.LifoQueue()
for i in range(5):
    lifo.put(i)
print("Ouput")
while not lifo.empty():
    print(f" {lifo.get()} ")

ผลลัพธ์: 4 3 2 1 0 (ตัวสุดท้ายที่ใส่จะออกก่อน)

3. Priority Queue

บางครั้งลำดับการประมวลผลในคิวต้องขึ้นอยู่กับ ความสำคัญ (priority) ของรายการ ไม่ใช่แค่ลำดับที่สร้างหรือเพิ่มเข้าไป ตัวอย่าง: งานธุรกิจสำคัญ (Business-Critical) ต้องการ CPU สูงสุด และมีความสำคัญมากกว่างานพิมพ์ (Print job) ของนักพัฒนา PriorityQueue ใช้ลำดับการจัดเรียง (sort order) ของเนื้อหาในคิว เพื่อตัดสินว่าจะดึงรายการไหนออกมา

import queue
import threading

# Class to get priority and description, validate priority
class Job:
    def __init__(self, priority, description):
        self.priority = priority
        self.description = description
        print('New job:', description)
        return
    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

# create a priority queue
q = queue.PriorityQueue()
q.put(Job(90, 'Developer-Print job'))
q.put(Job(2, 'Business-Report job'))
q.put(Job(1, 'Business-Critical Job'))

# process the job
def process_job(q):
    while True:
        next_job = q.get()
        print(f" *** Now, Processing the job - {next_job.description}")
        q.task_done()

# define threads
workers = [
    threading.Thread(target=process_job, args=(q,)),
    threading.Thread(target=process_job, args=(q,)),
]

# start and join threads
for w in workers:
    w.setDaemon(True)
    w.start()
q.join()

ผลลัพธ์

New job: Developer-Print job
New job: Business-Report job
New job: Business-Critical Job
*** Now, Processing the job - Business-Critical Job
*** Now, Processing the job - Business-Report job
*** Now, Processing the job - Developer-Print job

สรุป

ตัวอย่างนี้มีหลายเธรด consume jobs ซึ่งถูกประมวลผลตาม priority ของรายการในคิว ณ เวลาที่ get() ถูกเรียก ลำดับการประมวลผลขึ้นอยู่กับความสำคัญทางธุรกิจ ไม่ว่าจะเพิ่มเข้ามาก่อนหรือหลัง