爬虫代理ip池创建【使用redis TTL实现】

什么是ip代理

IP代理是一种通过中间服务器或计算机来代理网络请求的方法，它允许你在访问互联网资源时隐藏你的真实IP地址并使用代理服务器的IP地址。通常，代理服务器充当客户端和目标服务器之间的中间层，负责转发请求和响应。

IP代理池是一个存储和管理多个代理IP地址的系统或工具，通常由一组代理IP和相关的功能组成，可用在网络爬虫、数据采集等场景中突破ip限制，优化爬虫速度以及提高爬虫稳定性。

需求分析与解决方案设计

刚开始没有添加代理时：爬虫运行过程中，爬虫的请求速度一直维持在每秒10条左右，无论如何改变并发数，速度一直维持在10以内，在排查了网络，队列，数据库的状态之后，我将问题定位在了发送爬虫请求的ip被限制。
为了解决这个问题，我需要使用ip代理的方式发送爬虫请求，以求绕过目标网站的限制。我购买了网络上的高匿名ip代理，每次可获取一定量的代理ip,每个代理ip有效时间为30s-60s。为了维持50次请求每秒的速率，我采用代理池的方式管理ip，在请求发出之前从代理池取出随机可用代理，加在请求包内，绕过ip限制。考虑到代理有可用时间限制，以及爬虫运行为一个高并发的场景，我选择使用redis建立代理池，并维持代理池数量在一定的范围内动态波动。

redis TTL

在Redis中，TTL（Time To Live）是一个非常重要的概念，用于设置数据的过期时间。

TTL的作用

用来设置数据过期时间的机制。通过设置TTL，你可以让某个键（key）在一定时间后自动被删除，从而实现数据的自动清理和过期处理。

TTL的通用使用场景

缓存管理：在缓存中存储数据时，可以为每个数据项设置TTL，确保数据在一定时间后自动失效，从而避免缓存中过期的数据占用内存。
会话管理：可以使用TTL来管理用户会话，当用户不再活跃时，自动清除会话数据，实现会话的自动过期。
分布式锁：通过设置带有TTL的键来实现分布式锁，确保锁在一定时间后自动释放，避免死锁。
限流和计数：可以使用TTL来限制请求速率或计算某个事件的时间窗口内的次数。

redis命令设置TTL：

在Redis中，你可以使用EXPIRE或PEXPIRE命令来为一个键设置TTL。EXPIRE以秒为单位设置TTL，而PEXPIRE以毫秒为单位设置TTL。例如：

EXPIRE mykey 3600  # 为键mykey设置TTL为3600秒（1小时）
1

查看TTL：

你可以使用TTL或PTTL命令来查看键的剩余生存时间，分别以秒和毫秒为单位返回。如果键不存在或没有设置TTL，这些命令将返回-1。

TTL mykey  # 查看键mykey的剩余生存时间（秒）
1

移除TTL：

如果需要取消一个键的TTL，可以使用PERSIST命令，它会将键设置为永不过期。例如：

PERSIST mykey  # 取消键mykey的TTL，使其永不过期
1

过期回调：
Redis还提供了一种机制，可以在键过期时执行一个回调函数。这可以通过设置键的TTL并使用SET命令的EX选项来实现。例如：

SET mykey "Hello, Redis!" EX 3600  # 设置键mykey的值，并在1小时后过期
1

当键mykey在1小时后过期时，你可以在设置TTL时指定的时间触发一个回调函数，以执行一些自定义操作。

TTL是Redis中一个非常实用的特性，它允许你更好地管理数据的生命周期，自动清理不再需要的数据，以及实现一些有关数据过期和定时操作的功能。

代理池构建

由于redis的ttl是针对key使用的，而基础的数据结构内无法满足既有超时时间又能随机取出一个ip的方式；我选择使用redis的一个库来当做代理池。
使用随机取出单个key的方式获取ip,并且在添加代理时为代理设置过期时间，如果在过期时间内代理失效，则直接删除这个key。
详细构建方式如下：

1. 建立代理池生成器：
   请求购买的代理，并获取到代理的有效期
   使用redis 的字符串数据结构，设置到redis的库内
2. 建立维护服务：
   设置代理池的最大ip量，每隔一段时间检测代理池数据量，少于IP最大量则调用代理生成器添加，多于则不做操作
3. 建立代理获取服务：
   每次随机获取一个代理，如果在代理使用过程中提前失效，则在代理池中删除该代理

代理池具体实现示例 【python+redis】

class ProxiesGenerator:

    def __init__(self):
        self.redis_conn = redis.StrictRedis.from_url(REDIS_PROXIES_POOL_CONN_STR)
        self.req_proxy_url = '你的代理获取ip'
        # 代理池名称
        self.proxies_pool = 'ip_pools'
        self.proxies = []
        self.count = 0

    # 测试代理可用性
    def proxy_available_test(self, proxy):
        target_url = 'http://example.com/'
        retry_times = 3
        proxies = {
            "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": username, "pwd": password,
                                                            "proxy": proxy.replace('http://', '')},
        }
        for i in range(retry_times):
            try:
                response = requests.get(target_url, proxies=proxies, timeout=3, verify=False)
                # 检查是否成功访问网站
                if response.status_code == 200:
                    # print('响应时间'+"==>"+str(response.elapsed.total_seconds()))
                    if int(response.elapsed.total_seconds())>1:
                        logging.warning(proxy + "不可用")
                        return False
                    else:
                        logging.warning(proxy+"可用")
                        return True
                else:
                    pass
            except Exception as e:
                pass
        return False

    # 添加代理到代理池
    def add_proxies_new(self):
        # 判断有多少个可用的proxies,如果数量小于池标量，则开始添加
        if self.redis_conn.dbsize() <= PROXIES_POOL_LIMITS: # 对比redis池与代理最大限制数
            response = requests.get(url=PROXIES_REQ_URL, verify=False, timeout=2)
            proxies = []
            data = response.text
            for line in data.split('\n'):
                expires = int(line.split(',')[-1].strip()) # 代理可以时长
                ip = line.split(',')[0].split(':')[0].strip()
                port = line.split(',')[0].split(':')[1].strip()
                proxy_string = f'http://{ip}:{port}'
                print(proxy_string)
                print(expires)
                proxies.append([proxy_string,expires])

            with ThreadPoolExecutor(max_workers=200) as executor:
                for data in proxies:
                    executor.submit(self.check_single_avi, data)
        else:
            time.sleep(2)
	
	# 单线程测试代理可用性
    def check_single_avi(self,data):
        proxy_string = data[0]
        expires = int(data[1])-PROXY_DED_REQ_TIME # 代理的请求时间
        # 测试代理的可用性
        if self.proxy_available_test(proxy_string):
            self.redis_conn.setex(proxy_string, expires, proxy_string)
            logging.warning("proxy_string_into_redis"+proxy_string)

    # 获取一个代理
    def get_proxy(self):
        # 先随机获取一个key
        RANDOMKEY = self.redis_conn.randomkey()
        if RANDOMKEY:
            return RANDOMKEY.decode('utf-8')
        else:
            return ""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75