Python多线程检测代理IP可用性实例解析

引言

在网络爬虫或数据采集过程中，代理IP的可用性直接影响任务效率。手动验证大量代理IP耗时且不现实，因此编写自动化检测脚本十分必要。本文提供一个基于Python多线程的代理IP可用性检测实例，帮助读者快速筛选出有效代理。

测试样本

首先准备一个包含代理IP的文本文件（例如ip.txt），每行格式为IP:端口@协议#匿名度 地区，示例内容如下：

110.52.235.87:9999@HTTP#[高匿]湖南省岳阳市 联通
111.43.70.58:51547@HTTP#[未知]黑龙江省 移动(全省通用)
183.196.97.125:41397@HTTP#[未知]河北省廊坊市 移动
110.52.235.210:9999@HTTP#[高匿]湖南省岳阳市 联通
183.166.167.163:8080@HTTP#[未知]安徽省黄山市 电信
111.177.171.242:9999@HTTP#[未知]湖北省随州市 电信
123.127.93.188:44399@HTTP#........

Python脚本实现

脚本依赖requests库，请先安装：

pip install requests

完整代码如下：

# coding=utf-8
import os
import re
import requests
import threading
from datetime import datetime
from requests import RequestException
from time import sleep

BASE_DIR = os.path.dirname(__file__)
IP_TXT = os.path.join(BASE_DIR, datetime.now().strftime('%Y-%m-%d') + '-ip.txt')
SCR_IP_TXT = os.path.join(BASE_DIR, 'ip.txt')
MAX_TEST_THREADS = 100

class TestThread(threading.Thread):
    def __init__(self, _ip_li):
        self.li = _ip_li
        super(TestThread, self).__init__()

    def run(self):
        self.li = [_ for _ in self.li if self.test_ip_available(_)]

    def test_ip_available(self, ip):
        _proxy = {'https': ip}
        try:
            print u'正在检测IP: %s 有效性n' % ip
            r = requests.get('https://www.so.com/s?ie=utf-8&fr=none&src=360sou_newhome&q=123',
                              proxies=_proxy,
                             timeout=5)
            assert u'_360搜索' in r.text
        except (RequestException, AssertionError):
            return False
        print u'找到可用代理IP: %sn' % ip
        return True

def time_wrapper(func):
    def _wrapper():
        start_time = datetime.now()
        func()
        end_time = datetime.now()
        seconds = (end_time - start_time).total_seconds()
        print u'本次执行共消耗: %d分%d秒n' % (seconds / 60, seconds % 60)
    return _wrapper

@time_wrapper
def parse_ip():
    # 读取本地ip文件
    with open(SCR_IP_TXT) as f:
        _ip_list = re.findall(r'd+.d+.d+.d+:d+', f.read())
    print u'检索到: %d 个IP地址n' % len(_ip_list)

    # 平均分配_ip_list到各个线程检测
    threads = []
    avg = len(_ip_list) / MAX_TEST_THREADS
    if len(_ip_list) % MAX_TEST_THREADS != 0:
        avg += 1
    for i in range(MAX_TEST_THREADS):
        _thread = TestThread(_ip_list[i*avg:(i+1)*avg])
        threads.append(_thread)
        _thread.start()

    # 等待所有检测线程退出
    while threading.active_count() > 1:
        sleep(10)

    # 读取所有有效IP并写入文件
    _ip_list = []
    for th in threads:
        _ip_list.extend(th.li)
    _ip_list = set(_ip_list)
    print u'n总共找到 %d 个可用IPn' % len(_ip_list)
    with open(IP_TXT, 'w') as f:
        f.write('n'.join(_ip_list))

if __name__ == '__main__':
    parse_ip()

运行结果

执行脚本后，控制台输出类似如下信息：

检索到: 1899 个IP地址
..................
正在检测IP: 124.81.245.148:8080 有效性
正在检测IP: 111.177.160.17:9999 有效性
总共找到 60 个可用IP
本次执行共消耗: 2分13秒

总结

以上实例展示了如何利用Python多线程技术批量检测代理IP的可用性。通过合理分配线程数，可以显著提升检测效率。读者可根据实际需求调整MAX_TEST_THREADS参数和测试URL，以适配不同场景。希望本文能为大家提供参考。