Trong lập trình Python hiện đại, asyncio đã trở thành thư viện cốt lõi để xử lý các tác vụ bất đồng bộ (asynchronous programming). Khi cần thực thi nhiều tác vụ I/O (như gọi API, truy vấn cơ sở dữ liệu, đọc/ghi file hoặc cào dữ liệu web) cùng một lúc, hàm asyncio.gather() chính là công cụ mạnh mẽ giúp bạn tiết kiệm thời gian chờ đợi và tối ưu hiệu năng tối đa.

Bài viết này từ DCSOFT sẽ hướng dẫn bạn chi tiết từ khái niệm, cách hoạt động, ví dụ thực tế đến các mẹo xử lý ngoại lệ (exception) khi sử dụng asyncio.gather().

1. asyncio.gather() Là Gì?

asyncio.gather() là một hàm tiện ích trong mô-đun asyncio của Python, cho phép bạn truyền vào một danh sách các awaitable objects (thường là các coroutine hoặc Task) và thực thi chúng đồng thời (concurrently) trong cùng một Event Loop.

Thay vì chạy lần lượt từng tác vụ theo cơ chế đồng bộ (synchronous):

asyncio.gather() cho phép các tác vụ chờ đợi nhau, giúp tổng thời gian thực thi chỉ tiệm cận thời gian của tác vụ lâu nhất:

2. Cú Pháp Cơ Bản Của asyncio.gather()

import asyncio

async def fetch_data(task_id: int, delay: int):
    print(f"Bắt đầu tác vụ {task_id}")
    await asyncio.sleep(delay)  # Mô phỏng tác vụ I/O tốn thời gian
    print(f"Hoàn thành tác vụ {task_id}")
    return f"Kết quả từ tác vụ {task_id}"

async def main():
    # Chạy 3 tác vụ bất đồng bộ cùng lúc
    results = await asyncio.gather(
        fetch_data(1, 2),
        fetch_data(2, 3),
        fetch_data(3, 1)
    )
    
    print("Tất cả kết quả thu được:", results)

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

Giải thích kết quả:

• Các tác vụ 1, 2, 3 sẽ bắt đầu gần như cùng một lúc.

• Tác vụ 3 (chờ 1 giây) sẽ hoàn thành đầu tiên, tiếp theo là tác vụ 1 (2 giây) và cuối cùng là tác vụ 2 (3 giây).

• Biến results nhận về một danh sách (list) chứa kết quả trả về theo đúng thứ tự truyền vào: ['Kết quả từ tác vụ 1', 'Kết quả từ tác vụ 2', 'Kết quả từ tác vụ 3'], bất kể tác vụ nào hoàn thành trước.

3. Truyền Danh Sách Tác Vụ Động Với Unpacking *

Trong thực tế phát triển phần mềm, bạn thường xử lý một danh sách công việc có độ dài linh hoạt (ví dụ: cào dữ liệu từ 100 URL). Bạn có thể sử dụng toán tử unpacking * trong Python:

async def main():
    urls_or_ids = [1, 2, 3, 4, 5]
    
    # Tạo danh sách các coroutine
    tasks = [fetch_data(i, delay=2) for i in urls_or_ids]
    
    # Unpack danh sách vào asyncio.gather
    results = await asyncio.gather(*tasks)
    print(f"Thu thập thành công {len(results)} kết quả!")

4. Xử Lý Lỗi Và Ngoại Lệ Với Tham Số return_exceptions

Một vấn đề phổ biến khi triển khai ứng dụng thực tế là: Chuyện gì xảy ra nếu 1 trong các tác vụ bị lỗi (raise exception)?

Mặc định, nếu một tác vụ gặp lỗi, asyncio.gather() sẽ ném ngoại lệ đó ra ngay lập tức và các kết quả khác sẽ bị mất. Để kiểm soát điều này, hãy sử dụng tham số return_exceptions=True.

import asyncio

async def task_success():
    await asyncio.sleep(1)
    return "Thành công"

async def task_failed():
    await asyncio.sleep(1)
    raise ValueError("Lỗi kết nối CSDL!")

async def main():
    results = await asyncio.gather(
        task_success(),
        task_failed(),
        return_exceptions=True  # Không làm sập ứng dụng khi có lỗi
    )
    
    for res in results:
        if isinstance(res, Exception):
            print(f"[Xử lý lỗi]: Tìm thấy lỗi -> {res}")
        else:
            print(f"[Thành công]: Kết quả -> {res}")

asyncio.run(main())

Lời khuyên từ DCSOFT: Luôn đặt return_exceptions=True khi gọi các dịch vụ bên ngoài (External APIs, Web Scraping) để đảm bảo toàn bộ hệ thống không bị gián đoạn chỉ vì một tác vụ nhỏ bị lỗi.

5. So Sánh asyncio.gather() Với asyncio.as_completed() Và TaskGroup

6. Tổng Kết

asyncio.gather() là một giải pháp cực kỳ hiệu quả giúp lập trình viên Python khai thác tối đa sức mạnh bất đồng bộ cho các bài toán xử lý I/O bound. Việc làm chủ hàm này cùng kỹ năng quản lý lỗi sẽ giúp bạn xây dựng nên những hệ thống web API, hệ thống cào dữ liệu hay công cụ tự động hóa mượt mà và tối ưu tài nguyên.

Nếu bạn đang tìm kiếm các giải pháp tự động hóa, phát triển ứng dụng web hay tối ưu hệ thống chuyên nghiệp, hãy tham khảo thêm các dịch vụ công nghệ chất lượng cao tại dcsofttech.com!