本文使用开源协程网络库 Trip，解决验证代理时耗时的问题。
验证大量的网上代理一直是爬虫很麻烦的一个工作。 例如我要发出十万份请求，十个请求一个代理，每个代理验证的延时五秒钟。 那么需要的时间为：
100000 / 10 * 5 = 50000
这就是半天时间了，当然，我们可以开很多的线程。 这里我们开二十个线程运行这样的验证，那么需要的时间为：
100000 / 10 * 5 / 20 = 2500
2500 秒即四十分钟多，不包括抓取和处理，仍旧要消耗大量的时间。
而换用了协程框架后，你可以同时验证大量代理。 即使在一次验证中仅验证一百个代理，需要的时间也会变为：
100000 / 10 * 5 / 100 = 500
十万请求需要的代理在九分钟内就可以完成验证，这还仅是一个线程。 如果开多个线程同时处理还可以达到更快的速度。
另外，等待代理返回请求的过程当中完全不影响处理别的内容。 例如，这一个线程在验证的八分钟内完全可以继续充当公众号服务器继续回复消息，完全不会影响。
本文基于依赖库 Trip，请保证更新至最新版本：
python -m pip install trip -U
另外，本文涉及协程，建议使用之前基本了解协程是什么，可以写出一个合格的协程程序。
Trip 兼容 Python 2、3，所以建议使用同样兼容的 Tornado 进行入门了解。
由于 Trip 想要在协程的基础上充分还原 Requests 的操作，所以就有一个很简单的转换方式。
先用 Requests 的写法写好，之后将所有的网络连接操作加上yield，就基本没有问题了。
所以，下面的代码应该都不难理解。
现在进入正文，首先是一个仅方便测试的代码段。
由于大家都有自己的获取代理的方式，网上免费的代理也多如牛毛，我这里随便就挑了一个做测试：
import re
import trip
def get_proxies(number=10):
r = yield trip.get('http://www.89ip.cn/apijk/' +
'?&tqsl=%s&sxa=&sxb=&tta=&ports=&ktip=&cf=1' % number)
p = re.findall('((?:\d{1,3}.){3}\d{1,3}:\d+)', r.text)
print(p)
trip.run(get_proxies)
这段代码没什么要说的，89ip是我一分钟前搜来的网站，之后使用正则从中提取代理地址。
代理的验证用一个很简单的思路：
httpbin.org，根据是否成功返回判断代理是否可用所以，这个思路下代码是这样的：
from functools import partial
import trip
def test_proxy(proxy):
try:
r = yield trip.get('http://httpbin.org/get', timeout=5,
proxies={ 'http': proxy, 'https': proxy })
if 'httpbin.org' not in r.text:
raise Exception('Invalid reply')
except Exception as e:
print('[%s]: FAIL %s' % (proxy, e))
else:
print('[%s]: SUCC %s' % (proxy, r.elapsed))
raise trip.Return(proxy)
trip.run(partial(test_proxy, '58.251.249.152:8118'))
我们这里用了五秒的延时，五秒内没有反应的代理就舍弃。 当然你可以根据自己的需求改为十秒、十五秒。
运行后你就会发现该代理完成了验证，显示了是否验证成功。
你可能会问，这个代码的确跑了五秒钟，和普通的代码有什么区别呢？
区别在于，有了协程，一个线程在这五秒钟里可以跑成百上千个这段代码，详见下面一节。
如果你使用过 Tornado 或者一些其他的协程框架一定很熟悉这个操作：
r = yield [future1, future2, future3]
没错，这样三个操作就同时跑了起来 （我知道这不严谨，但看出这不严谨的也不需要我给你夸耀协程的好处了吧）。
所以最简单的，我们的主程序可以这样写：（这段代码不能单独跑，注意）
def main():
proxies = yield get_proxies(100)
r = yield [test_proxy(p) for p in proxies]
print(filter(lambda x: x, r))
第一句话获取 ip，第二句话让他们同时检测，第三句话过滤结果。
所以完整的代码是这样的（这段可以跑了，请随意）：
import re, time
import requests, trip
@trip.coroutine
def get_proxies(number=10):
r = yield trip.get('http://www.89ip.cn/apijk/' +
'?&tqsl=%s&sxa=&sxb=&tta=&ports=&ktip=&cf=1' % number)
p = re.findall('((?:\d{1,3}.){3}\d{1,3}:\d+)', r.text)
raise trip.Return(p)
@trip.coroutine
def test_proxy(proxy):
try:
r = yield trip.get('http://httpbin.org/get', timeout=5,
proxies={ 'http': proxy, 'https': proxy })
if 'httpbin.org' not in r.text:
raise Exception('Invalid reply')
except Exception as e:
pass
else:
raise trip.Return(proxy)
def main():
proxies = yield get_proxies(100)
r = yield [test_proxy(p) for p in proxies]
print(filter(lambda x: x, r))
start_time = time.time()
trip.run(main)
print(time.time() - start_time)
你可以运行一下，还是五秒钟，一百个代理的检测完成了。
这里用了最简单的一种主程序的方法，你当然也可以使用更高级些的：
@trip.coroutine
def worker(wl, pl):
while wl:
p = wl.pop()
r = yield test_proxy(p)
if r:
pl.append(r)
def main():
waiting_list, proxies_list = yield get_proxies(100), []
yield [worker(waiting_list, proxies_list) for i in range(10)]
print(proxies_list)
这就看你喜欢了。
你如果使用的是 Windows 系统，你可能会发现没法同时进行很多的请求。 这是很正常的问题，Windows 系统使用的是 select，没法同时处理很多请求。
另外，剩下来的抓取应该就不需要我多说了。 Requests 怎么抓，这里也怎么抓，Session、timeout 什么的也全都支持。
最后，有什么我没讲清楚的、想法、建议都可以与我邮件联系或者在文末回复。
望本文能够有所帮助！
这是一个专为移动设备优化的页面（即为了让你能够在 Google 搜索结果里秒开这个页面），如果你希望参与 V2EX 社区的讨论，你可以继续到 V2EX 上打开本讨论主题的完整版本。
V2EX 是创意工作者们的社区，是一个分享自己正在做的有趣事物、交流想法，可以遇见新朋友甚至新机会的地方。
V2EX is a community of developers, designers and creative people.