Gọi HTTP Bất Đồng Tốc Độ Cao Với aiohttp Trong Python: Đỉnh Cao Xử Lý Yêu Cầu Mạng

Để khai thác triệt để sức mạnh của lập trình bất đồng bộ (async/await), aiohttp chính là thư viện Client/Server HTTP không thể thiếu. Bài viết này sẽ hướng dẫn bạn cách làm chủ aiohttp từ cơ bản đến nâng cao.

🟩 PHẦN 1: FOUNDATIONS & CLIENT SESSION (NỀN TẢNG VÀ PHIÊN KẾT NỐI)

1. Giới hạn của requests trong môi trường Async

Thư viện requests hoạt động theo cơ chế đồng bộ blocking. Nếu bạn đặt requests.get() vào bên trong một hàm async def, toàn bộ Event Loop sẽ bị đóng băng tại thời điểm chờ phản hồi mạng, làm vô hiệu hóa hoàn toàn tính bất đồng bộ của ứng dụng.

2. Sức mạnh của aiohttp.ClientSession

aiohttp cung cấp một đối tượng trung tâm là ClientSession. Bạn nên khởi tạo ClientSession một lần duy nhất cho vòng đời của ứng dụng hoặc sử dụng dưới dạng một Context Manager (async with) để quản lý kết nối (Connection Pooling) tối ưu, giúp tái sử dụng các kết nối TCP thay vì mở mới liên tục.

import aiohttp
import asyncio

async def fetch_single_page(url: str):
    # Khởi tạo ClientSession quản lý kết nối mạng
    async with aiohttp.ClientSession() as session:
        # Thực hiện gọi HTTP GET bất đồng bộ
        async with session.get(url) as response:
            print(f"Trạng thái phản hồi từ {url}: {response.status}")
            # Đọc nội dung dưới dạng text hoặc json bất đồng bộ
            html_content = await response.text()
            return html_content

async def main():
    await fetch_single_page("https://www.python.org")

# asyncio.run(main())

🟨 PHẦN 2: CONCURRENT REQUESTS & ADVANCED PATTERNS (GỌI ĐỒNG THỜI VÀ NÂNG CAO)

1. Gọi hàng loạt API đồng thời với asyncio.gather()

Sức mạnh thực sự của aiohttp bùng nổ khi bạn kết hợp nó với asyncio.gather(). Thay vì gửi từng request một cách tuần tự, bạn có thể bắn hàng trăm request đồng thời chỉ trong một tích tắc:

import aiohttp
import asyncio

async def fetch_url(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as response:
        return await response.json()  # Hoặc response.text()

async def main():
    urls = [
        "https://api.github.com/users/octocat",
        "https://api.github.com/users/torvalds",
        "https://api.github.com/users/gvanrossum"
    ]

    # Chia sẻ một ClientSession duy nhất cho toàn bộ các request
    async with aiohttp.ClientSession() as session:
        # Tạo danh sách các coroutine
        tasks = [fetch_url(session, url) for url in urls]
        
        # Chạy đồng thời tất cả và hứng kết quả
        results = await asyncio.gather(*tasks)
        
        for idx, data in enumerate(results):
            print(f"User {idx + 1} Name:", data.get("name"))

# asyncio.run(main())

2. Cấu hình Timeout an toàn

Khi gọi dịch vụ bên ngoài, việc đối mặt với các sự cố mạng chậm hoặc treo hệ thống là không tránh khỏi. Bạn phải luôn cấu hình timeout để tránh việc ứng dụng bị kẹt vô thời hạn:

import aiohttp
import asyncio

async def fetch_with_timeout(url: str):
    # Thiết lập giới hạn thời gian chờ tối đa 5 giây
    timeout = aiohttp.ClientTimeout(total=5.0)
    
    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            async with session.get(url) as response:
                return await response.text()
        except asyncio.TimeoutError:
            print("-> Yêu cầu đã quá thời gian chờ (Timeout)!")
        except aiohttp.ClientError as e:
            print(f"-> Lỗi kết nối mạng: {e}")

3. Kiểm soát lưu lượng với asyncio.Semaphore

Khi bạn cào dữ liệu (scraping) hoặc gọi hàng ngàn request đồng thời tới một máy chủ, hệ thống đích có thể bị quá tải hoặc chặn IP của bạn. Hãy sử dụng asyncio.Semaphore để giới hạn số lượng request được chạy song song tại cùng một thời điểm:

import aiohttp
import asyncio

async def bounded_fetch(semaphore: asyncio.Semaphore, session: aiohttp.ClientSession, url: str):
    # Chỉ cho phép tối đa N tác vụ chạy đồng thời
    async with semaphore:
        async with session.get(url) as response:
            print(f"Đang xử lý: {url}")
            return await response.status

async def main():
    urls = [f"https://httpbin.org/delay/1?id={i}" for i in range(10)]
    
    # Giới hạn tối đa 3 request chạy đồng thời cùng lúc
    semaphore = asyncio.Semaphore(3)
    
    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(semaphore, session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print("Tất cả kết quả giới hạn luồng:", results)

# asyncio.run(main())

📌 Lời Kết

aiohttp là công cụ không thể thiếu khi bạn muốn tối ưu hóa hiệu năng mạng trong Python. Bằng cách kết hợp linh hoạt ClientSession, asyncio.gather() và cơ chế kiểm soát giới hạn Semaphore, bạn có thể dễ dàng xây dựng các ứng dụng quét dữ liệu tốc độ cao, xử lý hàng ngàn API call một cách mượt mà và an toàn.

Từ khóa tìm hiểu thêm: aiohttp python tutorial, aiohttp ClientSession best practices, asyncio Semaphore aiohttp, Python asynchronous HTTP requests.