Giới thiệu về Python AsyncIO

Một cách tiếp cận đơn giản để viết code asynchronous trong Python với asyncio

Lược dịch từ: https://realpython.com/async-io-python/

Gần đây, cùng với sự bùng nổ của các mô hình ngôn ngữ lớn (LLM), nhu cầu lập trình bất đồng bộ trong quá trình gọi LLM API ngày càng trở nên quan trọng. Việc gọi các API này thường mất nhiều thời gian, gây chậm trễ cho toàn bộ ứng dụng nếu xử lý theo cách tuần tự. Trong bài viết này, chúng ta sẽ cùng tìm hiểu lập trình bất đồng bộ trong Python với asyncio, và cách áp dụng nó để tăng tốc các chương trình cần tương tác nhiều với LLM API.

Trong bài viết lần này chúng ta sẽ cùng nhau tìm hiểu:

• Cách thư viện asyncio hỗ trợ lập trình bất đồng bằng cách sử dụng: coroutines, event loops, and non-blocking I/O operations.

• Và cách sử dụng asyncio khi ứng dụng cần nhiều thời gian để xử lý các I/O operations, ví dụ call API của bên thứ ba, đọc file. Và muốn chạy nhiều task concurrently mà không tạo thêm bất kỳ thread hay process.

Những khái niệm dễ gây nhầm lẫn

• Parallelism (tính song song) là khái niệm mô tả việc thực hiện nhiều tác vụ cùng một lúc — nghĩa là các tác vụ thực sự chạy song song trên nhiều lõi CPU.

• Multiprocessing là kỹ thuật lập trình tận dụng nhiều lõi CPU để đạt được parallelism. Mỗi tiến trình (process) chạy độc lập trên một lõi riêng biệt, giúp tăng hiệu năng cho các tác vụ nặng về tính toán (CPU-bound).

• Concurrency (tính đồng thời) là một khái niệm rộng hơn, nói đến khả năng xử lý nhiều tác vụ đan xen nhau trong cùng một khoảng thời gian. Tuy nhiên, các tác vụ này không nhất thiết phải chạy song song thực sự — chúng có thể được sắp xếp xen kẽ để tận dụng thời gian chờ, đặc biệt hiệu quả với các tác vụ I/O.

• Threading là một mô hình thực thi đồng thời, trong đó nhiều luồng (thread) chia sẻ cùng một không gian bộ nhớ và luân phiên nhau thực hiện các tác vụ. Một tiến trình duy nhất có thể chứa nhiều thread. Tuy nhiên, trong Python, threading có những giới hạn nhất định do sự tồn tại của GIL (Global Interpreter Lock), khiến cho các thread không thực sự chạy song song trên nhiều lõi CPU. Vấn đề này khá phức tạp và nằm ngoài phạm vi của bài viết này.

Bốn khái niệm trên không đồng nghĩa với "asynchronous" mà chúng ta đang đề cập đến trong bài viết này. asyncio hoạt động trong mô hình đơn luồng, đơn tiến trình (single-thread, single-process), hoàn toàn khác với threading hay multiprocessing. Nó không được xây dựng dựa trên các mô hình đó, mà dựa vào cơ chế event loop và lập trình bất đồng bộ bằng coroutine.

Vậy cái gì mới thực sự là “asynchronous”? Thật khó định nghĩa, nhưng tổng quát hoá nó có hai thuộc tính:

• Asynchronous routines can pause their execution while waiting for a result and allow other routines to run in the meantime.

• Asynchronous code facilitates the concurrent execution of tasks by coordinating asynchronous routines.

Ví dụ kinh điển về asynchronous các bạn có thể xem tại đây https://www.youtube.com/watch?v=iG6fr81xHKA&t=269s&ab_channel=PyCon2017

Kiện tướng cờ vua Judit Polgár tổ chức thì đấu với nhiều kỳ thủ nghiệp dư khác, cô ấy có hai cách để tổ chức cuộc chơi: synchronously and asynchronously.

Giả sử:

• Có tổng cộng 24 kỳ thủ

• Judit cần 5 giây để đi một nước, các kỳ thủ kia cần 55 giây để đi

• Trung bình mỗi ván đấu cần 30 lượt đi mỗi bên (tức 30 cặp nước đi qua lại).

Synchronous version: Judit chơi lần lượt từng ván, tập trung vào một người tại một thời điểm và chỉ bắt đầu ván tiếp theo khi ván hiện tại kết thúc.

• Mỗi ván mất khoảng:
  (5+55) giây × 30 = 1800 giây = 30 phút

• Tổng thời gian cho 24 ván:
  30 phút × 24 = 720 phút = 12 giờ

