Ở buổi này, chúng ta sẽ giải quyết hai vấn đề sống còn trong các hệ thống backend và dịch vụ mạng:
Tranh chấp tài nguyên (Race Condition): Khi nhiều coroutine cùng đọc/ghi một biến dữ liệu dẫn đến sai lệch số liệu. Giải pháp là asyncio.Lock.
Bộ đếm tải / Giới hạn tốc độ (Rate Limiting): Kiểm soát không cho mở quá nhiều kết nối cùng một lúc (giải quyết câu hỏi ở Buổi 2 khi có 1.000 URLs). Giải pháp là asyncio.Semaphore.
asyncio.LockNhiều người lầm tưởng: Asyncio chỉ chạy đơn luồng (Single Thread) nên không bao giờ xảy ra Race Condition. Điều này hoàn toàn sai!
Race Condition trong asyncio xuất hiện bất cứ khi nào:
Tác vụ A đọc một giá trị.
Tác vụ A gặp lệnh await và tạm nhường CPU cho Event Loop.
Tác vụ B nhảy vào sửa đổi giá trị đó.
Tác vụ A quay lại và ghi đè kết quả dựa trên dữ liệu cũ đã lỗi thời.
import asyncio
# Biến số dư dùng chung
balance = 100
lock = asyncio.Lock()
async def withdraw_unsafe(amount: int, client_id: int):
global balance
print(f"Khách {client_id}: Kiểm tra số dư (Hiện có {balance}$)...")
if balance >= amount:
# Giả lập thời gian truy vấn hệ thống xác thực mất 0.1s
# ĐIỂM CHẾT: Lệnh await khiến Event Loop chuyển sang khách khác!
await asyncio.sleep(0.1)
balance -= amount
print(f" -> Khách {client_id}: Rút thành công {amount}$! Số dư còn: {balance}$")
else:
print(f" -> Khách {client_id}: Giao dịch thất bại vì không đủ tiền!")
async def withdraw_safe(amount: int, client_id: int):
global balance
# Dùng Lock để tạo vùng độc quyền (Critical Section)
async with lock:
print(f"Khách {client_id}: Giữ Lock an toàn. Số dư hiện tại: {balance}$")
if balance >= amount:
await asyncio.sleep(0.1)
balance -= amount
print(f" -> Khách {client_id}: Rút thành công {amount}$! Số dư còn: {balance}$")
else:
print(f" -> Khách {client_id}: Giao dịch thất bại vì không đủ tiền!")Nếu chạy 2 khách hàng cùng rút 80$ với withdraw_unsafe: Cả hai đều kiểm tra thấy số dư còn 100$ $\implies$ cả 2 cùng rút thành công $\implies$ số dư bị âm (-60$)!
Với async with lock: Khách thứ 2 buộc phải đứng ngoài cửa chờ khách thứ 1 hoàn tất toàn bộ chu trình rồi mới được bước vào kiểm tra số dư.
Để hiểu chính xác cơ chế của Lock, chúng ta cần làm rõ một điểm mấu chốt: async with lock KHÔNG tự động "đóng băng" hay "khóa phép" biến balance ở cấp độ phần cứng hay RAM.
Biến balance vẫn chỉ là một biến số bình thường trong bộ nhớ. Python không ngăn cấm việc truy cập vào biến balance nếu một hàm nào đó cố tình đọc hay ghi nó một cách trực tiếp mà không thông qua lock.
Vậy tại sao biến balance lại được an toàn? Câu trả lời nằm ở nguyên lý: Kỷ luật đồng thuận (Cooperative Locking) và Chiếc chìa khóa phòng họp.
Hãy tưởng tượng biến balance giống như một quyển sổ cái nằm trong một căn phòng:
Căn phòng này có một ổ khóa duy nhất (chính là đối tượng lock = asyncio.Lock()).
Quy định chung cho mọi nhân viên (các Coroutine): "Ai muốn vào xem và sửa sổ cái, bắt buộc phải mượn chìa khóa, mở cửa vào phòng và chốt cửa lại. Khi nào xong việc mới được mở cửa và trả chìa khóa".
[ Ổ KHÓA DÙNG CHUNG: lock ]
|
+-------------------------------+-------------------------------+
| |
V V
[ Khách hàng 1 ] [ Khách hàng 2 ]
async with lock: async with lock:
1. Lấy chìa khóa (Thành công) 1. Đến mượn chìa khóa...
2. Vào phòng xem balance = 100 -> THẤY KHÓA ĐANG BẬT!
3. Gặp await asyncio.sleep(0.1) -> Buộc phải đứng chờ ở cửa
(Tạm dừng, nhưng VẪN GIỮ CHÌA) -> KHÔNG THỂ CHẠY TIẾP!
4. balance -= 80 (balance = 20) |
5. Bước ra, TRẢ CHÌA KHÓA |
| |
+-------------------> ĐƯỢC TRAO CHÌA KHÓA <-------------------------+
2. Bước vào phòng xem balance = 20
3. Không đủ tiền rút (20 < 80) -> Báo lỗi an toàn!
async with lock thực sự làm gì?Khi viết:
async with lock:
# Vùng độc quyền (Critical Section)
if balance >= amount:
await asyncio.sleep(0.1)
balance -= amountDưới tầng mã nguồn của Python, nó tương đương với:
# 1. Cố gắng lấy khóa (Acquire)
await lock.acquire()
try:
# 2. Vùng mã được bảo vệ an toàn
if balance >= amount:
await asyncio.sleep(0.1)
balance -= amount
finally:
# 3. Luôn luôn nhả khóa (Release) dù có lỗi xảy ra
lock.release()Cơ chế chặn diễn ra như sau:
Trạng thái nội bộ của lock:
Bên trong đối tượng lock chỉ có một biến cờ nhị phân _locked (True hoặc False) và một hàng đợi (Queue) chứa các Task đang chờ.
Khách 1 chạy await lock.acquire():
Thấy cờ đang False (khóa rảnh) $\rightarrow$ Khách 1 chuyển cờ thành True và bước vào thực thi khối code.
Điểm quan trọng nhất: Khi Khách 1 gặp await asyncio.sleep(0.1):
Khách 1 nhường CPU cho Event Loop để Event Loop chuyển sang chạy Khách 2.
Nhưng Khách 1 KHÔNG hề nhả khóa! Biến cờ _locked vẫn là True.
Khách 2 chạy await lock.acquire():
Khách 2 kiểm tra thấy cờ đang True (đã có người giữ khóa) $\rightarrow$ Lệnh await lock.acquire() lập tức đưa Khách 2 vào trạng thái ngủ đông, xếp hàng chờ.
Dòng code kiểm tra if balance >= amount của Khách 2 chưa hề được chạy đến!
Khách 1 hoàn thành khối lệnh:
Tự động chạy lock.release(), cờ chuyển về False.
Event Loop lập tức đánh thức Khách 2 dậy, trao khóa cho Khách 2 và lúc này Khách 2 mới bắt đầu đọc giá trị balance.
Vì Lock hoạt động dựa trên sự tự giác, nếu trong hệ thống có một hàm khác quên không dùng async with lock, nó vẫn có thể đọc/ghi đè biến balance và phá vỡ sự an toàn:
# Hàm này tuân thủ quy tắc -> Bị chặn chờ Khách 1
async def withdraw_safe(amount):
async with lock:
...
# Hàm này "phá rào" (Không dùng lock) -> VẪN ĐỌC ĐƯỢC BIẾN BALANCE!
async def check_balance_reckless():
# Dù Khách 1 đang giữ lock, hàm này vẫn nhảy vào in ra balance được!
print(balance)Quy tắc thiết kế: Để bảo vệ một biến dùng chung (như balance hay total_downloaded_bytes), mọi đoạn code có hành vi đọc hoặc ghi biến đó đều phải bọc trong cùng một đối tượng lock. Khi tất cả đều tôn trọng chiếc khóa, biến dữ liệu sẽ an toàn tuyệt đối.
asyncio.Semaphore – NGĂN CHẶN NGHẼN MẠNG & SẬP SERVERasyncio.Semaphore(value=N) quản lý một biến đếm số lượng vé nội bộ (ban đầu bằng $N$).
Khi một tác vụ chạy vào khối async with sem::
Nếu số vé $> 0$: Nó lấy 1 vé (số vé giảm đi 1) và chạy ngay.
Nếu số vé $= 0$: Nó buộc phải tạm dừng xếp hàng chờ ngoài cửa.
Khi tác vụ kết thúc khối lệnh: Nó trả lại 1 vé vào hộp (số vé tăng lên 1), cho phép tác vụ tiếp theo đang chờ bước vào.
💡 So sánh nhanh:
asyncio.Locktương đương vớiasyncio.Semaphore(1)(chỉ có duy nhất 1 vé).
asyncio.Semaphore(N)cho phép tối đa $N$ tác vụ cùng chạy song song.
Chúng ta sẽ mô phỏng việc cào/tải dữ liệu từ 15 đầu việc khác nhau, nhưng áp dụng Semaphore(3): Tại bất kỳ thời điểm nào, chỉ có tối đa 3 tác vụ được phép chạy đồng thời.
Tạo file buoi5_semaphore_rate_limiter.py:
import asyncio
import time
import random
class BoundedDownloader:
def __init__(self, max_concurrent: int):
# Bộ đếm giới hạn số lượng tác vụ đồng thời
self.semaphore = asyncio.Semaphore(max_concurrent)
self.lock = asyncio.Lock()
self.total_downloaded_bytes = 0
async def download_file(self, file_id: int):
"""Giả lập tải một file dung lượng ngẫu nhiên"""
print(f"⏳ [Xếp hàng] File {file_id:02d} đang chờ lượt...")
# Giới hạn số lượng tác vụ bước vào
async with self.semaphore:
duration = random.uniform(1.0, 2.5)
file_size_kb = random.randint(500, 1500)
print(f"🚀 [Đang tải] File {file_id:02d} bắt đầu tải ({file_size_kb} KB, dự kiến {duration:.1f}s)...")
await asyncio.sleep(duration) # Giả lập chờ I/O mạng
# Cập nhật tổng dung lượng an toàn với Lock
async with self.lock:
self.total_downloaded_bytes += file_size_kb
print(f"✅ [Xong] File {file_id:02d} đã tải xong! (Nhường vé cho file khác)")
async def run(self, total_files: int):
start_time = time.time()
print(f"=== BẮT ĐẦU TẢI {total_files} FILES (GIỚI HẠN TỐI ĐA 3 TÁC VỤ ĐỒNG THỜI) ===\n")
# Tạo 15 tasks cùng lúc
tasks = [self.download_file(i) for i in range(1, total_files + 1)]
await asyncio.gather(*tasks)
total_time = time.time() - start_time
print("\n" + "=" * 55)
print(f"⏱️ Tổng thời gian hoàn thành: {total_time:.2f} giây")
print(f"📦 Tổng dung lượng đã tải: {self.total_downloaded_bytes / 1024:.2f} MB")
print("=" * 55)
if __name__ == "__main__":
# Tối đa 3 tasks chạy cùng một lúc
downloader = BoundedDownloader(max_concurrent=3)
asyncio.run(downloader.run(total_files=10))Hiện tượng quan sát trên Terminal:
Ngay khi chương trình khởi động, chỉ có đúng 3 file đầu tiên được chuyển sang trạng thái 🚀 [Đang tải].
Các file từ 4 đến 10 đều ở trạng thái ⏳ [Xếp hàng].
Cứ mỗi khi có một file báo ✅ [Xong], lập tức sẽ có đúng một file tiếp theo được cấp vé để tải. Không bao giờ có hiện tượng 4 file cùng tải đồng thời!
Thử thách suy luận cho bé:
Thử thách 1: Trong đoạn code trên, nếu ta tăng max_concurrent=10 (bằng tổng số file) thì thời gian chạy sẽ giảm xuống còn bao nhiêu giây?
Thử thách 2 (Deadlock - Khóa chết):
Giả sử Coroutine 1 đang giữ Lock A và cố chờ Lock B.
Cùng lúc đó, Coroutine 2 đang giữ Lock B và cố chờ Lock A.
Điều gì sẽ xảy ra với hai coroutine này và hệ thống? Làm sao để phòng tránh hiện tượng này?