Hàm asyncio.gather() trong Python

Trong lập trình bất đồng bộ, việc thực hiện nhiều tác vụ cùng lúc là một yêu cầu phổ biến. Python cung cấp thư viện asyncio để hỗ trợ việc này, và một trong những hàm quan trọng trong asyncio là asyncio.gather(). Hàm này cho phép bạn chạy đồng thời nhiều tác vụ bất đồng bộ và thu thập kết quả của chúng khi tất cả đều hoàn thành. Trong bài viết này, bạn sẽ học cách sử dụng hàm asyncio.gather() để quản lý và thực hiện các tác vụ bất đồng bộ một cách hiệu quả trong Python.

Giới thiệu về hàm asyncio.gather() trong Python

Đôi khi, bạn có thể muốn chạy nhiều tác vụ bất đồng bộ và nhận kết quả khi tất cả hoàn thành. Để làm điều đó, bạn có thể sử dụng hàm asyncio.gather():

gather(*aws, return_exceptions=False) -> Future[tuple[()]]

Hàm asyncio.gather() có hai tham số:

• aws là một dãy các đối tượng awaitable. Nếu bất kỳ đối tượng nào trong aws là coroutine, hàm asyncio.gather() sẽ tự động chuyển nó thành một task.
• return_exceptions mặc định là False. Nếu có ngoại lệ xảy ra trong một đối tượng awaitable, nó sẽ ngay lập tức được truyền đến task đang chờ đợi asyncio.gather(). Các đối tượng awaitable khác sẽ tiếp tục chạy và không bị hủy bỏ.

Hàm asyncio.gather() trả về kết quả của các đối tượng awaitable dưới dạng một tuple với thứ tự giống như bạn đã truyền vào hàm.

Bài viết này được đăng tại [free tuts .net]

Nếu return_exceptions được đặt là True, asyncio.gather() sẽ thêm ngoại lệ (nếu có) vào kết quả và không truyền ngoại lệ đó đến hàm gọi.

Ví dụ về hàm asyncio.gather() trong Python

Dưới đây là một số ví dụ minh họa cách sử dụng hàm asyncio.gather().

Sử dụng asyncio.gather() để chạy nhiều tác vụ bất đồng bộ

Ví dụ sau sử dụng asyncio.gather() để chạy hai tác vụ bất đồng bộ và hiển thị kết quả:

import asyncio

async def call_api(message, result, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result

async def main():
    a, b = await asyncio.gather(
        call_api('Calling API 1 ...', 100, 1),
        call_api('Calling API 2 ...', 200, 2)
    )
    print(a, b)

asyncio.run(main())

Kết quả:

Calling API 1 ...
Calling API 2 ...
100 200

Cách hoạt động

Đầu tiên, định nghĩa một coroutine call_api() để mô phỏng một tác vụ bất đồng bộ. call_api() hiển thị một thông báo, trì hoãn một số giây, và trả về một kết quả:

async def call_api(message, result, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result

Tiếp theo, sử dụng hàm asyncio.gather() để chạy hai coroutine call_api():

async def main():
    a, b = await asyncio.gather(
        call_api('Calling API 1 ...', 100, 1),
        call_api('Calling API 2 ...', 200, 2)
    )
    print(a, b)

Corotine đầu tiên mất 1 giây và trả về 100 trong khi coroutine thứ hai mất 2 giây và trả về 200. Sau 2 giây, asyncio.gather() trả về kết quả dưới dạng một tuple chứa kết quả của hai coroutine.

Lưu ý rằng a là 100 và b là 200, tương ứng với kết quả của các coroutine mà chúng ta đã truyền vào hàm asyncio.gather().

Sử dụng asyncio.gather() để chạy nhiều tác vụ bất đồng bộ với ngoại lệ

Ví dụ sau đây minh họa cách sử dụng hàm asyncio.gather() để thực thi nhiều tác vụ bất đồng bộ khi có một tác vụ gây ra ngoại lệ:

import asyncio

class APIError(Exception):
    def __init__(self, message):
        self._message = message

    def __str__(self):
        return self._message

async def call_api_failed():
    await asyncio.sleep(1)
    raise APIError('API failed')

async def call_api(message, result, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result

async def main():
    try:
        a, b, c = await asyncio.gather(
            call_api('Calling API 1 ...', 100, 1),
            call_api('Calling API 2 ...', 200, 2),
            call_api_failed()
        )
        print(a, b, c)
    except APIError as e:
        print(f"Caught an APIError: {e}")

asyncio.run(main())

Kết quả sẽ cho thấy ngoại lệ APIError được bắt và in ra:

Calling API 1 ...
Calling API 2 ...
Caught an APIError: API failed

Sử dụng asyncio.gather() để trả về một ngoại lệ trong kết quả

Để nhận được ngoại lệ trong kết quả, bạn đặt tham số return_exceptions thành True như sau:

import asyncio

class APIError(Exception):
    def __init__(self, message):
        self._message = message

    def __str__(self):
        return self._message

async def call_api(message, result, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result

async def call_api_failed():
    await asyncio.sleep(1)
    raise APIError('API failed')

async def main():
    a, b, c = await asyncio.gather(
        call_api('Calling API 1 ...', 100, 1),
        call_api('Calling API 2 ...', 200, 2),
        call_api_failed(),
        return_exceptions=True
    )
    print(a, b, c)

asyncio.run(main())

Kết quả:

Calling API 1 ...
Calling API 2 ...
100 200 API failed

Trong ví dụ này, vì return_exceptions được đặt là True, asyncio.gather() sẽ trả về ngoại lệ dưới dạng một phần của kết quả.

Kết bài

Hàm asyncio.gather() chạy nhiều tác vụ bất đồng bộ, chuyển đổi coroutine thành task, và trả về một tuple chứa các kết quả theo thứ tự của các awaitable. Đặt return_exceptions thành True để cho phép các lỗi được trả về dưới dạng kết quả.