Proxy trong Python requests và aiohttp: cấu hình trong 5 phút

```html

Khi phát triển các trình phân tích, tự động hóa thu thập dữ liệu hoặc kiểm tra dịch vụ web từ Python, thường cần sử dụng máy chủ proxy. Các thư viện requests và aiohttp cung cấp các cơ chế linh hoạt để làm việc với proxy, nhưng việc cấu hình chúng có những điểm cần lưu ý quan trọng. Trong hướng dẫn này, chúng ta sẽ xem xét các phương pháp đồng bộ và bất đồng bộ, cung cấp ví dụ cho proxy HTTP và SOCKS5, xem xét xoay vòng IP và xử lý lỗi.

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

Thư viện requests là tiêu chuẩn cho các yêu cầu HTTP trong Python. Cấu hình proxy được thực hiện thông qua tham số proxies, nhận một từ điển với các giao thức và địa chỉ của máy chủ proxy.

Ví dụ đơn giản với proxy HTTP:

import requests

# Cấu hình proxy
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# Thực hiện yêu cầu qua proxy
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

Lưu ý: đối với các yêu cầu HTTPS, cũng cần chỉ định giao thức http:// trong giá trị proxy (không phải https://). Điều này là do kết nối với máy chủ proxy được thiết lập qua HTTP, và sau đó qua phương thức CONNECT tạo ra một đường hầm cho lưu lượng HTTPS.

Sử dụng biến môi trường:

Thư viện requests tự động đọc proxy từ các biến môi trường HTTP_PROXY và HTTPS_PROXY:

import os
import requests

# Thiết lập qua biến môi trường
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# Proxy sẽ được áp dụng tự động
response = requests.get('https://httpbin.org/ip')
print(response.json())

Cách tiếp cận này thuận tiện cho việc container hóa (Docker) hoặc khi proxy được cấu hình ở cấp độ hệ thống. Tuy nhiên, để linh hoạt, nên truyền tham số proxies một cách rõ ràng.

Xác thực và SOCKS5 trong requests

Hầu hết các dịch vụ proxy thương mại yêu cầu xác thực bằng tên đăng nhập và mật khẩu. Trong requests, điều này được thực hiện thông qua định dạng URL với thông tin xác thực.

Proxy HTTP với xác thực:

import requests

# Định dạng: http://username:password@host:port
proxies = {
    'http': 'http://user123:pass456@proxy.example.com:8080',
    'https': 'http://user123:pass456@proxy.example.com:8080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())

Cấu hình proxy SOCKS5:

Để làm việc với SOCKS5, cần một thư viện bổ sung requests[socks] hoặc PySocks. Cài đặt:

pip install requests[socks]

Ví dụ sử dụng SOCKS5:

import requests

# SOCKS5 không có xác thực
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# SOCKS5 với xác thực
proxies_auth = {
    'http': 'socks5://user:pass@123.45.67.89:1080',
    'https': 'socks5://user:pass@123.45.67.89:1080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies_auth)
print(response.json())

Proxy SOCKS5 đặc biệt hữu ích khi làm việc với proxy cư trú, vì giao thức này cung cấp một đường hầm lưu lượng đáng tin cậy hơn và hỗ trợ UDP (cần thiết cho một số ứng dụng).

Xoay vòng proxy trong requests

Khi phân tích một khối lượng lớn dữ liệu, việc sử dụng một địa chỉ IP duy nhất dẫn đến việc bị chặn. Xoay vòng proxy là việc thay đổi IP theo chu kỳ để phân phối tải và vượt qua giới hạn tỷ lệ.

Xoay vòng đơn giản qua danh sách:

import requests
import itertools

# Danh sách máy chủ proxy
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

# Tạo một bộ lặp vô hạn
proxy_pool = itertools.cycle(proxy_list)

# Thực hiện các yêu cầu với xoay vòng
for i in range(10):
    proxy = next(proxy_pool)
    proxies = {'http': proxy, 'https': proxy}
    
    try:
        response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=5)
        print(f"Yêu cầu {i+1}: IP = {response.json()['origin']}")
    except Exception as e:
        print(f"Lỗi với proxy {proxy}: {e}")

Xoay vòng với phiên để giữ cookies:

import requests
from itertools import cycle

class ProxyRotator:
    def __init__(self, proxy_list):
        self.proxy_pool = cycle(proxy_list)
        self.session = requests.Session()
    
    def get(self, url, **kwargs):
        proxy = next(self.proxy_pool)
        self.session.proxies = {'http': proxy, 'https': proxy}
        return self.session.get(url, **kwargs)

# Sử dụng
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
]

rotator = ProxyRotator(proxy_list)

for i in range(5):
    response = rotator.get('https://httpbin.org/ip', timeout=5)
    print(f"Yêu cầu {i+1}: {response.json()['origin']}")

Xoay vòng ngẫu nhiên để không thể đoán trước:

import requests
import random

proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

def get_random_proxy():
    proxy = random.choice(proxy_list)
    return {'http': proxy, 'https': proxy}

# Mỗi yêu cầu với proxy ngẫu nhiên
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"Yêu cầu {i+1}: {response.json()['origin']}")

Xoay vòng ngẫu nhiên hiệu quả hơn khi làm việc với các trang web theo dõi các mẫu yêu cầu. Việc thay đổi IP theo trình tự có thể trông đáng ngờ, trong khi việc chọn ngẫu nhiên mô phỏng hành vi của nhiều người dùng khác nhau.

Cấu hình proxy trong aiohttp

Thư viện aiohttp được thiết kế cho các yêu cầu HTTP bất đồng bộ và rất quan trọng cho các trình phân tích có tải cao. Cấu hình proxy khác với requests — sử dụng tham số proxy (số ít).

Ví dụ cơ bản với proxy HTTP:

import aiohttp
import asyncio

async def fetch_with_proxy():
    proxy = 'http://123.45.67.89:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            data = await response.json()
            print(data)

# Chạy
asyncio.run(fetch_with_proxy())

Proxy với xác thực:

Trong aiohttp, xác thực được truyền qua đối tượng aiohttp.BasicAuth hoặc trực tiếp trong URL:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # Biến thể 1: Thông tin xác thực trong URL
    proxy = 'http://user123:pass456@proxy.example.com:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            print(await response.json())

# Biến thể 2: Qua BasicAuth (cho một số proxy)
async def fetch_with_basic_auth():
    proxy = 'http://proxy.example.com:8080'
    proxy_auth = aiohttp.BasicAuth('user123', 'pass456')
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', 
                                proxy=proxy, 
                                proxy_auth=proxy_auth) as response:
            print(await response.json())

asyncio.run(fetch_with_auth_proxy())

SOCKS5 trong aiohttp:

Để sử dụng SOCKS5, cần thư viện aiohttp-socks:

pip install aiohttp-socks

import asyncio
from aiohttp_socks import ProxyConnector
import aiohttp

async def fetch_with_socks5():
    connector = ProxyConnector.from_url('socks5://user:pass@123.45.67.89:1080')
    
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('https://httpbin.org/ip') as response:
            print(await response.json())

asyncio.run(fetch_with_socks5())

Khi làm việc với proxy di động để phân tích mạng xã hội hoặc thị trường, nên sử dụng aiohttp — tính bất đồng bộ cho phép xử lý hàng trăm yêu cầu song song mà không làm tắc nghẽn luồng thực thi.

Xoay vòng bất đồng bộ và nhóm proxy

Đối với các trình phân tích có tải cao, việc xoay vòng proxy hiệu quả với xử lý lỗi và tự động thay thế các IP không hoạt động là rất quan trọng. Hãy xem xét các mẫu nâng cao cho aiohttp.

Lớp để quản lý nhóm proxy:

import aiohttp
import asyncio
from itertools import cycle
from typing import List, Optional

class ProxyPool:
    def __init__(self, proxy_list: List[str]):
        self.proxy_list = proxy_list
        self.proxy_cycle = cycle(proxy_list)
        self.failed_proxies = set()
    
    def get_next_proxy(self) -> Optional[str]:
        """Lấy proxy hoạt động tiếp theo"""
        for _ in range(len(self.proxy_list)):
            proxy = next(self.proxy_cycle)
            if proxy not in self.failed_proxies:
                return proxy
        return None  # Tất cả các proxy không khả dụng
    
    def mark_failed(self, proxy: str):
        """Đánh dấu proxy là không hoạt động"""
        self.failed_proxies.add(proxy)
        print(f"Proxy {proxy} đã được đánh dấu là không khả dụng")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """Thực hiện yêu cầu với việc tự động thay đổi proxy khi có lỗi"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("Tất cả các proxy không khả dụng")
            
            try:
                async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10), **kwargs) as response:
                    return await response.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                print(f"Lỗi với proxy {proxy}: {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"Không thể thực hiện yêu cầu sau {max_retries} lần thử")

# Sử dụng
async def main():
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
        'http://user:pass@proxy3.example.com:8080',
    ]
    
    pool = ProxyPool(proxy_list)
    
    async with aiohttp.ClientSession() as session:
        # Thực hiện 10 yêu cầu với việc tự động xoay vòng
        tasks = [pool.fetch(session, 'https://httpbin.org/ip') for _ in range(10)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Yêu cầu {i+1} kết thúc với lỗi: {result}")
            else:
                print(f"Yêu cầu {i+1}: IP = {result.get('origin')}")

asyncio.run(main())

Xử lý song song với giới hạn đồng thời:

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # Giới hạn số yêu cầu đồng thời
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as response:
                data = await response.json()
                return {'url': url, 'ip': data.get('origin'), 'status': response.status}
        except Exception as e:
            return {'url': url, 'error': str(e)}

async def main():
    urls = [f'https://httpbin.org/ip' for _ in range(50)]  # 50 yêu cầu
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # Giới hạn: không quá 10 yêu cầu đồng thời
    semaphore = asyncio.Semaphore(10)
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, url, next(proxy_cycle), semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
        
        # Phân tích kết quả
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"Số yêu cầu thành công: {len(successful)}")
        print(f"Số yêu cầu không thành công: {len(failed)}")

asyncio.run(main())

Việc sử dụng asyncio.Semaphore là rất quan trọng khi làm việc với proxy — quá nhiều kết nối đồng thời qua một IP có thể dẫn đến việc bị chặn từ trang web mục tiêu hoặc nhà cung cấp proxy.

Xử lý lỗi và thời gian chờ

Làm việc với proxy thường đi kèm với nhiều lỗi hơn: thời gian chờ, ngắt kết nối, từ chối của máy chủ proxy. Xử lý lỗi đúng cách là chìa khóa cho sự ổn định của trình phân tích.

Các lỗi điển hình khi làm việc với proxy:

Lỗi	Nguyên nhân	Giải pháp
`ProxyError`	Máy chủ proxy không khả dụng	Chuyển sang proxy khác
`ConnectTimeout`	Proxy không phản hồi kịp thời	Tăng thời gian chờ hoặc thay đổi proxy
`ProxyAuthenticationRequired`	Tên đăng nhập/mật khẩu không chính xác	Kiểm tra thông tin xác thực
`SSLError`	Vấn đề với chứng chỉ SSL	Tắt kiểm tra SSL (không khuyến nghị)
`TooManyRedirects`	Proxy tạo ra vòng lặp chuyển hướng	Thay đổi proxy hoặc giới hạn chuyển hướng

Xử lý lỗi trong requests:

import requests
from requests.exceptions import ProxyError, ConnectTimeout, RequestException

def fetch_with_retry(url, proxies, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(
                url, 
                proxies=proxies, 
                timeout=(5, 10),  # (thời gian chờ kết nối, thời gian chờ đọc)
                allow_redirects=True,
                verify=True  # Kiểm tra chứng chỉ SSL
            )
            response.raise_for_status()  # Gây ra ngoại lệ khi 4xx/5xx
            return response.json()
            
        except ProxyError as e:
            print(f"Cố gắng {attempt + 1}: Proxy không khả dụng - {e}")
        except ConnectTimeout as e:
            print(f"Cố gắng {attempt + 1}: Thời gian chờ kết nối - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"Cố gắng {attempt + 1}: Lỗi HTTP {e.response.status_code}")
            if e.response.status_code == 407:  # Proxy Authentication Required
                print("Lỗi xác thực proxy!")
                break  # Không thử lại khi có lỗi xác thực
        except RequestException as e:
            print(f"Cố gắng {attempt + 1}: Lỗi chung - {e}")
        
        if attempt < max_retries - 1:
            print(f"Thử lại sau 2 giây...")
            import time
            time.sleep(2)
    
    raise Exception(f"Không thể thực hiện yêu cầu sau {max_retries} lần thử")

# Sử dụng
proxies = {'http': 'http://user:pass@proxy.example.com:8080', 'https': 'http://user:pass@proxy.example.com:8080'}
try:
    data = fetch_with_retry('https://httpbin.org/ip', proxies)
    print(data)
except Exception as e:
    print(f"Lỗi nghiêm trọng: {e}")

Xử lý lỗi trong aiohttp:

import aiohttp
import asyncio
from aiohttp import ClientError, ClientProxyConnectionError

async def fetch_with_retry(session, url, proxy, max_retries=3):
    for attempt in range(max_retries):
        try:
            timeout = aiohttp.ClientTimeout(total=10, connect=5)
            async with session.get(url, proxy=proxy, timeout=timeout) as response:
                response.raise_for_status()
                return await response.json()
                
        except ClientProxyConnectionError as e:
            print(f"Cố gắng {attempt + 1}: Lỗi kết nối đến proxy - {e}")
        except asyncio.TimeoutError:
            print(f"Cố gắng {attempt + 1}: Thời gian chờ")
        except aiohttp.ClientHttpProxyError as e:
            print(f"Cố gắng {attempt + 1}: Lỗi HTTP proxy - {e}")
            if e.status == 407:
                print("Lỗi xác thực proxy!")
                break
        except ClientError as e:
            print(f"Cố gắng {attempt + 1}: Lỗi chung của khách hàng - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"Không thể thực hiện yêu cầu sau {max_retries} lần thử")

async def main():
    proxy = 'http://user:pass@proxy.example.com:8080'
    async with aiohttp.ClientSession() as session:
        try:
            data = await fetch_with_retry(session, 'https://httpbin.org/ip', proxy)
            print(data)
        except Exception as e:
            print(f"Lỗi nghiêm trọng: {e}")

asyncio.run(main())

Cấu hình thời gian chờ:

Cấu hình thời gian chờ đúng cách là rất quan trọng cho sự ổn định. Các giá trị được khuyến nghị:

Thời gian chờ kết nối: 5-10 giây (thời gian để thiết lập kết nối với proxy)
Thời gian chờ đọc: 10-30 giây (thời gian để nhận phản hồi từ trang web mục tiêu)
Thời gian chờ tổng: 30-60 giây (thời gian tổng cho yêu cầu)

Đối với các proxy cư trú chậm, nên tăng thời gian chờ lên 20-30 giây cho mỗi kết nối, vì việc định tuyến qua các nhà cung cấp thực có thể mất nhiều thời gian hơn.

Thực tiễn tốt nhất và tối ưu hóa

Làm việc hiệu quả với proxy đòi hỏi phải tuân thủ một tập hợp các quy tắc để giảm thiểu việc bị chặn và tối đa hóa hiệu suất.

1. Sử dụng Session để tái sử dụng kết nối:

# requests: Session tái sử dụng các kết nối TCP
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # Nhanh hơn so với requests.get()

# aiohttp: Session là bắt buộc cho tính bất đồng bộ
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. Thiết lập User-Agent và tiêu đề thực tế:

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.5',
    'Accept-Encoding': 'gzip, deflate, br',
    'DNT': '1',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1'
}

proxies = {'http': proxy, 'https': proxy}
response = requests.get('https://example.com', headers=headers, proxies=proxies)

3. Giới hạn tỷ lệ (số yêu cầu mỗi giây):

import time
import requests

class RateLimiter:
    def __init__(self, max_requests_per_second):
        self.max_requests = max_requests_per_second
        self.interval = 1.0 / max_requests_per_second
        self.last_request_time = 0
    
    def wait(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < self.interval:
            time.sleep(self.interval - elapsed)
        self.last_request_time = time.time()

# Sử dụng: không quá 2 yêu cầu mỗi giây
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

for url in urls:
    limiter.wait()
    response = requests.get(url, proxies=proxies)

4. Ghi nhật ký và giám sát proxy:

import logging
from collections import defaultdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProxyMonitor:
    def __init__(self):
        self.stats = defaultdict(lambda: {'success': 0, 'failed': 0, 'total_time': 0})
    
    def log_request(self, proxy, success, response_time):
        stats = self.stats[proxy]
        if success:
            stats['success'] += 1
        else:
            stats['failed'] += 1
        stats['total_time'] += response_time
        
        # Ghi nhật ký mỗi 10 yêu cầu
        total = stats['success'] + stats['failed']
        if total % 10 == 0:
            avg_time = stats['total_time'] / total
            success_rate = stats['success'] / total * 100
            logger.info(f"Proxy {proxy}: {total} yêu cầu, thành công {success_rate:.1f}%, trung bình {avg_time:.2f}s")

monitor = ProxyMonitor()

# Trong mã yêu cầu
import time
start = time.time()
try:
    response = requests.get(url, proxies=proxies, timeout=10)
    monitor.log_request(proxy, True, time.time() - start)
except Exception as e:
    monitor.log_request(proxy, False, time.time() - start)
    logger.error(f"Lỗi với proxy {proxy}: {e}")

5. Bộ nhớ cache DNS để tăng tốc:

# aiohttp với bộ nhớ cache DNS
import aiohttp
from aiohttp.resolver import AsyncResolver

resolver = AsyncResolver(nameservers=['8.8.8.8', '8.8.4.4'])
connector = aiohttp.TCPConnector(resolver=resolver, ttl_dns_cache=300)

async with aiohttp.ClientSession(connector=connector) as session:
    # Các yêu cầu sẽ sử dụng bộ nhớ cache DNS trong 5 phút
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. Xử lý captcha và chặn:

Mẹo: Khi nhận được trạng thái 403, 429 hoặc captcha, nên:

Thay đổi proxy sang IP từ một subnet khác
Tăng độ trễ giữa các yêu cầu (đến 5-10 giây)
Thay đổi User-Agent và các tiêu đề khác
Sử dụng cookies từ các phiên thành công trước đó

So sánh requests và aiohttp cho proxy

Việc chọn giữa requests và aiohttp phụ thuộc vào nhiệm vụ và khối lượng dữ liệu. Hãy xem xét các sự khác biệt chính.

Tiêu chí	requests	aiohttp
Đồng bộ	Đồng bộ (chặn)	Bất đồng bộ (không chặn)
Hiệu suất	~10-50 yêu cầu/giây	~100-1000 yêu cầu/giây
Đơn giản của mã	Dễ hơn cho người mới bắt đầu	Cần kiến thức về async/await
Cấu hình proxy	Từ điển `proxies`	Tham số `proxy`
Hỗ trợ SOCKS5	Qua `requests[socks]`	Qua `aiohttp-socks`
Sử dụng bộ nhớ	Ít hơn (một luồng)	Nhiều hơn (nhiều tác vụ)
Tốt hơn cho	Các kịch bản đơn giản, <100 yêu cầu	Các trình phân tích, >1000 yêu cầu

Khi nào nên sử dụng requests:

Các kịch bản đơn giản cho các nhiệm vụ tạm thời
Lập mẫu và kiểm tra
Khối lượng yêu cầu nhỏ (đến 100 mỗi phút)
Khi đơn giản của mã và khả năng đọc là quan trọng
Tích hợp với các thư viện đồng bộ

Khi nào nên sử dụng aiohttp:

Phân tích khối lượng lớn dữ liệu (nghìn trang)
Giám sát nhiều nguồn thời gian thực
Dịch vụ API với tải cao
Khi tốc độ xử lý là rất quan trọng
Làm việc với WebSocket qua proxy

So sánh thực tế về hiệu suất:

# Thử nghiệm: 100 yêu cầu qua proxy

# requests (đồng bộ) - ~50 giây
import requests
import time

start = time.time()
proxies = {'http': proxy, 'https': proxy}
for i in range(100):
    response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(f"requests: {time.time() - start:.2f} giây")

# aiohttp (bất đồng bộ) - ~5 giây
import aiohttp
import asyncio

async def fetch_all():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get('https://httpbin.org/ip', proxy=proxy)
            for _ in range(100)
        ]
        await asyncio.gather(*tasks)

start = time.time()
asyncio.run(fetch_all())
print(f"aiohttp: {time.time() - start:.2f} giây")

Khi sử dụng proxy trung tâm dữ liệu để phân tích với tốc độ cao, aiohttp cho thấy lợi thế gấp 10-20 lần so với requests nhờ vào việc xử lý song song các yêu cầu.

Kết luận

Cấu hình proxy trong Python thông qua các thư viện requests và aiohttp là một kỹ năng cơ bản cho việc phát triển các trình phân tích, tự động hóa thu thập dữ liệu và vượt qua các hạn chế địa lý. Thư viện requests phù hợp cho các kịch bản đơn giản và lập mẫu nhờ vào API đồng bộ dễ hiểu, trong khi aiohttp cung cấp hiệu suất cao khi xử lý hàng ngàn yêu cầu thông qua kiến trúc bất đồng bộ.

Các điểm chính để làm việc hiệu quả với proxy trong Python: xử lý lỗi và thời gian chờ đúng cách, thực hiện xoay vòng địa chỉ IP để phân phối tải, sử dụng Session để tái sử dụng kết nối, cấu hình các tiêu đề và User-Agent thực tế, giám sát hiệu suất của các máy chủ proxy. Đối với proxy SOCKS5, cần các thư viện bổ sung — requests[socks] hoặc aiohttp-socks.

Khi chọn loại proxy cho việc phân tích, hãy xem xét đặc điểm của nhiệm vụ: cho các trình phân tích có tải cao với hàng ngàn yêu cầu, hãy chọn các proxy trung tâm dữ liệu nhanh, cho việc vượt qua các hệ thống chống bot nghiêm ngặt và làm việc với mạng xã hội, hãy chọn các proxy cư trú với IP thực của người dùng gia đình, và cho các nhiệm vụ yêu cầu tính ẩn danh tối đa và mô phỏng lưu lượng di động, các proxy di động với IP của các nhà cung cấp dịch vụ di động là tối ưu.

Nếu bạn dự định phát triển các trình phân tích có hiệu suất cao hoặc tự động hóa thu thập dữ liệu từ nhiều nguồn, chúng tôi khuyên bạn nên thử proxy cư trú — chúng cung cấp mức độ ẩn danh cao, rủi ro bị chặn tối thiểu và hoạt động ổn định với hầu hết các dịch vụ web bảo mật. Đối với các nhiệm vụ kỹ thuật với tốc độ xử lý cao, các proxy trung tâm dữ liệu với độ trễ thấp và băng thông cao cũng là lựa chọn tốt.

```

Cài đặt proxy trong Python requests và aiohttp: hướng dẫn đầy đủ với ví dụ mã