Ở 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.

PHẦN 1: BẢN CHẤT CỦA COROUTINE VS TASK

1. 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!

2. 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]                    |
    +-------------------------------------------------+

PHẦN 2: THỰC NGHIỆM ĐỐI CHỨNG (HIỂU SÂU VỀ 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())

PHẦN 3: GOM NHÓM ĐA TÁC VỤ VỚI asyncio.gather

Khi 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:

1. Cú pháp: results = await asyncio.gather(*tasks)

2. Đặ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.

Kỹ thuật bắt lỗi với 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!

Câu hỏi:  

Giả sử  có nhiều ( ví dụ 3 ) task chạy song song, làm cách nào để lấy kết quả của task chạy nhanh nhất để print ra sớm nhất, rồi lần lượt tới kết quả của task chạy nhanh thứ 2 và thứ 3. Khi bắt đầu chạy chúng ta không biết task nào nhanh nhất.

Để giải quyết bài toán này, Python asyncio cung cấp sẵn hàm chuyên dụng là asyncio.as_completed().

Hàm này nhận vào một danh sách các coroutine/task, cho chúng chạy đồng thời, và trả về một bộ lặp (iterator) mà mỗi khi bạn await, nó sẽ trả về kết quả của task nào vừa về đích sớm nhất, bất kể thứ tự ban đầu bạn đưa vào là gì.

import asyncio
import random
import time

async def worker(task_id: int, delay: float):
    """Mô phỏng tác vụ mạng/tính toán với độ trễ ngẫu nhiên."""
    print(f"⏳ Task {task_id} bắt đầu chạy (dự kiến mất {delay:.2f}s)...")
    await asyncio.sleep(delay)
    return f"Kết quả từ Task {task_id} (hoàn thành sau {delay:.2f}s)"

async def main():
    # Giả lập 3 task với thời gian hoàn thành ngẫu nhiên từ 1 đến 4 giây
    # Lúc này ta hoàn toàn không biết task nào sẽ nhanh nhất
    tasks = [
        worker(1, delay=random.uniform(1.0, 4.0)),
        worker(2, delay=random.uniform(1.0, 4.0)),
        worker(3, delay=random.uniform(1.0, 4.0)),
    ]

    print("🚀 Bắt đầu chạy cả 3 task đồng thời...\n")
    start_time = time.perf_counter()

    # asyncio.as_completed() tự động sắp xếp theo thứ tự về đích
    for rank, coro_done in enumerate(asyncio.as_completed(tasks), start=1):
        # Lệnh await này sẽ lập tức mở khóa ngay khi có 1 task vừa chạy xong
        result = await coro_done
        elapsed = time.perf_counter() - start_time
        print(f"🏆 [Vị trí {rank} - {elapsed:.2f}s]: {result}")

if __name__ == "__main__":
    asyncio.run(main())

2. Kết quả chạy thực tế (Ví dụ minh họa)

Giả sử trong một lượt chạy, Task 2 ngẫu nhiên mất 1.1s, Task 1 mất 2.4s, Task 3 mất 3.8s:

🚀 Bắt đầu chạy cả 3 task đồng thời...

⏳ Task 1 bắt đầu chạy (dự kiến mất 2.41s)...
⏳ Task 2 bắt đầu chạy (dự kiến mất 1.15s)...
⏳ Task 3 bắt đầu chạy (dự kiến mất 3.82s)...

🏆 [Vị trí 1 - 1.15s]: Kết quả từ Task 2 (hoàn thành sau 1.15s)
🏆 [Vị trí 2 - 2.41s]: Kết quả từ Task 1 (hoàn thành sau 2.41s)
🏆 [Vị trí 3 - 3.82s]: Kết quả từ Task 3 (hoàn thành sau 3.82s)

Tổng thời gian toàn bộ chương trình chỉ bằng thời gian của task lâu nhất (~3.8s), chứ không phải cộng dồn thời gian của cả 3 task lại.

3. Phân biệt với asyncio.gather() và asyncio.wait()

• asyncio.gather(*tasks):

  • Thu thập kết quả theo đúng thứ tự danh sách truyền vào ban đầu (luôn trả về Task 1, rồi Task 2, Task 3).

  • Bạn bắt buộc phải đợi tất cả các task hoàn tất thì mới lấy được danh sách kết quả, không thể lấy ngay kết quả của task về sớm nhất để in ra trước.

• asyncio.as_completed(tasks):

  • Bỏ qua thứ tự khai báo ban đầu.

  • Ưu tiên tính thời gian thực: Ai xong trước trả về trước. Thích hợp cho các bài toán: gửi request tới nhiều server dự phòng (ai trả lời trước thì dùng trước), quét nhiều cổng mạng, hoặc xử lý các frame video/ảnh khi hoàn thành phân tích.

