Cấu hình proxy cho Python aiohttp để scraping bất đồng bộ hiệu quả

Tại sao cần proxy khi scraping với aiohttp?

Khi chạy scraper bất đồng bộ với aiohttp, các request đi ra từ cùng một địa chỉ IP rất dễ bị trang web chặn do:

• Giới hạn tốc độ (rate limit) theo IP.
• Phát hiện bot qua fingerprint và hành vi request.
• Yêu cầu truy cập nội dung theo vùng địa lý (geo‑blocking).

Proxy giải quyết ba vấn đề trên bằng cách thay đổi IP xuất phát, cho phép bạn:

1. Xoay IP sau mỗi request hoặc theo phiên (sticky session).
2. Chọn IP tại quốc gia cần thiết để kiểm tra nội dung đa ngôn ngữ.
3. Giảm nguy cơ bị liệt kê đen (blacklist) vì lưu lượng được phân tán.

RoProxy cung cấp pool proxy residential, datacenter và mobile với API xoay IP tự động, phù hợp cho các tác vụ scraping quy mô lớn.

Chuẩn bị môi trường

pip install aiohttp aiohttp-socks tenacity

• aiohttp: client HTTP bất đồng bộ hiệu năng cao.
• aiohttp-socks: hỗ trợ proxy SOCKS5 (RoProxy cung cấp cả HTTP và SOCKS5).
• tenacity: thư viện retry linh hoạt, giúp xử lý lỗi mạng tự động.

Cấu hình proxy cơ bản

Đoạn mã sau minh hoạ một GET request đơn giản qua proxy HTTP:

import aiohttp

PROXY_URL = "http://user:pass@proxy.roproxy.com:8000"

async def fetch(session, url):
    async with session.get(url, proxy=PROXY_URL) as resp:
        return await resp.text()

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, "https://httpbin.org/ip")
        print(html)

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

Giải thích:

• PROXY_URL chứa thông tin xác thực (username:password) và endpoint proxy.
• Tham số proxy= trong session.get áp dụng cho request đó mà không ảnh hưởng các request khác.

Sử dụng proxy xoay (rotating) với RoProxy

RoProxy cho phép xoay IP qua endpoint /rotate hoặc cung cấp danh sách proxy sẵn có. Cách đơn giản nhất là lấy một proxy mới trước mỗi batch request:

import aiohttp
import asyncio

ROTATE_API = "https://api.roproxy.com/v1/rotate?token=YOUR_TOKEN"

async def get_proxy(session):
    async with session.get(ROTATE_API) as resp:
        data = await resp.json()
        return f"http://{data['username']}:{data['password']}@{data['host']}:{data['port']}"

async def fetch_with_rotate(session, url):
    proxy = await get_proxy(session)
    async with session.get(url, proxy=proxy) as resp:
        return await resp.text()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_rotate(session, f"https://httpbin.org/ip?i={i}") for i in range(10)]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(r)

asyncio.run(main())

Lưu ý: RoProxy cũng hỗ trợ sticky session bằng cách thêm tham số session_id vào API rotate, giúp giữ nguyên IP trong một khoảng thời gian (ví dụ 10 phút).

Triển khai sticky session khi cần thiết

Một số trang web yêu cầu đăng nhập hoặc giữ giỏ hàng, do đó bạn cần cùng một IP cho nhiều request liên tiếp. Ví dụ:

STICKY_API = "https://api.roproxy.com/v1/sticky?token=YOUR_TOKEN&session_id=mysession&ttl=600"

async def get_sticky_proxy(session):
    async with session.get(STICKY_API) as resp:
        data = await resp.json()
        return f"http://{data['username']}:{data['password']}@{data['host']}:{data['port']}"

Sử dụng get_sticky_proxy thay cho get_proxy trong các hàm scraping cần duy trì phiên.

Xử lý lỗi và retry tự động

