Python爬虫多线程并发时的503错误处理最佳实践_小淘米

2025-06-26 16:55

标签:

一、503 错误产生的原因

在 HTTP 协议中，503 错误表示服务器当前无法处理请求，通常是因为服务器暂时过载或维护。在多线程爬虫场景下，503 错误可能由以下几种原因引起：

服务器负载过高：当多个线程同时向服务器发送请求时，服务器可能因负载过高而拒绝部分请求，返回 503 错误。
请求频率过快：如果爬虫的请求频率超过了服务器的处理能力，服务器可能会认为这是一种攻击行为，从而返回 503 错误。
服务器配置问题：某些服务器可能配置了特定的防护机制，如防火墙或反爬虫策略，当检测到异常请求时会返回 503 错误。
网络问题：网络不稳定或代理服务器故障也可能导致 503 错误。

二、503 错误处理的最佳实践

（一）合理控制并发线程数量

过多的并发线程会增加服务器的负载，导致 503 错误。因此，合理控制并发线程的数量是避免 503 错误的关键。可以通过设置线程池来限制并发线程的数量。

import concurrent.futures
import requests

def fetch_url(url):
    try:
        response = requests.get(url)
        response.raise_for_status()
        return response.text
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 503:
            print(f"503 error occurred for {url}")
            # Handle 503 error
        else:
            raise

def main():
    urls = ["http://example.com/page1", "http://example.com/page2", ...]
    max_workers = 10  # 控制并发线程数量
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(fetch_url, url) for url in urls]
        for future in concurrent.futures.as_completed(futures):
            try:
                data = future.result()
                # Process data
            except Exception as e:
                print(f"Error: {e}")

if __name__ == "__main__":
    main()

（二）设置合理的请求间隔

为了避免因请求频率过快导致的 503 错误，可以在请求之间设置合理的间隔时间。这可以通过在请求代码中添加 time.sleep() 来实现。

import time
import requests

def fetch_url(url):
    try:
        response = requests.get(url)
        response.raise_for_status()
        return response.text
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 503:
            print(f"503 error occurred for {url}")
            # Handle 503 error
        else:
            raise

def main():
    urls = ["http://example.com/page1", "http://example.com/page2", ...]
    for url in urls:
        fetch_url(url)
        time.sleep(1)  # 设置请求间隔为 1 秒

if __name__ == "__main__":
    main()

（三）使用代理服务器和用户代理

使用代理服务器可以隐藏爬虫的真实 IP 地址，减少被服务器封禁的风险。同时，代理服务器可以分散请求，降低单个 IP 的请求频率。服务器可能会根据请求的用户代理（User-Agent）来判断请求是否来自爬虫。通过设置随机的用户代理，可以降低被服务器识别为爬虫的风险。

import requests
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 代理配置
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

# 用户代理池
user_agents = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3",
    "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.103 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0.3 Safari/605.1.15",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:88.0) Gecko/20100101 Firefox/88.0"
]

def get_proxy():
    """获取认证代理"""
    return f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}"

def create_session():
    """创建带有重试机制的会话"""
    session = requests.Session()
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

def fetch_url(url):
    """获取URL内容"""
    session = create_session()
    proxy = get_proxy()
    headers = {"User-Agent": random.choice(user_agents)}
    
    try:
        response = session.get(
            url,
            proxies={"http": proxy, "https": proxy},
            headers=headers,
            timeout=10
        )
        response.raise_for_status()
        print(f"成功获取: {url} [状态码: {response.status_code}]")
        return response.text
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 503:
            print(f"503错误: {url} - 服务器暂时不可用")
            # 可以在这里添加重试逻辑或记录到日志
        else:
            print(f"HTTP错误 {e.response.status_code}: {url}")
        raise
    except Exception as e:
        print(f"请求异常: {url} - {str(e)}")
        raise

def main():
    """主函数"""
    urls = [
        "http://example.com/page1",
        "http://example.com/page2",
        "http://example.com/page3"
    ]
    
    for url in urls:
        try:
            fetch_url(url)
            time.sleep(1)  # 请求间隔
        except Exception as e:
            print(f"处理 {url} 时出错: {e}")
            continue

if __name__ == "__main__":
    import random  # 为user_agents随机选择
    main()

（四）重试机制

当遇到 503 错误时，可以设置重试机制，等待一段时间后再次尝试请求。这可以通过 requests 库的 Session 对象和 Retry 类来实现。

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def fetch_url(url):
    session = requests.Session()
    retries = Retry(total=5, backoff_factor=1, status_forcelist=[503])
    session.mount("http://", HTTPAdapter(max_retries=retries))
    try:
        response = session.get(url)
        response.raise_for_status()
        return response.text
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 503:
            print(f"503 error occurred for {url}")
            # Handle 503 error
        else:
            raise

def main():
    urls = ["http://example.com/page1", "http://example.com/page2", ...]
    for url in urls:
        fetch_url(url)

if __name__ == "__main__":
    main()

三、综合实践案例

以下是一个综合运用上述最佳实践的完整代码示例：

import concurrent.futures
import requests
import time
import random
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

user_agents = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3",
    "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.103 Safari/537.36",
    # 添加更多用户代理
]

proxies = ["http://proxy1.example.com:8080", "http://proxy2.example.com:8080", ...]

def fetch_url(url):
    headers = {"User-Agent": random.choice(user_agents)}
    session = requests.Session()
    retries = Retry(total=5, backoff_factor=1, status_forcelist=[503])
    session.mount("http://", HTTPAdapter(max_retries=retries))
    try:
        response = session.get(url, headers=headers, proxies=random.choice(proxies))
        response.raise_for_status()
        return response.text
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 503:
            print(f"503 error occurred for {url}")
            # Handle 503 error
        else:
            raise

def main():
    urls = ["http://example.com/page1", "http://example.com/page2", ...]
    max_workers = 10  # 控制并发线程数量
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(fetch_url, url) for url in urls]
        for future in concurrent.futures.as_completed(futures):
            try:
                data = future.result()
                # Process data
            except Exception as e:
                print(f"Error: {e}")
            time.sleep(1)  # 设置请求间隔为 1 秒

if __name__ == "__main__":
    main()