• asyncio.wait(tasks, return_when=FIRST_COMPLETED):

  • Dùng khi bạn chỉ muốn lấy duy nhất task chạy nhanh nhất rồi hủy bỏ tất cả các task còn lại (ví dụ: query song song 3 máy chủ DNS, máy chủ nào trả về trước thì ngắt kết nối 2 máy chủ kia để tiết kiệm tài nguyên).

Câu hỏi:

Trong ví dụ minh họa trên ta thấy async def main() là một Coroutine, trong coroutine này ta gọi các coroutine ascyn def worker và add chúng vào Event Loop. Tôi hiểu rằng bản thân couritine main cũng là được add vào chính Event Loop mà nó tạo ra đúng không, khi đó main và các worker cùng là các task ngang hàng trong Event loop này phải không ?

Cách hiểu trên gần như chính xác hoàn toàn về mặt cơ chế vận hành, chỉ có một điểm đính chính nhỏ về mặt khái niệm: Event Loop không phải do main() tạo ra, mà là do asyncio.run() tạo ra trước.

1. Ai tạo ra Event Loop và main() trở thành Task như thế nào?

Khi bạn chạy dòng lệnh ngoài cùng:

asyncio.run(main())

Trình tự chính xác bên dưới Python diễn ra như sau:

1. asyncio.run() tạo Event Loop: asyncio.run() đứng ra khởi tạo một Event Loop mới toanh cho luồng (thread) hiện tại.

2. Bọc main() thành Task gốc (Root Task):

   • asyncio.run() nhận Coroutine main(), sau đó gọi ngầm lệnh loop.create_task(main()).

   • Lúc này, main() chính thức trở thành một Task đầu tiên nằm trong Event Loop.

3. Chạy Event Loop cho tới khi main() xong:

   • asyncio.run() kích hoạt Event Loop bằng phương thức loop.run_until_complete(main_task).

   • Vòng đời của toàn bộ chương trình phụ thuộc vào Task main: Khi main kết thúc (trả về kết quả hoặc lỗi), asyncio.run() sẽ hủy các task ngầm còn sót lại, đóng Event Loop và giải phóng tài nguyên.

2. Mối quan hệ giữa Task main và các Task worker: Có thực sự "ngang hàng"?

Câu trả lời là VỪA CÓ, VỪA KHÔNG, tùy thuộc vào góc nhìn:

               ┌────────────────────────────────────────────────────────┐
               │                       EVENT LOOP                       │
               │                                                        │
               │   HÀNG ĐỢI ĐIỀU PHỐI (Scheduler Ready Queue)           │
               │   ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌─────┐ │
               │   │Task: main│◄─►│ Task: w1 │◄─►│ Task: w2 │◄─►│ ... │ │
               │   └────┬─────┘   └────┬─────┘   └────┬─────┘   └─────┘ │
               └────────┼──────────────┼──────────────┼─────────────────┘
                        │              │              │
                        │ (await)      │              │
                        ▼              ▼              ▼
           Tạm dừng nhường CPU ──► Cấp CPU chạy ──► Cấp CPU chạy

Dưới góc nhìn của Event Loop (Bộ điều phối Scheduler) $\rightarrow$ HOÀN TOÀN NGANG HÀNG

• Khi asyncio.as_completed(tasks) được gọi, nó tự động bọc 3 coroutine worker thành 3 đối tượng Task độc lập và nạp vào Event Loop.

• Với Event Loop, nó chỉ nhìn thấy một hàng đợi chứa 4 thực thể Task: [Task_main, Task_worker1, Task_worker2, Task_worker3].

• Không có sự phân cấp ưu tiên (No Priority): Cả 4 task đều được đối xử bình đẳng. Khi Task_main gặp lệnh await coro_done, nó tự nguyện tạm dừng và nhường quyền. Event Loop lập tức cấp chu kỳ CPU cho Task_worker1, Task_worker2... chạy tiếp.

Dưới góc nhìn Phả hệ / Vòng đời (Parent - Child Lifecycle) $\rightarrow$ CÓ QUAN HỆ PHỤ THUỘC

• main đóng vai trò là Task cha (Root / Orchestrator), còn các worker là Task con / thành viên:

  • Nếu một worker kết thúc sớm, main chỉ nhận kết quả và vẫn tiếp tục sống.

  • Nhưng nếu main bị lỗi (Exception) hoặc kết thúc sớm (ví dụ bạn return sớm trước khi đọc hết worker), thì asyncio.run() sẽ hủy bỏ (cancel) toàn bộ các worker còn lại và đóng Event Loop.

Tóm tắt trực quan

import asyncio

async def worker(task_name: str, duration: float):
    """Một coroutine con mô phỏng công việc tốn thời gian."""
    await asyncio.sleep(duration)
    return f"{task_name} hoàn thành"

