วิธีสร้างคิวแบบมัลติเธรดด้วย Python

บทนำ

ในตัวอย่างนี้ เราจะสร้างคิวงาน (task queue) ที่เก็บงานทั้งหมดที่ต้องดำเนินการ และพูลเธรด (thread pool) ที่โต้ตอบกับคิวเพื่อประมวลผลองค์ประกอบแต่ละรายการอย่างเป็นอิสระ

คิว (Queue) คืออะไร?

คิวเป็นโครงสร้างข้อมูลที่เป็นชุดขององค์ประกอบต่างๆ ที่จัดเก็บในลำดับเฉพาะ มาลองเข้าใจผ่านตัวอย่างในชีวิตจริงกัน

สมมติว่าคุณยืนคิวจ่ายบิลที่เคาน์เตอร์ซูเปอร์มาร์เก็ต (อย่าถามว่าเป็นซูเปอร์มาร์เก็ตไหน)

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

1. คนเข้ามาที่ปลายคิวหนึ่งและออกจากอีกปลายหนึ่ง
2. ถ้าคน A เข้าคิวก่อนคน B คน A จะออกจากคิวก่อนคน B (เว้นแต่คน B จะเป็นดาราหรือมีสิทธิ์พิเศษ)
3. เมื่อทุกคนจ่ายบิลเสร็จแล้ว จะไม่มีใครเหลืออยู่ในคิว

กลับมาที่การเขียนโปรแกรม คิวทำงานในลักษณะคล้ายกัน:

1. enqueue - เพิ่มองค์ประกอบที่ปลายคิว
2. dequeue - นำองค์ประกอบออกจากหน้าคิว

ยังมีอีกสองแบบคือ First In First Out (FIFO) - องค์ประกอบที่เพิ่มเข้ามาก่อนจะถูกนำออกก่อน และ Last In First Out (LIFO) - องค์ประกอบที่เพิ่มเข้ามาสุดท้ายจะถูกนำออกก่อน

Python นำโครงสร้างข้อมูลคิวมาใช้อย่างไร?

โมดูล queue ใน Python ให้การนำคิวมาใช้งานอย่างง่าย แต่ละคิวมีเมธอดดังนี้:

• get(): คืนค่าองค์ประกอบถัดไป
• put(): เพิ่มองค์ประกอบใหม่
• qsize(): จำนวนองค์ประกอบปัจจุบันในคิว
• empty(): คืนค่า Boolean ว่าคิวว่างหรือไม่
• full(): คืนค่า Boolean ว่าคิวเต็มหรือไม่

ตัวอย่างการใช้งาน

1. ฟังก์ชันคำนวณตารางคูณ

เราจะสร้างฟังก์ชันที่รับอาร์กิวเมนต์ x แล้ววนลูปตัวเลข 1 ถึง x เพื่อคำนวณการคูณ ตัวอย่างเช่น ส่ง 5 เข้าไป จะวน 1 ถึง 5 คูณ 5 ได้ผลลัพธ์เป็นลิสต์

def print_multiply(x):
    output_value = []
    for i in range(1, x + 1):
        output_value.append(i * x)
    print(f"Output \n *** The multiplication result for the {x} is - {output_value}")

print_multiply(5)

ผลลัพธ์:

*** The multiplication result for the 5 is - [5, 10, 15, 20, 25]

2. ฟังก์ชันประมวลผลคิว

ฟังก์ชัน process_queue() จะพยายามดึงองค์ประกอบถัดไปจากอ็อบเจกต์คิว ตรรกะค่อนข้างง่าย คือ ส่งต่อองค์ประกอบจนกว่าคิวจะว่าง ใช้ sleep หน่วงเวลาเล็กน้อย

def process_queue():
    while True:
        try:
            value = my_queue.get(block=False)
        except queue.Empty:
            return
        else:
            print_multiply(value)
            time.sleep(2)

3. สร้างคลาสเธรด

สร้างคลาสที่เมื่อสร้างอินสแตนซ์และเริ่มต้น จะเรียก process_queue()

class MultiThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name
    def run(self):
        print(f" ** Starting the thread - {self.name}")
        process_queue()
        print(f" ** Completed the thread - {self.name}")

4. เตรียมข้อมูลเข้าและเติมคิว

# setting up variables
input_values = [5, 10, 15, 20]
# fill the queue
my_queue = queue.Queue()
for x in input_values:
    my_queue.put(x)

5. รวมโค้ดทั้งหมด

import queue
import threading
import time

# Class
class MultiThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name
    def run(self):
        print(f"Output \n ** Starting the thread - {self.name}")
        process_queue()
        print(f" ** Completed the thread - {self.name}")

# Process the queue
def process_queue():
    while True:
        try:
            value = my_queue.get(block=False)
        except queue.Empty:
            return
        else:
            print_multiply(value)
            time.sleep(2)

# function to multiply
def print_multiply(x):
    output_value = []
    for i in range(1, x + 1):
        output_value.append(i * x)
    print(f" \n *** The multiplication result for the {x} is - {output_value}")

# Input variables
input_values = [2, 4, 6, 5, 10, 3]
# fill the queue
my_queue = queue.Queue()
for x in input_values:
    my_queue.put(x)

# initializing and starting 4 threads
thread1 = MultiThread('First')
thread2 = MultiThread('Second')
thread3 = MultiThread('Third')
thread4 = MultiThread('Fourth')

# Start the threads
thread1.start()
thread2.start()
thread3.start()
thread4.start()

# Join the threads
thread1.join()
thread2.join()
thread3.join()
thread4.join()

ผลลัพธ์ตัวอย่าง

** Starting the thread - First
*** The multiplication result for the 2 is - [2, 4]
** Starting the thread - Second
*** The multiplication result for the 4 is - [4, 8, 12, 16]
** Starting the thread - Third
*** The multiplication result for the 6 is - [6, 12, 18, 24, 30, 36]
** Starting the thread - Fourth
*** The multiplication result for the 5 is - [5, 10, 15, 20, 25]
*** The multiplication result for the 10 is - [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
*** The multiplication result for the 3 is - [3, 6, 9]
** Completed the thread - Third
** Completed the thread - Fourth
** Completed the thread - Second
** Completed the thread - First

สรุป

เราได้สร้างคิวแบบมัลติเธรดสำเร็จแล้ว มี 4 เธรด แต่มีค่า 6 ค่าที่ต้องประมวลผล เธรดที่มาถึงคิวก่อนจะทำงานก่อน อื่นๆ จะรอคิว - คล้ายกับชีวิตจริงที่มี 3 เคาน์เตอร์แต่มีคนรอ 10 คน คนที่จ่ายเสร็จจะออกจากคิว ให้คนถัดไปเข้ามาแทน