Mạng không ổn định hoặc proxy bị chặn có thể gây ra exception. tenacity giúp retry với backoff exponent:

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import aiohttp.ClientError

@retry(
    wait=wait_exponential(multiplier=1, min=2, max=10),
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type(aiohttp.ClientError)
)
async def safe_fetch(session, url, proxy):
    async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=15)) as resp:
        resp.raise_for_status()
        return await resp.text()

Kết hợp với safe_fetch trong các task scraping để tăng độ bền vững.

Ví dụ hoàn chỉnh: scraper bất đồng bộ thu thập dữ liệu từ nhiều trang

import asyncio
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import aiohttp.ClientError

ROTATE_API = "https://api.roproxy.com/v1/rotate?token=YOUR_TOKEN"
SEM = asyncio.Semaphore(20)  # giới hạn 20 request đồng thời

async def get_proxy(session):
    async with session.get(ROTATE_API) as resp:
        data = await resp.json()
        return f"http://{data['username']}:{data['password']}@{data['host']}:{data['port']}"

@retry(
    wait=wait_exponential(multiplier=1, min=2, max=10),
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type(aiohttp.ClientError)
)
async def fetch_page(session, url):
    async with SEM:
        proxy = await get_proxy(session)
        async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=20)) as resp:
            resp.raise_for_status()
            return await resp.text()

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_page(session, u) for u in urls]
        pages = await asyncio.gather(*tasks, return_exceptions=True)
        for i, page in enumerate(pages):
            if isinstance(page, Exception):
                print(f"[ERROR] {urls[i]} -> {page}")
            else:
                print(f"[OK] {urls[i]} length={len(page)}")

if __name__ == "__main__":
    target_urls = [f"https://example.com/page/{i}" for i in range(1, 101)]
    asyncio.run(main(target_urls))

Điểm nổi bật:

• Semaphore kiểm soát độ song song, tránh quá tải proxy và server đích.
• Mỗi request tự động lấy proxy mới → IP luôn thay đổi.
• Retry với backoff giúp vượt qua lỗi transient.

Tối ưu hiệu suất: connection pooling, keep‑alive, limit concurrency

1. Reuse ClientSession – Tạo một session duy nhất cho toàn bộ chương trình để tận dụng connection pool của aiohttp.
2. Đặt limit và limit_per_host trong aiohttp.TCPConnector:

   connector = aiohttp.TCPConnector(limit=100, limit_per_host=20, ttl_dns_cache=300)
   async with aiohttp.ClientSession(connector=connector) as session:
       ...
   

3. Bật keep‑alive (mặc định đã bật) và điều chỉnh keepalive_timeout nếu cần.
4. Giảm overhead DNS bằng ttl_dns_cache và use_dns_cache=True.
5. Theo dõi latency – Ghi log thời gian phản hồi của mỗi proxy, loại bỏ proxy chậm khỏi pool.

Kết luận & lời khuyên sử dụng RoProxy

• Chọn loại proxy phù hợp: residential cho độ tin cậy cao, datacenter cho tốc độ, mobile khi cần IP di động thực.
• Kết hợp rotating + sticky: dùng rotating cho crawl rộng, sticky cho luồng đăng nhập/đặt hàng.
• Giám sát sức khỏe proxy – Tích hợp Prometheus/Grafana (xem bài “Tự động hóa kiểm tra sức khỏe proxy với Prometheus và RoProxy”) để phát hiện proxy down sớm.
• Tuân thủ pháp lý và robots.txt – Luôn kiểm tra điều khoản sử dụng của trang đích trước khi scraping quy mô lớn.

Với các mẫu mã trên, bạn có thể xây dựng scraper aiohttp mạnh mẽ, bền vững và sẵn sàng mở rộng. Hãy thử ngay với tài khoản dùng thử RoProxy để trải nghiệm pool proxy chất lượng cao, xoay IP tự động và hỗ trợ SOCKS5/HTTP đầy đủ.