def inspect_event_loop():
    """Hàm đồng bộ kiểm tra và in ra trạng thái các Task trong Event Loop."""
    current = asyncio.current_task()
    all_active_tasks = asyncio.all_tasks()

    print(f"\n🔍 [Snapshot] Tổng số Task đang tồn tại: {len(all_active_tasks)}")
    print(f"👉 Task đang thực thi lệnh này: {current.get_name()}")
    print("-" * 65)

    for i, task in enumerate(all_active_tasks, start=1):
        # Lấy thông tin chi tiết của từng Task
        name = task.get_name()
        is_done = task.done()
        is_cancelled = task.cancelled() if is_done else False
        status = "DONE" if is_done else ("CANCELLED" if is_cancelled else "PENDING")
        coro_name = task.get_coro().__qualname__

        is_current_flag = " (⭐️ Đang chạy)" if task is current else ""
        print(f"  {i}. Name: {name:<15} | Status: {status:<8} | Coro: {coro_name}{is_current_flag}")
    print("-" * 65 + "\n")

async def main():
    print("🚀 Bắt đầu hàm main()...")

    # 1. Tạo 3 task chạy ngầm và đặt tên tường minh qua tham số 'name'
    t1 = asyncio.create_task(worker("Worker-A", 3.0), name="Task-Worker-A")
    t2 = asyncio.create_task(worker("Worker-B", 1.0), name="Task-Worker-B")
    t3 = asyncio.create_task(worker("Worker-C", 5.0), name="Task-Worker-C")

    # Kiểm tra trạng thái ngay sau khi vừa tạo
    print("--- THỜI ĐIỂM 1: Vừa khởi tạo các Task ---")
    inspect_event_loop()

    # Chờ 1.5 giây để Task-Worker-B chạy xong
    await asyncio.sleep(1.5)

    print("--- THỜI ĐIỂM 2: Sau 1.5 giây (Task B đã xong) ---")
    inspect_event_loop()

    # Dọn dẹp: chờ các task còn lại hoàn tất trước khi main đóng
    await asyncio.gather(t1, t3)

if __name__ == "__main__":
    # asyncio.run tự động bọc main() thành một Task gốc
    asyncio.run(main())

🚀 Bắt đầu hàm main()...
--- THỜI ĐIỂM 1: Vừa khởi tạo các Task ---

🔍 [Snapshot] Tổng số Task đang tồn tại: 4
👉 Task đang thực thi lệnh này: Task-1
-----------------------------------------------------------------
  1. Name: Task-1          | Status: PENDING  | Coro: main (⭐️ Đang chạy)
  2. Name: Task-Worker-B   | Status: PENDING  | Coro: worker
  3. Name: Task-Worker-A   | Status: PENDING  | Coro: worker
  4. Name: Task-Worker-C   | Status: PENDING  | Coro: worker
-----------------------------------------------------------------

--- THỜI ĐIỂM 2: Sau 1.5 giây (Task B đã xong) ---

🔍 [Snapshot] Tổng số Task đang tồn tại: 3
👉 Task đang thực thi lệnh này: Task-1
-----------------------------------------------------------------
  1. Name: Task-1          | Status: PENDING  | Coro: main (⭐️ Đang chạy)
  2. Name: Task-Worker-A   | Status: PENDING  | Coro: worker
  3. Name: Task-Worker-C   | Status: PENDING  | Coro: worker
-----------------------------------------------------------------

PHẦN 4: BÀI TẬP THỰC CHIẾN – CÔNG CỤ WEB HEALTH CHECKER

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())

PHẦN 5: QUAN SÁT & THỬ THÁCH SUY LUẬN CHO BÉ

1. 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.

2. 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).

1. Phân biệt: Tạo Client vs Thực sự tạo Socket

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:

    1. Tra cứu DNS để đổi tên miền google.com thành IP.

    2. Tạo một kết nối Socket TCP tới IP đó (bắt tay 3 bước SYN, SYN-ACK, ACK).

    3. Thực hiện bắt tay TLS/SSL nếu là https.

    4. Gửi HTTP Request và nhận Response về.

2. Sức mạnh thực sự của 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.

3. Tóm lại

• 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

THỬ THÁCH 1: HIỂN THỊ KẾT QUẢ NGAY KHI XONG VỚI asyncio.as_completed

1. Bản chất khác biệt: gather vs as_completed

• asyncio.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.

2. Viết code nâng cấp cho WebHealthChecker

Cậ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!

THỬ THÁCH 2: ĐIỀU GÌ XẢY RA NẾU NÉM 1.000 URLs VÀO 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:

1. Ở phía Máy tính của bạn (Client-side):

• 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.

2. Ở phía Máy chủ đích (Server-side):

• 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.

3. Giải pháp chuẩn bị cho Buổi 5: 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.