Điều phối Tác vụ Bất đồng bộ với Giới hạn Concurrency bằng Semaphore

Mục tiêu kiến thức

• Sử dụng asyncio.Semaphore để giới hạn số lượng tác vụ I/O chạy song song (Concurrency Throttling).
• Phối hợp asyncio.gather để thu thập toàn bộ kết quả mà vẫn bảo toàn nguyên vẹn thứ tự ban đầu.
• Ngăn chặn lỗi cạn kiệt kết nối mạng (Socket Exhaustion) hoặc bị chặn IP (Rate Limit Exceeded) khi gọi API ngoài.

Mô tả bài toán

Hệ thống AI Empire Academy cần gửi đồng loạt 1000 truy vấn trích xuất đặc trưng văn bản tới cụm máy chủ AI. Nếu khởi chạy cùng lúc tất cả các coroutine, máy chủ mục tiêu sẽ trả về mã lỗi 429 Too Many Requests.

Hãy viết hàm bất đồng bộ: async def bounded_gather(coro_factories: list[Callable[[], Coroutine]], max_concurrency: int) -> list[Any]

Quy tắc hoạt động:

• Kiểm tra tham số: Nếu max_concurrency <= 0, ném ngoại lệ ValueError("max_concurrency phai lon hon 0").
• Khởi tạo và điều phối:
• Mỗi phần tử trong coro_factories là một hàm không có tham số, khi gọi sẽ sinh ra một coroutine (ví dụ: lambda: fetch_result(i)).
• Sử dụng asyncio.Semaphore(max_concurrency) để giới hạn: tại bất kỳ thời điểm nào, số lượng coroutine đang thực thi trong khối semaphore không bao giờ vượt quá max_concurrency.
• Kết quả:
• Sử dụng asyncio.gather để chờ toàn bộ tác vụ hoàn thành.
• Danh sách trả về phải chứa kết quả theo đúng thứ tự tương ứng với coro_factories ban đầu, bất kể tác vụ nào hoàn thành trước hay sau.
• Xử lý ngoại lệ: Nếu một tác vụ bất đồng bộ ném ra ngoại lệ, ngoại lệ đó phải được lan truyền ra ngoài theo hành vi mặc định của asyncio.gather.

Ràng buộc

• 1 ≤ len(coro_factories) ≤ 1000.
• 1 ≤ max_concurrency ≤ 100.
• Thời gian chạy tối đa: 1500ms.
• Giới hạn bộ nhớ: 256MB.