Cách xoay vòng proxy trong Python: Hướng dẫn đầy đủ 2026

Cách xoay vòng proxy trong Python: Hướng dẫn đầy đủ 2026

Retry

pip install requests
pip install "requests[socks]"  # chỉ cần thiết cho các URL proxy socks5://

import requests

proxies = {
    "http": "http://10.10.1.10:3128",
    "https": "http://10.10.1.10:1080",
}
requests.get("http://example.org", proxies=proxies)

import os
import requests

os.environ["HTTP_PROXY"] = "http://127.0.0.1:8002"

session = requests.Session()
session.proxies.update({"http": "http://127.0.0.1:8001"})

resp = session.get("http://internal.test/", timeout=5)
print(resp.json()["served_by"])

proxy-2

import itertools
import requests

PROXIES = [
    "http://127.0.0.1:8001",
    "http://127.0.0.1:8002",
    "http://127.0.0.1:8003",
]

proxy_cycle = itertools.cycle(PROXIES)

for i in range(6):
    proxy = next(proxy_cycle)
    resp = requests.get(
        "http://internal.test/products",
        proxies={"http": proxy, "https": proxy},
        timeout=5,
    )
    print(f"yêu cầu {i + 1}: gửi qua {proxy} -> {resp.json()['served_by']}")

yêu cầu 1: gửi qua http://127.0.0.1:8001 -> proxy-1
yêu cầu 2: gửi qua http://127.0.0.1:8002 -> proxy-2
yêu cầu 3: gửi qua http://127.0.0.1:8003 -> proxy-3
yêu cầu 4: gửi qua http://127.0.0.1:8001 -> proxy-1
yêu cầu 5: gửi qua http://127.0.0.1:8002 -> proxy-2
yêu cầu 6: gửi qua http://127.0.0.1:8003 -> proxy-3

import random
import requests

def get_with_rotation(url, proxy_pool, max_attempts=4, timeout=3):
    pool = list(proxy_pool)
    random.shuffle(pool)
    last_error = None
    for attempt, proxy in enumerate(pool[:max_attempts], start=1):
        try:
            resp = requests.get(
                url,
                proxies={"http": proxy, "https": proxy},
                timeout=timeout,
            )
            resp.raise_for_status()
            return proxy, resp
        except (requests.exceptions.ProxyError,
                requests.exceptions.ConnectionError,
                requests.exceptions.Timeout) as exc:
            print(f"cố gắng {attempt}: {proxy} thất bại ({exc.__class__.__name__}), xoay vòng")
            last_error = exc
            continue
    raise last_error

cố gắng 1: http://127.0.0.1:8004 thất bại (ProxyError), xoay vòng
thành công qua http://127.0.0.1:8003 -> proxy-3

import itertools
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

PROXIES = ["http://127.0.0.1:8001", "http://127.0.0.1:8002", "http://127.0.0.1:8003"]
_lock = threading.Lock()
_cycle = itertools.cycle(PROXIES)

def next_proxy():
    with _lock:
        return next(_cycle)

def fetch(path):
    proxy = next_proxy()
    resp = requests.get(f"http://internal.test{path}", proxies={"http": proxy, "https": proxy}, timeout=5)
    return path, proxy, resp.json()["served_by"]

đường dẫn = [f"/item/{i}" cho i trong phạm vi(9)]
với ThreadPoolExecutor(max_workers=4) làm hồ bơi:
    tương lai = [hồ bơi.submit(fetch, p) cho p trong đường dẫn]
    cho tương lai trong as_completed(tương lai):
        in(tương lai.result())