Python爬虫代理IP配置与多线程轮换实战教程详解

一、代理IP与多线程轮换核心原理

二、前期环境准备

pip install requests  # 网络请求核心库
pip install concurrent.futures  # 多线程并发库
pip install queue  # 线程安全队列，用于存储代理IP

三、完整实战代码实现（多线程代理轮换）

import requests
from queue import Queue
from concurrent.futures import ThreadPoolExecutor, as_completed

# 1. 初始化线程安全代理队列（可通过星空代理API批量获取高质量IP）
proxy_queue = Queue(maxsize=0)

# 模拟从代理服务商获取的高质量代理IP（星空代理海量IP池可动态批量调取）
proxy_list = [
    "http://111.11.11.11:8080",
    "http://222.22.22.22:8080",
    "http://333.33.33.33:8080",
    "http://444.44.44.44:8080"
]

# 将代理IP存入线程安全队列
for proxy in proxy_list:
    proxy_queue.put(proxy)

# 2. 代理IP有效性验证函数
def check_proxy(proxy):
    """验证代理IP是否可用，过滤失效IP"""
    test_url = "https://www.baidu.com"
    try:
        response = requests.get(test_url, proxies={"http": proxy, "https": proxy}, timeout=5)
        if response.status_code == 200:
            return True
    except Exception:
        return False
    return False

# 3. 单线程爬取函数（自动轮换代理）
def spider_crawl(url):
    """单任务爬取，自动获取可用代理，失败自动切换IP"""
    # 从队列获取代理IP
    if proxy_queue.empty():
        return f"任务失败：{url}，无可用代理IP"
    proxy = proxy_queue.get()

    try:
        # 配置代理发起请求
        proxies = {"http": proxy, "https": proxy}
        response = requests.get(url, proxies=proxies, timeout=10)
        # 请求成功，将IP放回队列，循环复用有效IP
        proxy_queue.put(proxy)
        return f"爬取成功：{url}，当前代理IP：{proxy}"
    except Exception as e:
        # 请求失败，剔除失效IP，重新尝试
        return spider_crawl(url)

# 4. 多线程批量爬取主函数
def multi_thread_crawl(url_list, max_workers=8):
    """
    多线程并发爬取
    :param url_list: 待爬取URL列表
    :param max_workers: 最大并发线程数
    """
    # 优先过滤无效代理IP
    valid_proxies = [p for p in proxy_list if check_proxy(p)]
    global proxy_queue
    proxy_queue = Queue()
    for p in valid_proxies:
        proxy_queue.put(p)

    # 线程池并发执行任务
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_task = {executor.submit(spider_crawl, url): url for url in url_list}
        # 遍历任务结果
        for future in as_completed(future_task):
            print(future.result())

# 程序入口
if __name__ == "__main__":
    # 待爬取目标URL列表
    target_urls = [
        "https://www.baidu.com",
        "https://www.sogou.com",
      
    ]
    # 启动8线程并发爬取
    multi_thread_crawl(target_urls, max_workers=8)

四、核心代码逻辑详解

五、实战避坑与优化技巧

六、总结