路漫漫其修遠兮，吾將上下而求索

0%

爬蟲實戰 - Using Python3 Part7

Posted on Edited on Disqus:

Thread 爬蟲優化

前陣子介紹了如何把 Thread 應用在爬蟲上，今天要來介紹如何因應硬體資源來調整 Thread 的執行數量。

在上一篇教學中，筆者把每頁的查詢都新增一個 Thread 去處理，假設有100頁的話便需要100條 Thread ，若電腦硬體不夠強根本無法這樣處理，所以需要控管單一時間內允許執行的 Thread 數量。

想法如下：

  1. 事先宣告好所有的爬蟲任務放進陣列內
  2. 設定 Thread 可運行的數量
  3. Thread 檢查是否有爬蟲任務，若有則執行任務；若無則結束離開
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def startCrawler(*args):
    queue = args[0]
    while queue.qsize() > 0:
        job = queue.get()
        job.do()

def main():
    que = queue.Queue()
    for i in range(1, 13):
        que.put(axePage(url))
    for i in range(THREADLIMIT):
        worker = threading.Thread(target=startCrawler, args=(que,))
        worker.start()

if __name__ == "__main__":
    main()

threading.Thread(target=startCrawler, args=(que,)) 這行可能比較弔詭，他的意思是宣告一條Thread 執行 startCrawler function，並且傳遞 que 參數給他，所以在 startCrawler 便可以讀取任務清單，當有任務的話自然就交給 Thread 去執行。
想知道更多使用方法的話可以參考python 官方的教學

有別於上一篇 Thread 的介紹，我們這次不把 Thread 宣告在每個單一執行的任務內，而是宣告特定數量的 Thread 去讀取爬蟲的任務

接下來我們實作axePage class 這個類別：

1
2
3
4
5
6
7
class axePage:
    def __init__(self, url):
        self.url = url
    def do(self):
        result = requests.get(self.url)
        result.encoding = 'utf-8'
        self.jsondata = parseUrlToJson(result.text)

與前一個版本的class 比起來真的精簡不少

最後把所有code 組合起來如下：

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
# -*- coding: utf8 -*-
from lxml import etree, html
import requests, threading, queue

class axePage:
    def __init__(self, url):
        self.url = url
    def do(self):
        result = requests.get(self.url)
        result.encoding = 'utf-8'
        self.jsondata = parseUrlToJson(result.text)

def parseUrlToJson(content):
    root = etree.fromstring(content, etree.HTMLParser())
    jsondata = ""
    rows = root.xpath("//table[@class='table']/tr[position()>1]")
    for row in rows:
        column = row.xpath("./td/text()")
        jsondata += '{"town": "%s", "village": "%s", "name": "%s"},' % (column[0], column[1], column[2])
    return jsondata 

def startCrawler(*args):
    queue = args[0]
    while queue.qsize() > 0:
        job = queue.get()
        job.do()

def main():
    workerList = []
    que = queue.Queue()
    THREADLIMIT = 3
    for i in range(1, 13):
        que.put(axePage("http://axe-level-1.herokuapp.com/lv2/?page=%d" % i))

    for i in range(THREADLIMIT):
        worker = threading.Thread(target=startCrawler, args=(que,))
        worker.start()
        workerList.append(worker)

    for i in range(THREADLIMIT):
        workerList[i].join()
    print('done')

if __name__ == "__main__":
    main()

這邊就不另外把資料輸出，僅提供另一種形式的 Thread 寫法給大家參考