Quản Lý Tác Vụ Bất Đồng Bộ Nâng Cao: Làm Chủ asyncio.create_task Và asyncio.gather

Bài viết này sẽ dẫn dắt bạn đi qua hai vũ khí cốt lõi: Task (Tác vụ chạy nền) và hàm asyncio.gather() (Điều phối nhóm).

🟩 PHẦN 1: WORKING WITH TASKS (LÀM VIỆC VỚI TASK)

1. Kích hoạt chạy nền với asyncio.create_task()

Khi bạn gọi một hàm async bình thường, nó sẽ bắt buộc bạn phải await ngay lập tức. Tuy nhiên, nếu bạn muốn đưa một Coroutine vào chạy ngầm ngay lập tức (để nó tự động thực thi trong Event Loop trong lúc code chính vẫn tiếp tục chạy), hãy bọc nó bằng asyncio.create_task():

async def background_sync():
    print("-> Bắt đầu đồng bộ ngầm...")
    await asyncio.sleep(2)
    print("-> Đồng bộ ngầm hoàn tất!")
    return "Sync Success"

async def main():
    # Biến coroutine thành Task và cho phép nó chạy nền ngay lập tức
    task = asyncio.create_task(background_sync())
    
    print("-> Code chính vẫn tiếp tục chạy mà không bị nghẽn...")
    await asyncio.sleep(1)
    
    # Hứng kết quả từ Task khi cần thiết
    result = await task
    print("Nhận kết quả từ Task:", result)

# asyncio.run(main())

2. Chủ động hủy tác vụ với .cancel()

Trong quá trình chạy, nếu một tác vụ nền kéo dài quá lâu, tốn tài nguyên hoặc không còn cần thiết (ví dụ: người dùng đã bấm hủy thao tác), bạn hoàn toàn có thể gọi phương thức task.cancel() để chủ động dừng nó. Hãy luôn chuẩn bị sẵn khối try...except asyncio.CancelledError bên trong coroutine để dọn dẹp tài nguyên an toàn khi bị hủy:

async def long_running_task():
    try:
        print("-> Tác vụ dài đang chạy...")
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("-> Tác vụ đã bị hủy an toàn, đang dọn dẹp...")
        raise  # Nên raise lại để Event Loop ghi nhận trạng thái Cancelled

🟨 PHẦN 2: MASTERING ASYNCIO.GATHER() (LÀM CHỦ GATHER)

Khi bạn có một danh sách các tác vụ độc lập và muốn chúng chạy đồng thời, asyncio.gather() là lựa chọn hàng đầu.

1. Cú pháp giải nén với dấu sao (*)

Khi bạn đã gom các Coroutine hoặc Task vào bên trong một list, bạn bắt buộc phải dùng dấu sao (*) để giải nén các phần tử truyền vào hàm gather:

coroutine_list = [fetch_data(1), fetch_data(2), fetch_data(3)]

# Dùng dấu * để giải nén danh sách đưa vào gather
results = await asyncio.gather(*coroutine_list)

2. Bảo toàn tuyệt đối thứ tự kết quả

Một điểm cực kỳ thông minh của asyncio.gather() là danh sách kết quả trả về sẽ luôn xếp đúng theo thứ tự các Coroutine đầu vào, bất kể Coroutine nào hoàn thành trước hay hoàn thành sau. Nếu Coroutine số 3 chạy xong trước Coroutine số 1, Python vẫn tự động sắp xếp kết quả của số 1 ở vị trí đầu tiên trong mảng trả về.

3. Cô lập lỗi với tham số return_exceptions=True

Mặc định, nếu một tác vụ trong nhóm gather gặp lỗi (crash), toàn bộ các tác vụ khác sẽ bị hủy và ngoại lệ sẽ văng lên làm sập chương trình.

Để hệ thống an toàn hơn, bạn hãy bật cấu hình return_exceptions=True. Khi đó, nếu có một tác vụ lỗi, gather sẽ không làm sập chương trình mà lưu vết ngoại lệ đó trực tiếp vào mảng kết quả để bạn chủ động xử lý:

async def risky_task(number: int):
    if number == 2:
        raise ValueError("Lỗi xử lý dữ liệu số 2!")
    return number * 10

async def main():
    # Bật return_exceptions=True để cô lập lỗi
    results = await asyncio.gather(
        risky_task(1),
        risky_task(2),
        risky_task(3),
        return_exceptions=True
    )
    
    print(results)
    # Kết quả in ra: [10, ValueError("Lỗi xử lý dữ liệu số 2!"), 30]
    # Các tác vụ khác vẫn chạy bình thường và trả về kết quả!

📌 Lời Kết

Sự kết hợp uyển chuyển giữa asyncio.create_task (để kiểm soát tác vụ chạy nền) và asyncio.gather (để gom nhóm và đồng bộ hóa) sẽ giúp bạn làm chủ hoàn toàn luồng điều khiển trong lập trình bất đồng bộ Python, tối ưu hóa hiệu năng cho các ứng dụng thông minh.

Từ khóa tìm hiểu thêm: Python asyncio create_task, asyncio gather return_exceptions, Python handle CancelledError, Asyncio concurrent tasks Python.