Nếu bạn scrape, crawl hoặc tự động hóa bất cứ thứ gì ở quy mô lớn trong Python, sớm muộn gì trang web mục tiêu cũng nhận ra rằng mọi yêu cầu đều đến từ một IP và chặn bạn. Xoay proxy trong Python phân tán lưu lượng của bạn qua nhiều địa chỉ IP để không một IP nào kích hoạt giới hạn tốc độ hay bị cấm. Hướng dẫn này trình bày chính xác mã cho requests, httpx và aiohttp, cùng cách chọn giữa một cổng xoay và một pool IP được quản lý.
Có hai quyết định bao trùm định hình mọi thứ bên dưới. Thứ nhất: phiên xoay vòng so với phiên sticky — một IP mới cho mỗi yêu cầu, hay giữ một IP cho cả một luồng nhiều bước. Thứ hai: IP residential so với datacenter — xem ma trận quyết định loại proxy. Làm đúng hai điều đó thì phần mã là phần dễ.
1. Xoay bằng cổng (khuyến nghị). Bạn kết nối tới một điểm cuối duy nhất và nhà cung cấp cấp cho bạn một IP thoát mới trên mỗi kết nối mới. Không có danh sách phải quản lý, không có IP chết phải loại bỏ. Pool residential động của JIBAO hoạt động theo cách này: cùng host và port, một IP mới cho mỗi yêu cầu.
2. Danh sách IP tự quản lý. Bạn giữ một danh sách các mục host:port và chọn một mục cho mỗi yêu cầu. Kiểm soát nhiều hơn, nhưng bạn tự lo việc kiểm tra tình trạng, thử lại và theo dõi bị cấm. Hữu ích với IP datacenter chuyên dụng khi bạn cần các địa chỉ đã biết, ổn định.
Trường hợp đơn giản nhất — định tuyến một yêu cầu qua một cổng tự xoay cho bạn. Dùng socks5h:// (ký tự h ở cuối nghĩa là DNS được phân giải ở phía proxy, tránh rò rỉ các truy vấn tra cứu):
import requests
# JIBAO rotating residential gateway: a new exit IP per connection
PROXY = "socks5h://USERNAME:[email protected]:10001"
proxies = {"http": PROXY, "https": PROXY}
for _ in range(5):
r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=30)
print(r.json()["origin"]) # a different IP each loopCần requests hỗ trợ SOCKS5? Cài thêm gói: pip install "requests[socks]". Nếu bạn thích một điểm cuối proxy HTTP, đổi scheme thành http:// — phần còn lại y hệt.
import random
import requests
PROXY_POOL = [
"socks5h://USERNAME-country-us:[email protected]:10001",
"socks5h://USERNAME-country-uk:[email protected]:10001",
"socks5h://USERNAME-country-de:[email protected]:10001",
]
def fetch(url):
proxy = random.choice(PROXY_POOL)
return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
resp = fetch("https://example.com")
print(resp.status_code)Các luồng đăng nhập, giỏ hàng và kết quả phân trang sẽ hỏng nếu IP thay đổi giữa chừng tác vụ. Ghim một IP bằng cách thêm một token phiên vào username; dùng lại cùng token để giữ cùng một IP thoát trong suốt thời gian tồn tại của nó:
import requests
SESSION_ID = "task-0001"
PROXY = f"socks5h://USERNAME-session-{SESSION_ID}:[email protected]:10001"
s = requests.Session()
s.proxies = {"http": PROXY, "https": PROXY}
s.post("https://example.com/login", data={"u": "me", "p": "secret"})
s.get("https://example.com/dashboard") # same IP, session intacthttpx cho bạn một client hiện đại với HTTP/2 và async gốc. Lưu ý rằng các phiên bản mới hơn nhận một tham số proxy= duy nhất:
import httpx
PROXY = "socks5h://USERNAME:[email protected]:10001"
with httpx.Client(proxy=PROXY, timeout=30) as client:
for _ in range(5):
print(client.get("https://httpbin.org/ip").json()["origin"])Hỗ trợ SOCKS trong httpx cần pip install "httpx[socks]". Với HTTP/2, thêm http2=True vào client — nó làm lưu lượng của bạn trông giống một trình duyệt thật hơn, kết hợp tốt với khớp dấu vân tay TLS.
Để có thông lượng cao, hãy bắn nhiều yêu cầu cùng một lúc. Mỗi kết nối qua cổng đều có IP thoát riêng, nên tính đồng thời và việc xoay diễn ra cùng lúc:
import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector
PROXY = "socks5://USERNAME:[email protected]:10001"
async def fetch(url):
connector = ProxyConnector.from_url(PROXY)
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as r:
return await r.text()
async def main():
urls = ["https://httpbin.org/ip"] * 20
results = await asyncio.gather(*(fetch(u) for u in urls))
print(f"fetched {len(results)} pages")
asyncio.run(main())Cài connector SOCKS bằng pip install aiohttp_socks. Tạo một connector mới cho mỗi tác vụ để mỗi yêu cầu mở một kết nối proxy mới.
Việc xoay chỉ đáng giá nếu bạn thử lại các thất bại trên một IP mới. Hãy lùi theo cấp số nhân và xoay token phiên ở mỗi lần thử:
import requests
from time import sleep
def fetch_with_retry(url, max_retries=4):
for attempt in range(max_retries):
proxy = f"socks5h://USERNAME-session-r{attempt}:[email protected]:10001"
try:
r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
if r.status_code in (403, 429):
raise requests.HTTPError(f"blocked: {r.status_code}")
r.raise_for_status()
return r.text
except requests.RequestException:
sleep(2 ** attempt) # 1s, 2s, 4s, 8s
raise RuntimeError(f"failed after {max_retries} tries: {url}")Trước một lần chạy lớn, hãy xác nhận IP thực sự thay đổi. Gọi một endpoint echo vài lần và kiểm tra các IP khác nhau. Để cũng xác nhận dấu vân tay TLS của bạn không tự tố giác, hãy chạy yêu cầu đó qua công cụ kiểm tra dấu vân tay JA3/TLS miễn phí của JIBAO — một IP xoay vòng với dấu vân tay Python lộ liễu vẫn bị chặn.
Khi việc xoay đã vững, bức tường tiếp theo bạn gặp thường là dấu vân tay, không phải IP. Nếu các IP residential sạch vẫn bị 403, hãy đọc cách vượt dấu vân tay TLS bằng curl_cffi và toàn bộ công thức vượt Cloudflare.
Nhận 500MB lưu lượng miễn phí và một cổng residential xoay vòng cấp cho bạn một IP mới trên mỗi yêu cầu.
Bắt Đầu Dùng Thử Miễn PhíNgười dùng mới nhận 500MB khi đăng ký, cùng tiền thưởng cho lần nạp đầu tiên. Ưu đãi có thời hạn.