aiohttp 2026: 초당 10K+ 요청을 처리하는 고성능 비동기 웹 스크래퍼 구축 — Python 가이드

pip install aiohttp==3.11.0

# speedups 포함 (프로덕션 권장)
pip install aiohttp[speedups]==3.11.0

# 스크래핑용 추가 도구 설치
pip install aiohttp==3.11.0 aiofiles==24.1.0 beautifulsoup4==4.12.3 lxml==5.3.0

import aiohttp
import asyncio
import sys

print(f"aiohttp version: {aiohttp.__version__}")
print(f"Python version: {sys.version}")

async def check():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://httpbin.org/get") as resp:
            data = await resp.json()
            print(f"Status: {resp.status}")
            print(f"Response keys: {list(data.keys())}")

asyncio.run(check())

import aiohttp
import asyncio

urls = [
    "https://httpbin.org/get?param=1",
    "https://httpbin.org/get?param=2",
    "https://httpbin.org/get?param=3",
]

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(r["args"])

asyncio.run(main())

import aiohttp
import asyncio
from bs4 import BeautifulSoup

async def scrape_titles(session, urls):
    """여러 URL에서 동시에 페이지 제목을 추출합니다."""
    titles = []
    for url in urls:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                html = await resp.text()
                soup = BeautifulSoup(html, "lxml")
                title = soup.find("title")
                titles.append({"url": url, "title": title.text if title else "N/A"})
        except Exception as e:
            titles.append({"url": url, "title": f"Error: {e}"})
    return titles

async def main():
    urls = ["https://example.com", "https://httpbin.org/html"]
    async with aiohttp.ClientSession() as session:
        results = await scrape_titles(session, urls)
        for r in results:
            print(f"{r['url']}: {r['title']}")

asyncio.run(main())

import aiohttp
import asyncio
from lxml import html as lh

async def extract_links(session, url):
    """lxml을 사용하여 페이지에서 모든 href 링크를 추출합니다."""
    async with session.get(url) as resp:
        text = await resp.text()
        tree = lh.fromstring(text)
        links = tree.xpath("//a/@href")
        return [l for l in links if l.startswith("http")]

async def main():
    async with aiohttp.ClientSession() as session:
        links = await extract_links(session, "https://example.com")
        print(f"Found {len(links)} external links")

asyncio.run(main())

import aiohttp
import aiofiles
import asyncio
import json

async def scrape_and_save(session, url, filepath):
    """데이터를 스크래핑하고 비동기적으로 디스크에 씁니다."""
    async with session.get(url) as resp:
        data = await resp.json()
        async with aiofiles.open(filepath, "w") as f:
            await f.write(json.dumps(data, indent=2))

async def main():
    async with aiohttp.ClientSession() as session:
        await scrape_and_save(
            session,
            "https://httpbin.org/json",
            "/tmp/scraped_data.json"
        )

asyncio.run(main())

import aiohttp
import aiosqlite
import asyncio

async def scrape_to_db(session, db, url):
    """스크래핑한 데이터를 SQLite에 비동기적으로 저장합니다."""
    async with session.get(url) as resp:
        data = await resp.json()
        await db.execute(
            "INSERT INTO scraped (url, data) VALUES (?, ?)",
            (url, json.dumps(data))
        )
        await db.commit()

async def main():
    async with aiosqlite.connect("scraped.db") as db:
        await db.execute("CREATE TABLE IF NOT EXISTS scraped (url TEXT, data TEXT)")
        async with aiohttp.ClientSession() as session:
            await scrape_to_db(session, db, "https://httpbin.org/json")

asyncio.run(main())

import aiohttp
import asyncio

PROXY_URL = "http://username:password@proxy.webshare.io:80"

async def fetch_with_proxy(session, url):
    """WebShare 로테이션 프록시를 통해 요청을 라우팅합니다."""
    async with session.get(url, proxy=PROXY_URL) as resp:
        return await resp.text()

