Ở buổi này, chúng ta sẽ học cách biến một Coroutine thụ động thành một Task chạy ngầm chủ động, gom nhóm chúng lại bằng asyncio.gather, và bắt lỗi an toàn để hệ thống không bị sập giữa chừng.
Coroutine (Bản thiết kế công việc):
Khi gọi download_async(1), bạn mới chỉ có một "tấm phiếu order".
Nếu chỉ viết await download_async(1) rồi await download_async(2), hai tác vụ vẫn chạy tuần tự nối đuôi nhau!
Task (Công việc đã được lên lịch chạy ngay):
Khi bọc coroutine vào asyncio.create_task(coroutine): Bạn vừa ném tấm phiếu đó vào danh sách chờ của Event Loop.
Ngay tại thời điểm gọi lệnh create_task, tác vụ đã bắt đầu được lên lịch chạy ngầm ngay lập tức trong nền, không cần phải chờ đến khi bạn gọi await.
[ Coroutine: download(1) ]
↓
asyncio.create_task(download(1))
↓
+-------------------------------------------------+
| Hàng đợi thực thi của Event Loop |
| -> [Task 1: Đang chạy ngầm] |
| -> [Task 2: Đang chạy ngầm] |
+-------------------------------------------------+
create_task)Hãy cho bé quan sát đoạn code nhỏ này để thấy rõ Task chạy ngầm như thế nào:
import asyncio
import time
async def worker(name, delay):
print(f"[{name}] Bắt đầu làm việc...")
await asyncio.sleep(delay)
print(f"[{name}] Đã xong sau {delay}s!")
return f"Kết quả từ {name}"
async def main():
start = time.time()
# 1. Tạo 2 Task chạy nền ngay lập tức
t1 = asyncio.create_task(worker("Task-A", 2))
t2 = asyncio.create_task(worker("Task-B", 3))
print(">> Đã khởi tạo 2 tasks xong, CPU làm việc khác trong lúc 2 task chạy ngầm...")
await asyncio.sleep(1) # CPU làm việc khác mất 1 giây
print(">> Bây giờ mới bắt đầu await lấy kết quả:")
# Lúc này t1 chỉ còn phải chờ thêm 1s (vì đã chạy được 1s trước đó)
res1 = await t1
# t2 chỉ còn phải chờ thêm 1s nữa
res2 = await t2
print(f"Kết quả nhận được: {res1} | {res2}")
print(f"⏱️ Tổng thời gian: {time.time() - start:.2f}s") # Chỉ mất ~3s, không phải 1 + 2 + 3 = 6s!
asyncio.run(main())asyncio.gatherKhi có hàng chục hoặc hàng trăm tasks, việc ngồi gõ từng dòng await t1, await t2 rất cồng kềnh. asyncio.gather sinh ra để giải quyết việc này:
Cú pháp: results = await asyncio.gather(*tasks)
Đặc điểm:
Tự động bọc các coroutine thành task (nếu chưa bọc).
Chờ cho đến khi tất cả các task hoàn thành.
Trả về một danh sách kết quả (List) theo đúng thứ tự truyền vào, bất kể task nào hoàn thành trước hay sau.
return_exceptions=True:Mặc định trong asyncio.gather, nếu có 1 task bị lỗi ném ra ngoại lệ (Exception), toàn bộ gather sẽ bị ngắt và ném lỗi đó ra ngay lập tức, các kết quả của task khác có thể bị thất lạc.
Khi thêm return_exceptions=True: Nếu task nào gặp lỗi, phần tử tương ứng trong danh sách kết quả sẽ chứa chính đối tượng Exception đó thay vì làm sập chương trình. Các task còn lại vẫn chạy bình thường!
Hai bố con sẽ viết một chương trình kiểm tra trạng thái sức khỏe (Status & Ping) của nhiều website đồng thời. Để gọi HTTP bất đồng bộ, chúng ta cài đặt thư viện chuyên dụng httpx (tương tự requests nhưng chạy async).
Mở Terminal và cài đặt:
pip install httpx
Tạo file buoi2_health_checker.py:
import asyncio
import time
import httpx
class WebHealthChecker:
def __init__(self, urls):
self.urls = urls
async def check_url(self, client: httpx.AsyncClient, url: str) -> dict:
"""Kiểm tra một URL đơn lẻ và đo thời gian phản hồi (latency)"""
start_time = time.time()
try:
# Gửi request GET bất đồng bộ với timeout 5 giây
response = await client.get(url, timeout=5.0)
latency = (time.time() - start_time) * 1000 # đổi ra mili-giây (ms)
return {
"url": url,
"status": response.status_code,
"latency_ms": round(latency, 2),
"error": None
}
except Exception as e:
latency = (time.time() - start_time) * 1000
return {
"url": url,
"status": None,
"latency_ms": round(latency, 2),
"error": type(e).__name__ # Tên lỗi (ConnectTimeout, ConnectError,...)
}
async def run_all(self):
print(f"🚀 Bắt đầu kiểm tra {len(self.urls)} website đồng thời...")
start_total = time.time()
# Dùng chung 1 AsyncClient để tối ưu tái sử dụng kết nối mạng (Connection Pooling)
async with httpx.AsyncClient() as client:
# Tạo danh sách các task
tasks = [self.check_url(client, url) for url in self.urls]
# Gom nhóm và thực thi song song toàn bộ
# return_exceptions=True đảm bảo không bao giờ bị văng lỗi dừng hệ thống
results = await asyncio.gather(*tasks, return_exceptions=True)
total_time = time.time() - start_total
# In bảng kết quả trực quan
print("\n" + "=" * 65)
print(f"{'URL':<30} | {'STATUS':<8} | {'LATENCY':<10} | {'GHI CHÚ'}")
print("=" * 65)
for res in results:
if isinstance(res, Exception):
print(f"❌ Task bị lỗi crash ngoài ý muốn: {res}")
continue
status = str(res["status"]) if res["status"] else "FAILED"
latency = f"{res['latency_ms']} ms"
note = res["error"] if res["error"] else "OK"
print(f"{res['url']:<30} | {status:<8} | {latency:<10} | {note}")
print("=" * 65)
print(f"⏱️ Tổng thời gian kiểm tra toàn bộ: {total_time:.2f} giây\n")
if __name__ == "__main__":
# Danh sách thử nghiệm gồm cả trang sống, trang chết, trang giả lập lỗi
test_urls = [
"https://google.com",
"https://github.com",
"https://youtube.com",
"https://python.org",
"https://trang-web-khong-ton-tai-123456.com", # Trang gây lỗi ConnectError
"https://httpbin.org/delay/2", # Trang giả lập phản hồi chậm 2s
"https://cloudflare.com",
"https://wikipedia.org"
]
checker = WebHealthChecker(test_urls)
asyncio.run(checker.run_all())Hiện tượng quan sát được:
Trong danh sách có trang httpbin.org/delay/2 bị cố tình làm trễ mất 2 giây, nhưng toàn bộ chương trình chỉ mất khoảng hơn 2 giây một chút để hoàn thành cả 8 trang!
Trang không tồn tại sẽ báo lỗi ConnectError rất gọn gàng mà không kéo sập các trang khác.
Thử thách suy luận:
Thử thách 1 (asyncio.as_completed):
asyncio.gather bắt chúng ta phải chờ tất cả các trang xong mới in ra một lần.
Làm thế nào để trang nào phản hồi xong trước thì in ngay ra màn hình lập tức, không cần đợi trang chậm 2 giây? (Gợi ý: Tìm hiểu vòng lặp for future in asyncio.as_completed(tasks):).
Thử thách 2 (Kiểm soát tiến độ):
Giả sử có 1.000 URLs, nếu ném hết 1.000 URLs vào gather cùng lúc thì điều gì sẽ xảy ra với cổng mạng (Socket/File descriptor) của máy tính hoặc trang web đích? (Đây chính là lý do chúng ta sẽ cần học đến Semaphore ở Buổi 5!).
-----------------------------------------------------------------------------------------------------------
câu hỏi : giải thích thêm giúp tôi một chút về đối tượng được tạo ra bởi httpx.AsyncClient(), có phải đây là cách tạo ra 1 kết nối ( socket ) tới url cho trước phải không ?
Câu trả lời chính xác là: Chưa phải.
Khi anh gọi client = httpx.AsyncClient(), chương trình hoàn toàn chưa tạo ra một kết nối Socket nào tới bất kỳ URL cụ thể nào cả.
Thay vào đó, httpx.AsyncClient() tạo ra một Trình quản lý tập trung (Client Session) với hạt nhân bên trong là một Hồ chứa kết nối (Connection Pool).
Hãy hình dung sự khác nhau qua hai thời điểm:
Thời điểm 1: Khi gọi client = httpx.AsyncClient()
httpx chỉ tạo ra một bộ khung điều phối trong bộ nhớ RAM (chứa các cấu hình mặc định như: thời gian chờ timeout, danh sách headers, cookie jar, giới hạn số lượng kết nối tối đa...).
Lúc này, httpx chưa biết anh sẽ gọi tới trang web nào (google.com, github.com hay IP nội bộ), do đó chưa có bất kỳ kết nối socket TCP nào được mở.
Thời điểm 2: Khi thực thi lệnh await client.get("[https://google.com](https://google.com)")
Lúc này httpx mới thực sự hành động:
Tra cứu DNS để đổi tên miền google.com thành IP.
Tạo một kết nối Socket TCP tới IP đó (bắt tay 3 bước SYN, SYN-ACK, ACK).
Thực hiện bắt tay TLS/SSL nếu là https.
Gửi HTTP Request và nhận Response về.
AsyncClient: Cơ chế Connection Pool (Hồ chứa Socket)Tại sao chúng ta không dùng cú pháp gọi thẳng như httpx.get(url) (tương tự requests.get) mà phải khởi tạo AsyncClient() rồi dùng chung?
Bí mật nằm ở chỗ Connection Pool (Tái sử dụng Socket):
[ Không dùng AsyncClient - Mỗi request mở 1 Socket riêng ]:
Request 1 -> Mở Socket -> Bắt tay TCP -> Gửi/Nhận -> Đóng Socket (Hủy)
Request 2 -> Mở Socket -> Bắt tay TCP -> Gửi/Nhận -> Đóng Socket (Hủy)
(Rất tốn CPU và độ trễ mạng mạng do liên tục mở/đóng)
[ Dùng chung 1 AsyncClient - Có Connection Pool ]:
+----------------------------------+
| httpx.AsyncClient Pool |
| [Socket Google] [Socket GitHub] |
+----------------------------------+
Request 1 -> Mở Socket Google -> Gửi/Nhận -> GIỮ NGUYÊN SOCKET TRONG POOL (Keep-Alive)
Request 2 -> Cần gọi Google tiếp? -> LẤY LẠI SOCKET CŨ DÙNG NGAY (Bỏ qua bước bắt tay!)
Cơ chế Keep-Alive: Sau khi nhận dữ liệu xong, AsyncClient không vội vàng ngắt socket. Nó giữ socket đó mở trong hồ chứa (Pool).
Nếu trong danh sách URLs của anh có 5 request gửi tới cùng một tên miền (ví dụ 5 API của cùng một máy chủ backend):
Request đầu tiên sẽ mất thời gian tạo socket.
4 request tiếp theo sẽ tái sử dụng ngay lập tức socket đang mở đó, giúp tốc độ phản hồi nhanh hơn gấp 2 đến 3 lần vì không tốn độ trễ (latency) cho việc kết nối lại từ đầu.
httpx.AsyncClient() không phải là 1 socket gắn chết với 1 URL.
Nó là một người quản lý đa năng sở hữu một hồ chứa nhiều socket bên trong.
Tùy thuộc vào việc anh get() hay post() tới những URL nào, nó sẽ tự động mở các socket tương ứng tới các server đó, điều phối dữ liệu bất đồng bộ và giữ lại các socket đó để tái sử dụng một cách thông minh nhất.
------------------------------------------------------------------------------------------------------------
HƯỚNG DẪN 2 THỬ THÁCH Ở PHẦN TRÊN
asyncio.as_completedgather vs as_completedasyncio.gather(*tasks):
Giống như việc cả lớp cùng đi ăn: Phải đợi tất cả mọi người ăn xong thì cả nhóm mới đứng dậy ra về một lượt.
Giữ nguyên thứ tự kết quả tương ứng theo danh sách task truyền vào, bất kể task nào chạy xong trước hay sau.
asyncio.as_completed(tasks):
Giống như quầy phục vụ thức ăn nhanh: Bất kỳ ai nhận được phần ăn trước thì người đó được phục vụ ra bàn ăn ngay, ai gọi món lâu thì nhận sau.
Trả về một trình lặp (Iterator) sinh ra các Future. Khi duyệt qua vòng lặp bằng for future in asyncio.as_completed(tasks):, mỗi lần await future, bạn sẽ nhận được kết quả của tác vụ vừa hoàn thành sớm nhất.
WebHealthCheckerCập nhật lại phương thức run_all() trong file buoi2_health_checker.py:
import asyncio
import time
import httpx
class FastWebHealthChecker:
def __init__(self, urls):
self.urls = urls
async def check_url(self, client: httpx.AsyncClient, url: str) -> dict:
start_time = time.time()
try:
response = await client.get(url, timeout=5.0)
latency = (time.time() - start_time) * 1000
return {
"url": url,
"status": response.status_code,
"latency_ms": round(latency, 2),
"error": None
}
except Exception as e:
latency = (time.time() - start_time) * 1000
return {
"url": url,
"status": None,
"latency_ms": round(latency, 2),
"error": type(e).__name__
}
async def run_streaming(self):
print(f"🚀 Bắt đầu kiểm tra {len(self.urls)} URLs với asyncio.as_completed()...")
print("-> Trang nào phản hồi xong trước sẽ hiển thị ngay lập tức!\n")
print(f"{'URL':<35} | {'STATUS':<8} | {'LATENCY':<10} | {'GHI CHÚ'}")
print("-" * 70)
start_total = time.time()
async with httpx.AsyncClient() as client:
# 1. Tạo danh sách các coroutine
tasks = [self.check_url(client, url) for url in self.urls]
# 2. Duyệt qua as_completed: Tác vụ nào xong trước sẽ nhả ra trước
for future in asyncio.as_completed(tasks):
# await future ở đây chỉ đợi đúng tác vụ vừa xong tiếp theo
res = await future
status = str(res["status"]) if res["status"] else "FAILED"
latency = f"{res['latency_ms']} ms"
note = res["error"] if res["error"] else "OK"
# In ra màn hình ngay lập tức tại thời điểm nhận được phản hồi
print(f"{res['url']:<35} | {status:<8} | {latency:<10} | {note}")
print("-" * 70)
print(f"⏱️ Toàn bộ quá trình hoàn tất trong: {time.time() - start_total:.2f} giây")
if __name__ == "__main__":
test_urls = [
"https://httpbin.org/delay/2", # Cố tình chậm mất 2 giây
"https://google.com", # Cực nhanh (~50-100ms)
"https://github.com", # Nhanh
"https://python.org", # Nhanh
"https://cloudflare.com" # Nhanh
]
checker = FastWebHealthChecker(test_urls)
asyncio.run(checker.run_streaming())Quan sát khi chạy: Các trang google.com, github.com, python.org sẽ hiện lên màn hình gần như tức thì (trong vài trăm mili-giây đầu tiên). Riêng dòng httpbin.org/delay/2 sẽ xuất hiện sau cùng sau 2 giây. Hai bố con không cần phải nhìn màn hình trống trơn đứng chờ 2 giây nữa!
gather CÙNG LÚC?Khi gọi asyncio.gather(*tasks) với 1.000 URLs, Event Loop sẽ cố gắng mở đồng thời 1.000 kết nối mạng trong cùng một tích tắc. Điều này dẫn tới 3 sự cố nghiêm trọng ở cấp độ hệ thống và mạng:
Hết File Descriptor (Lỗi OSError: [Errno 24] Too many open files):
Trong hệ điều hành (đặc biệt là Linux/macOS), mỗi Socket kết nối mạng được hệ điều hành quản lý như một File Descriptor (FD).
Hệ điều hành luôn đặt giới hạn an toàn (thường mặc định là 1024 FDs cho mỗi tiến trình). Nếu mở đồng loạt 1.000+ socket, chương trình Python sẽ bị hệ điều hành chặn lại và văng lỗi sập chương trình ngay lập tức.
Nghẽn cổng (Port Exhaustion) & Tràn RAM:
Mỗi kết nối TCP cần một cổng ngẫu nhiên (Ephemeral Port) và bộ nhớ đệm (Buffer). Mở quá nhiều cùng lúc làm cạn kiệt tài nguyên mạng của máy.
Bị coi là tấn công từ chối dịch vụ (DDoS):
Nếu 1.000 URLs đó thuộc về cùng một trang web (ví dụ cùng cào dữ liệu từ 1 trang tin hay 1 sàn thương mại điện tử), việc bắn 1.000 requests cùng 1 giây sẽ kích hoạt tường lửa (Firewall/Cloudflare).
Máy chủ đích sẽ lập tức chặn (Block) địa chỉ IP của bạn hoặc trả về mã lỗi 429 Too Many Requests.
asyncio.SemaphoreĐể giải quyết bài toán này, người ta dùng Semaphore (Cột đèn điều tiết giao thông).
Nếu có 1.000 URLs, Semaphore sẽ khống chế: "Tại một thời điểm, chỉ cho phép tối đa 10 kết nối được phép chạy. Khi 1 kết nối xong xuôi, mới cho URL thứ 11 bước vào".
Nhờ đó hệ thống vừa chạy cực nhanh, vừa ổn định mà không bao giờ bị tràn Socket hay bị chặn IP.