Cách triển khai hàng đợi đa luồng (Multithreaded Queue) với Python

Giới thiệu

Trong ví dụ này, chúng ta sẽ tạo một hàng đợi tác vụ (task queue) chứa tất cả các tác vụ cần thực thi và một nhóm luồng (thread pool) tương tác với hàng đợi để xử lý từng phần tử riêng lẻ.

Trước tiên, hãy hiểu khái niệm: Queue (Hàng đợi) là gì? Hàng đợi là một cấu trúc dữ liệu là tập hợp các phần tử được duy trì theo một thứ tự cụ thể. Hãy hình dung qua ví dụ thực tế.

Giả sử bạn đứng xếp hàng để thanh toán tại quầy thu ngân của một cửa hàng tạp hóa.

Trong hàng người chờ thanh toán, bạn sẽ nhận thấy:

Mọi người vào hàng ở một đầu và rời khỏi hàng ở đầu kia.
Nếu người A vào hàng trước người B, thì người A sẽ rời hàng trước người B (trừ khi người B là người nổi tiếng hoặc có quyền ưu tiên).
Khi mọi người đã thanh toán xong, hàng sẽ không còn ai.

Quay lại lập trình, hàng đợi hoạt động theo cách tương tự:

enqueue - Thêm phần tử vào cuối hàng đợi.
dequeue - Loại bỏ phần tử từ đầu hàng đợi.

Có hai chế độ phổ biến:

FIFO (First In First Out) - Phần tử thêm vào đầu tiên sẽ được lấy ra đầu tiên.
LIFO (Last In First Out) - Phần tử thêm vào cuối cùng sẽ được lấy ra đầu tiên.

Python triển khai cấu trúc dữ liệu Queue như thế nào?

Module queue trong Python cung cấp triển khai đơn giản của cấu trúc dữ liệu hàng đợi. Mỗi hàng đợi có các phương thức sau:

get(): Trả về phần tử tiếp theo.
put(): Thêm một phần tử mới.
qsize(): Số lượng phần tử hiện tại trong hàng đợi.
empty(): Trả về giá trị Boolean, cho biết hàng đợi có rỗng không.
full(): Trả về giá trị Boolean, cho biết hàng đợi đã đầy chưa.

Ví dụ minh họa

1. Hàm tính bảng nhân

Tạo một hàm nhận tham số x, lặp qua các số từ 1 đến x và thực hiện phép nhân, cuối cùng trả về danh sách kết quả.

def print_multiply(x):
    output_value = []
    for i in range(1, x + 1):
        output_value.append(i * x)
    print(f"Output \n *** Kết quả nhân cho {x} là - {output_value}")

print_multiply(5)

Kết quả:

*** Kết quả nhân cho 5 là - [5, 10, 15, 20, 25]

2. Hàm xử lý hàng đợi

Hàm process_queue() sẽ cố gắng lấy phần tử tiếp theo từ đối tượng hàng đợi. Logic rất đơn giản: tiếp tục lấy phần tử cho đến khi hàng đợi rỗng. Sử dụng sleep để tạo độ trễ nhỏ.

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

3. Lớp luồng đa nhiệm (MultiThread)

Tạo một lớp, khi khởi tạo và bắt đầu instance mới, hàm process_queue() sẽ được gọi.

class MultiThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name

    def run(self):
        print(f" ** Bắt đầu luồng - {self.name}")
        process_queue()
        print(f" ** Hoàn thành luồng - {self.name}")

4. Điền dữ liệu vào hàng đợi


# Thiết lập biến đầu vào
input_values = [5, 10, 15, 20]

# Tạo hàng đợi và điền dữ liệu
my_queue = queue.Queue()
for x in input_values:
    my_queue.put(x)

5. Chương trình hoàn chỉnh


import queue
import threading
import time

# Lớp luồng
class MultiThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name

    def run(self):
        print(f"Output \n ** Bắt đầu luồng - {self.name}")
        process_queue()
        print(f" ** Hoàn thành luồng - {self.name}")

# Xử lý hàng đợi
def process_queue():
    while True:
        try:
            value = my_queue.get(block=False)
        except queue.Empty:
            return
        else:
            print_multiply(value)
            time.sleep(2)

# Hàm nhân
def print_multiply(x):
    output_value = []
    for i in range(1, x + 1):
        output_value.append(i * x)
    print(f" \n *** Kết quả nhân cho {x} là - {output_value}")

# Dữ liệu đầu vào
input_values = [2, 4, 6, 5, 10, 3]

# Khởi tạo hàng đợi
my_queue = queue.Queue()
for x in input_values:
    my_queue.put(x)

# Khởi tạo và bắt đầu 4 luồng
thread1 = MultiThread('First')
thread2 = MultiThread('Second')
thread3 = MultiThread('Third')
thread4 = MultiThread('Fourth')

# Bắt đầu các luồng
thread1.start()
thread2.start()
thread3.start()
thread4.start()

# Chờ các luồng hoàn thành
thread1.join()
thread2.join()
thread3.join()
thread4.join()

Kết quả mẫu (thứ tự có thể khác do đa luồng):


** Bắt đầu luồng - First
*** Kết quả nhân cho 2 là - [2, 4]
** Bắt đầu luồng - Second
*** Kết quả nhân cho 4 là - [4, 8, 12, 16]
** Bắt đầu luồng - Third
*** Kết quả nhân cho 6 là - [6, 12, 18, 24, 30, 36]
** Bắt đầu luồng - Fourth
*** Kết quả nhân cho 5 là - [5, 10, 15, 20, 25]
*** Kết quả nhân cho 10 là - [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
*** Kết quả nhân cho 3 là - [3, 6, 9]
** Hoàn thành luồng - Third
** Hoàn thành luồng - Fourth
** Hoàn thành luồng - Second
** Hoàn thành luồng - First

6. Giải thích

Chúng ta đã triển khai thành công khái niệm hàng đợi. Có 4 luồng nhưng 6 giá trị cần xử lý, nên luồng nào đến hàng đợi trước sẽ được thực thi trước, các luồng khác sẽ chờ đợi.

Điều này tương tự thực tế: giả sử có 3 quầy thu ngân nhưng 10 người chờ thanh toán, 10 người sẽ xếp vào 3 hàng, ai thanh toán xong sẽ rời hàng và nhường chỗ cho người tiếp theo.