Giáo trình Python miễn phí · Chương 12/12: Đa luồng và đồng bộ · Bài 24/24.

Mục tiêu bài học

Đồng bộ dữ liệu chia sẻ và phân phối công việc cho số worker có giới hạn, tránh xác nhận hoàn thành quá sớm.

Race condition và Lock

Race condition xảy ra khi kết quả phụ thuộc thứ tự xen kẽ của các thao tác chưa được đồng bộ đúng. Một câu lệnh nhìn ngắn như tăng biến có thể gồm nhiều bước. Không dựa vào GIL như cơ chế bảo vệ quy tắc nghiệp vụ.

Lock cho phép một thread vào vùng găng tại một thời điểm. Dùng with lock để nhả khóa khi thoát khối, kể cả có ngoại lệ. Giữ vùng khóa ngắn, không chờ mạng hoặc lấy nhiều khóa theo thứ tự thay đổi vì dễ gây nghẽn, deadlock.

from threading import Lock, Thread

tong = 0
khoa = Lock()

def tang():
    global tong
    for _ in range(1000):
        with khoa:
            tong += 1

threads = [Thread(target=tang) for _ in range(3)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
print(tong)

Kết quả:

3000

Queue và xác nhận công việc

queue.Queue trao đổi dữ liệu giữa các thread. get() lấy một việc; task_done() xác nhận việc tương ứng đã xử lý. Queue.join() chờ các việc đã put() được xác nhận, khác Thread.join() chờ thread kết thúc. Mỗi get thành công cần một task_done; đặt trong finally để xử lý lỗi không làm bộ đếm treo.

from queue import Queue
from threading import Thread

cong_viec = Queue()
ket_qua = Queue()

def worker():
    while True:
        item = cong_viec.get()
        try:
            if item is None:
                return
            try:
                so = int(item)
                ket_qua.put((str(item), so * so))
            except ValueError:
                ket_qua.put((str(item), "Không phải số nguyên"))
        finally:
            cong_viec.task_done()

threads = [Thread(target=worker) for _ in range(2)]
for thread in threads:
    thread.start()
for item in [2, 3, "abc"]:
    cong_viec.put(item)
for _ in threads:
    cong_viec.put(None)
cong_viec.join()
for thread in threads:
    thread.join()
for item, result in sorted(ket_qua.get() for _ in range(3)):
    print(item, result)

Kết quả:

2 4
3 9
abc Không phải số nguyên

Thiết kế số worker và điểm dừng

Ví dụ dùng hai worker và một tín hiệu None cho mỗi worker để dừng. Với nhiều việc, Queue(maxsize=…) giới hạn số việc chờ và tạo áp lực ngược thay vì đẩy hết vào bộ nhớ. Tác vụ mạng cần timeout riêng vì Queue.join() không tự giới hạn một công việc đang treo.

Nếu chỉ cần chạy nhóm hàm và lấy kết quả, ThreadPoolExecutor có thể giảm quản lý thread thủ công. Hiểu Lock, Queue và điểm dừng vẫn quan trọng để tránh chia sẻ trạng thái sai. Người học có thể tiếp tục chọn chuyên đề web, tự động hóa hoặc dữ liệu theo mục tiêu.

Lỗi thường gặp và lưu ý

• Thiếu task_done() làm Queue.join() có thể chờ mãi; gọi thừa gây ValueError.
• Không dùng Queue.empty() như bằng chứng chắc chắn không còn việc khi các thread khác vẫn thêm hoặc lấy dữ liệu.

Ghi nhớ

Bộ 24 bài kết thúc ở cách tổ chức chương trình có dữ liệu, ngoại lệ, tài nguyên và công việc đồng thời. Dùng mục lục để quay lại nền tảng khi đọc chương trình thực tế.

Tài liệu đối chiếu

• Queue
• Threading