多线程爬虫使用代理IP指南

原创

华科云商小徐

发布于 2025-06-04 06:06:41

4100

代码可运行

文章被收录于专栏：小徐学爬虫小徐学爬虫

运行总次数：0

代码可运行

多线程爬虫能有效提高工作效率，如果配合代理IP爬虫效率更上一层楼。作为常年使用爬虫做项目的人来说，选择优质的IP池子尤为重要，之前我讲过如果获取免费的代理ip搭建自己IP池，虽然免费但是IP可用率极低。

在多线程爬虫中使用代理IP可以有效防止IP被封禁，提高爬取效率。以下是我总结的一些思路和代码示例：

核心步骤：

1、获取代理IP池

从免费/付费代理网站或服务商API获取代理IP列表
验证代理有效性（必须步骤）
存储代理到队列（线程安全）

2、设计多线程架构

任务队列：存储待爬URL
代理队列：存储可用代理
工作线程：从任务队列取URL，从代理队列取代理执行请求

3、代理异常处理

捕获代理超时/失效异常
将失效代理移出队列
自动切换新代理重试

Python实现示例（使用`threading`和`requests`）

import threading
import queue
import requests
import time

# 代理IP池（示例，实际应从API获取）
PROXIES = [
    "http://203.0.113.1:8080",
    "http://203.0.113.2:3128",
    "http://203.0.113.3:80"
]

# 待爬URL队列（示例）
URL_QUEUE = queue.Queue()
for i in range(1, 101):
    URL_QUEUE.put(f"https://example.com/data?page={i}")

# 有效代理队列（线程安全）
PROXY_QUEUE = queue.Queue()
for proxy in PROXIES:
    PROXY_QUEUE.put(proxy)

def verify_proxy(proxy):
    """验证代理有效性"""
    try:
        resp = requests.get(
            "https://httpbin.org/ip",
            proxies={"http": proxy, "https": proxy},
            timeout=5
        )
        return resp.status_code == 200
    except:
        return False

def worker():
    """工作线程函数"""
    while not URL_QUEUE.empty():
        url = URL_QUEUE.get()
        
        # 获取有效代理
        proxy = None
        while not PROXY_QUEUE.empty():
            test_proxy = PROXY_QUEUE.get()
            if verify_proxy(test_proxy):
                proxy = test_proxy
                break
        
        if not proxy:
            print("无可用代理！")
            break
            
        try:
            # 使用代理发送请求
            headers = {"User-Agent": "Mozilla/5.0"}
            resp = requests.get(
                url,
                proxies={"http": proxy, "https": proxy},
                headers=headers,
                timeout=10
            )
            
            # 处理响应数据
            if resp.status_code == 200:
                print(f"成功爬取 {url} 使用代理 {proxy}")
                # 解析数据...
            else:
                print(f"状态码异常: {resp.status_code}")
                
            # 归还有效代理
            PROXY_QUEUE.put(proxy)
                
        except (requests.exceptions.ProxyError, 
                requests.exceptions.ConnectTimeout,
                requests.exceptions.ReadTimeout) as e:
            
            print(f"代理 {proxy} 失效: {str(e)}")
            # 不再归还失效代理
            
        except Exception as e:
            print(f"请求异常: {str(e)}")
            PROXY_QUEUE.put(proxy)  # 非代理问题则归还
        finally:
            URL_QUEUE.task_done()

# 创建并启动线程
threads = []
for _ in range(5):  # 创建5个工作线程
    t = threading.Thread(target=worker)
    t.daemon = True
    t.start()
    threads.append(t)

# 等待所有任务完成
URL_QUEUE.join()
print("所有任务完成")

关键优化技巧：

1、代理验证

# 定期验证代理池
def refresh_proxies():
    while True:
        for _ in range(PROXY_QUEUE.qsize()):
            proxy = PROXY_QUEUE.get()
            if verify_proxy(proxy):
                PROXY_QUEUE.put(proxy)
            else:
                print(f"移除失效代理: {proxy}")
        time.sleep(300)  # 每5分钟刷新一次

2、自动重试机制

max_retries = 3
for attempt in range(max_retries):
    try:
        # 请求代码...
        break  # 成功则跳出重试
    except:
        if attempt == max_retries - 1:
            print("重试失败，放弃任务")