async def main():
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)
    async with aiohttp.ClientSession(connector=connector) as session:
        html = await fetch_with_proxy(session, "https://httpbin.org/ip")
        print(html[:200])

asyncio.run(main())

import aiohttp

connector = aiohttp.TCPConnector(
    limit=200,              # 총 동시 연결 수
    limit_per_host=20,      # 호스트당 연결 수 (서버를 존중하세요!)
    ttl_dns_cache=300,      # DNS 캐시 TTL (초)
    use_dns_cache=True,     # DNS 캐싱 활성화
    enable_cleanup_closed=True,
    force_close=False,      # 연결 유지
)

timeout = aiohttp.ClientTimeout(
    total=30,               # 요청당 총 타임아웃
    connect=5,              # TCP 연결 타임아웃
    sock_read=15,           # 소켓 읽기 타임아웃
)

session = aiohttp.ClientSession(
    connector=connector,
    timeout=timeout,
    headers={"User-Agent": "MyBot/1.0"},
)

import aiohttp
import asyncio

async def bounded_fetch(session, url, semaphore):
    """세마포어로 동시 요청을 제한합니다."""
    async with semaphore:
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    semaphore = asyncio.Semaphore(50)  # 최대 50개 동시 요청
    urls = [f"https://httpbin.org/get?i={i}" for i in range(500)]
    
    connector = aiohttp.TCPConnector(limit=100)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [bounded_fetch(session, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        successes = sum(1 for r in results if not isinstance(r, Exception))
        print(f"Successful: {successes}/500")

asyncio.run(main())

import aiohttp
import asyncio
import random

async def fetch_with_retry(session, url, max_retries=3):
    """지수 백오프로 실패한 요청을 재시도합니다."""
    for attempt in range(max_retries):
        try:
            async with session.get(url) as resp:
                if resp.status == 200:
                    return await resp.json()
                elif resp.status in (429, 503, 502):
                    wait = (2 ** attempt) + random.uniform(0, 1)
                    await asyncio.sleep(wait)
                else:
                    resp.raise_for_status()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)
    return None

async def main():
    async with aiohttp.ClientSession() as session:
        data = await fetch_with_retry(session, "https://httpbin.org/json")
        print(data)

asyncio.run(main())

import aiohttp
import asyncio

async def websocket_scraper():
    """WebSocket 엔드포인트에서 실시간 데이터를 스크래핑합니다."""
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect("wss://echo.websocket.org") as ws:
            await ws.send_str("Hello Server")
            
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    print(f"Received: {msg.data}")
                    if "done" in msg.data.lower():
                        await ws.close()
                        break
                elif msg.type == aiohttp.WSMsgType.ERROR:
                    print(f"WebSocket error: {ws.exception()}")
                    break

asyncio.run(websocket_scraper())

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY scraper.py .
CMD ["python", "scraper.py"]

# docker-compose.yml
version: "3.8"
services:
  scraper:
    build: .
    restart: unless-stopped
    environment:
      - PYTHONUNBUFFERED=1
    deploy:
      resources:
        limits:
          memory: 2G
    logging:
      driver: "json-file"
      options:
        max-size: "100m"
        max-file: "3"

import aiohttp
import asyncio
from prometheus_client import Counter, Histogram, start_http_server

REQUEST_COUNT = Counter("scraper_requests_total", "총 요청 수", ["status"])
REQUEST_DURATION = Histogram("scraper_request_duration_seconds", "요청 소요 시간")

async def monitored_fetch(session, url):
    with REQUEST_DURATION.time():
        try:
            async with session.get(url) as resp:
                REQUEST_COUNT.labels(status=str(resp.status)).inc()
                return await resp.text()
        except Exception as e:
            REQUEST_COUNT.labels(status="error").inc()
            raise

# 포트 9090에서 메트릭 서버 시작
start_http_server(9090)