Asynchronously version: Judit di chuyển luân phiên qua từng bàn cờ. Cô đi nước đầu tiên với người thứ nhất, rồi tiếp tục sang người thứ hai, và cứ thế cho đến người thứ 24, sau đó quay lại bàn đầu để tiếp tục nước kế tiếp.

• Một vòng qua 24 bàn:
  5 giây × 24 = 120 giây = 2 phút

• Để hoàn thành 30 lượt đi:
  2 phút x 30 = 60 phút

Chỉ có một Judit Polgár, và cô chỉ có thể đi một nước cờ tại một thời điểm. Tuy nhiên, nhờ cách chơi bất đồng bộ, cô ấy đã rút ngắn thời gian thi đấu từ 12 giờ xuống còn 1 giờ. Lập trình bất đồng bộ (async I/O) trong Python cũng áp dụng đúng nguyên lý này.

Thay vì để một tác vụ dài làm chặn đứng toàn bộ chương trình (như một ván cờ kéo dài với từng người), async I/O sử dụng event loop để điều phối các tác vụ khác nhau, cho phép chúng luân phiên thực thi tại thời điểm tối ưu.

Nói cách khác, async I/O giúp chương trình xử lý những hàm mất thời gian (như chờ phản hồi API) theo cách mà các tác vụ khác vẫn có thể tiếp tục chạy trong lúc chờ đợi. Quay lại ví dụ cờ vua: trong khi một kỳ thủ đang suy nghĩ nước đi, Judit hoàn toàn có thể tiếp tục thi đấu với người tiếp theo — và đó chính là bản chất của async I/O.

Async I/O in Python With asyncio

Tiếp theo, chúng ta sẽ cùng nhau tìm hiểu cách sử dụng asyncio trong Python để giải quyết bài toán bất đồng bộ

Trung tâm của asyncio là khái niệm coroutine. Coroutine là object có khả năng tạm dừng (suspend) quá trình thực thi và quay trở lại sau đó. Có nghĩa là nó có thể trao lại quyền thư thi cho event loop để thực thi các coroutine khác.

Ví dụ, in ra một chương trinh hello world đơn giản như sau

import time

def count():
    print("One")
    time.sleep(1)
    print("Two")
    time.sleep(1)

def main():
    for _ in range(3):
        count()

if __name__ == "__main__":
    start = time.perf_counter()
    main()
    elapsed = time.perf_counter() - start
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

## Output:
One
Two
One
Two
One
Two
countsync.py executed in 6.03 seconds.

Đoạn code trên in ra từ “One” sau đó đợi 1 giấy, in tiếp từ Two sau đó đợi tiếp 1 giây nữa, tổng cộng sẽ tốn khoảng 6 giây để thực thi hết 3 lần như vậy. Chúng ta sẽ update doạn code trên bằng cách sử dụng asyncio

import asyncio

async def count():
    print("One")
    await asyncio.sleep(1)
    print("Two")
    await asyncio.sleep(1)

async def main():
    await asyncio.gather(count(), count(), count())

if __name__ == "__main__":
    import time

    start = time.perf_counter()
    asyncio.run(main())
    elapsed = time.perf_counter() - start
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

## Output:
One
One
One
Two
Two
Two
countasync.py executed in 2.00 seconds.

Khi sử dụng từ khóa async def, hàm count() trở thành một coroutine — tức là một hàm có thể tạm dừng và tiếp tục thực thi sau. Trong ví dụ này, coroutine sẽ in ra “One”, tạm dừng 1 giây, rồi tiếp tục in ra “Two”.

Tại thời điểm await asyncio.sleep(1), coroutine trao lại quyền điều khiển cho event loop, như thể nói: “Tôi sẽ tạm nghỉ trong 1 giây, bạn cứ tiếp tục xử lý các tác vụ khác trong lúc đó.” Đây chính là cách asyncio tận dụng thời gian chờ để xử lý các công việc khác thay vì bị chặn lại. Và đó là lý do nó chỉ tổn 2 giây để thực thi so với 6 giây ở ví dụ trước.

Một số thư viện tương thích với async I/O trong Python

Dưới đây và một vài thư viện trong Python sử dụng cơ chế Asynchronously programming.

aiohttp Asynchronous HTTP Client/Server for asyncio and Python.

Để gửi một async request trong aiohttp chúng ta cần ba bước, thay vì chỉ một bước như gửi một request thông thường

## async version:

async with aiohttp.ClientSession() as session:
    async with session.get('http://python.org') as response:
        print(await response.text())

## sync version:
response = requests.get('http://python.org')
print(response.text)

