Trip: 協程Requests實戰，獲取免費代理

本文使用開源協程網路庫 Trip，解決驗證代理時耗時的問題。

簡介

驗證大量的網上代理一直是爬蟲很麻煩的一個工作。 例如我要發出十萬份請求，十個請求一個代理，每個代理驗證的延時五秒鐘。 那麼需要的時間為：

100000 / 10 * 5 = 50000

這就是半天時間了，當然，我們可以開很多的線程。 這裡我們開二十個線程運行這樣的驗證，那麼需要的時間為：

100000 / 10 * 5 / 20 = 2500

2500 秒即四十分鐘多，不包括抓取和處理，仍舊要消耗大量的時間。

而換用了協程框架後，你可以同時驗證大量代理。 即使在一次驗證中僅驗證一百個代理，需要的時間也會變為：

100000 / 10 * 5 / 100 = 500

十萬請求需要的代理在九分鐘內就可以完成驗證，這還僅是一個線程。 如果開多個線程同時處理還可以達到更快的速度。

另外，等待代理返回請求的過程當中完全不影響處理別的內容。 例如，這一個線程在驗證的八分鐘內完全可以繼續充當公眾號伺服器繼續回復消息，完全不會影響。

開始之前

本文基於依賴庫 Trip，請保證更新至最新版本：

python -m pip install trip -U n

另外，本文涉及協程，建議使用之前基本了解協程是什麼，可以寫出一個合格的協程程序。

Trip 兼容 Python 2、3，所以建議使用同樣兼容的 Tornado 進行入門了解。

具體實現

由於 Trip 想要在協程的基礎上充分還原 Requests 的操作，所以就有一個很簡單的轉換方式。

先用 Requests 的寫法寫好，之後將所有的網路連接操作加上yield，就基本沒有問題了。

所以，下面的代碼應該都不難理解。

獲取代理

現在進入正文，首先是一個僅方便測試的代碼段。

由於大家都有自己的獲取代理的方式，網上免費的代理也多如牛毛，我這裡隨便就挑了一個做測試：

import renimport tripnndef get_proxies(number=10):n r = yield trip.get(http://www.89ip.cn/apijk/ +n ?&tqsl=%s&sxa=&sxb=&tta=&ports=&ktip=&cf=1 % number)n p = re.findall(((?:d{1,3}.){3}d{1,3}:d+), r.text)n print(p)nntrip.run(get_proxies) n

這段代碼沒什麼要說的，89ip是我一分鐘前搜來的網站，之後使用正則從中提取代理地址。

驗證代理

代理的驗證用一個很簡單的思路：

• 請求httpbin.org，根據是否成功返回判斷代理是否可用
• 當然實際使用過程當中可以直接請求目標站點
• 清除返回內容為空的代理

所以，這個思路下代碼是這樣的：

from functools import partialnimport tripnndef test_proxy(proxy):n try:n r = yield trip.get(http://httpbin.org/get, timeout=5,n proxies={ http: proxy, https: proxy })n if httpbin.org not in r.text:n raise Exception(Invalid reply)n except Exception as e:n print([%s]: FAIL %s % (proxy, e))n else:n print([%s]: SUCC %s % (proxy, r.elapsed))n raise trip.Return(proxy)nntrip.run(partial(test_proxy, 58.251.249.152:8118)) n

我們這裡用了五秒的延時，五秒內沒有反應的代理就捨棄。 當然你可以根據自己的需求改為十秒、十五秒。

運行後你就會發現該代理完成了驗證，顯示了是否驗證成功。

你可能會問，這個代碼的確跑了五秒鐘，和普通的代碼有什麼區別呢？

區別在於，有了協程，一個線程在這五秒鐘里可以跑成百上千個這段代碼，詳見下面一節。

充分利用這五秒鐘

如果你使用過 Tornado 或者一些其他的協程框架一定很熟悉這個操作：

r = yield [future1, future2, future3] n

沒錯，這樣三個操作就同時跑了起來 （我知道這不嚴謹，但看出這不嚴謹的也不需要我給你誇耀協程的好處了吧）。

所以最簡單的，我們的主程序可以這樣寫：（這段代碼不能單獨跑，注意）

def main():n proxies = yield get_proxies(100)n r = yield [test_proxy(p) for p in proxies]n print(filter(lambda x: x, r)) n

第一句話獲取 ip，第二句話讓他們同時檢測，第三句話過濾結果。

完整的代碼

所以完整的代碼是這樣的（這段可以跑了，請隨意）：

import re, timennimport requests, tripnn@trip.coroutinendef get_proxies(number=10):n r = yield trip.get(http://www.89ip.cn/apijk/ +n ?&tqsl=%s&sxa=&sxb=&tta=&ports=&ktip=&cf=1 % number)n p = re.findall(((?:d{1,3}.){3}d{1,3}:d+), r.text)n raise trip.Return(p)nn@trip.coroutinendef test_proxy(proxy):n try:n r = yield trip.get(http://httpbin.org/get, timeout=5,n proxies={ http: proxy, https: proxy })n if httpbin.org not in r.text:n raise Exception(Invalid reply)n except Exception as e:n passn else:n raise trip.Return(proxy)nndef main():n proxies = yield get_proxies(100)n r = yield [test_proxy(p) for p in proxies]n print(filter(lambda x: x, r))nnstart_time = time.time()ntrip.run(main)nprint(time.time() - start_time)n

你可以運行一下，還是五秒鐘，一百個代理的檢測完成了。

之後的話

這裡用了最簡單的一種主程序的方法，你當然也可以使用更高級些的：

@trip.coroutinendef worker(wl, pl):n while wl:n p = wl.pop()n r = yield test_proxy(p)n if r:n pl.append(r)nndef main():n waiting_list, proxies_list = yield get_proxies(100), []n yield [worker(waiting_list, proxies_list) for i in range(10)]n print(proxies_list)n

這就看你喜歡了。

另外，剩下來的抓取應該就不需要我多說了。 Requests 怎麼抓，這裡也怎麼抓，Session、timeout 什麼的也全都支持。

最後，有什麼我沒講清楚的、想法、建議都可以與我郵件聯繫或者在文末回復。

望本文能夠有所幫助！