Nhưng tại sao lại vậy? Vì aiohttp là một thư viện asynchronous, nó được thiết kế để xử lý các tác vụ mạng một cách non-blocking. Trong ví dụ sync version, request block ba lần, nhưng với async, nó trao cơ hội cho event loop ba lần để chuyển đổi context

• Đầu tiên, khi gọi `.get()`, khi cả 2 version gửi GET request tới server. Đối với aiohttp, nó đánh dấu bằng async with để đảm bảo nó không block và trao quyền cho event loop làm việc khác

• Tiếp theo, khi dùng response.text trong requests, bạn chỉ đang đọc một thuộc tính. Lệnh gọi .get() trước đó đã tải toàn bộ dữ liệu phản hồi và giải mã nó theo cách blocking. Trong khi đó, aiohttp khi .get() được gọi chỉ tải về phần headers, và cho phép bạn quyết định thời điểm “trả giá” để tải phần thân (body) sau đó, thông qua một thao tác bất đồng bộ thứ hai. Vì vậy mới cần await response.text().

• Và cuối cùng, câu lệnh async with aiohttp.ClientSession() không thực hiện I/O khi bắt đầu khối lệnh, nhưng khi kết thúc nó sẽ đảm bảo đóng tất cả tài nguyên còn lại một cách an toàn. Việc này cũng diễn ra bất đồng bộ nên cần được đánh dấu rõ ràng. ClientSession còn là một công cụ tối ưu hiệu năng, vì nó quản lý một pool kết nối cho bạn, giúp tái sử dụng thay vì mở/đóng kết nối mới cho mỗi request. Bạn thậm chí có thể điều chỉnh kích thước pool bằng cách truyền vào một đối tượng connector.

Trong ví dụ tiếp theo chúng ta cần truy cập một danh sách 10 URLs và lấy nôi dụng từ chúng. Nếu chỉ đơn giản là sử dụng request, giải sử mỗi trang web cần 5s để lấy nội dung, vậy chúng ta cần 5s * 10 = 50s. Nhưng với aiohttp, mỗi khi send request tới 1 url, event loop không đợi và tiếp tục gửi request tới url thứ 2, như vậy chúng ta sẽ cần tối đã 5 s cho tất cả 10 requests. Ví dụ

import asyncio
import aiohttp

URLS = [f"https://httpbin.org/get?i={i}" for i in range(10)]

async def fetch(session, url):
    async with session.get(url) as response:
        text = await response.text()
        return {
            "url": url,
            "status": response.status,
            "length": len(text),
            "body": text
        }

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in URLS]
        results = await asyncio.gather(*tasks)
    return results

if __name__ == "__main__":
    results = asyncio.run(main())
    print(f"Fetched {len(results)} results")

aiofiles: https://pypi.org/project/aiofiles/

Một thư viện khác của python hỗ trợ việc lập trình bất đồng bộ cũng hay được sử dụng là aiofiles.

aiofiles là một thư viện Python giúp thực hiện đọc/ghi file bất đồng bộ (async I/O) bằng cách tận dụng asyncio và chạy thao tác file trong thread pool.
Điều này giúp chương trình không bị block event loop khi thao tác với file — phù hợp cho các ứng dụng async như web server, crawler, hoặc xử lý dữ liệu song song.

Có hai cách làm việc với iofiles cho việc đọc dữ liệu

## cách 1:
async with aiofiles.open('filename', mode='r') as f:
    contents = await f.read()
print(contents)


## cách 2:
async with aiofiles.open('filename') as f:
    async for line in f:
        ...

Kết luận

Chúng ta đã tìm hiểu cách asyncio I/O vận hành và một số thư viện Python hỗ trợ xử lý bất đồng bộ. Lập trình bất đồng bộ đặc biệt phù hợp khi làm việc với các tác vụ blocking như gọi API bên thứ ba, đọc/ghi file, hoặc thao tác mạng.

• Async I/O kết hợp multiprocessing: Hoàn toàn khả thi và đặc biệt hiệu quả khi tận dụng nhiều CPU core.

• Async I/O kết hợp multithreading: Lập trình đa luồng phức tạp hơn, tiêu tốn tài nguyên CPU; việc tạo quá nhiều thread có thể dẫn đến treo hoặc giảm hiệu năng hệ thống.

Vì vậy, khi thiết kế hệ thống, cần cân nhắc kỹ giữa async I/O, multiprocessing và multithreading để lựa chọn mô hình phù hợp. Cuối cùng, bất đồng bộ không phải lúc nào cũng giúp tăng hiệu suất — cần đo đạc, thử nghiệm và tối ưu hóa dựa trên đặc thù của từng ứng dụng.