一個簡單的多線程Python爬蟲（一）

一個簡單的多線程Python爬蟲

最近想要抓取拉勾網的數據，最開始是使用Scrapy的，但是遇到了下面兩個問題:

1. 前端頁面是用JS模板引擎生成的
2. 接口主要是用POST提交參數的

目前不會處理使用JS模板引擎生成的HTML頁面，用POST的提交參數的話，接口統一，也沒有必要使用Scrapy，所以就萌生了自己寫一個簡單的Python爬蟲的想法。

本文中的部分鏈接可能需要翻牆。

參考資料：

1. http://www.ibm.com/developerworks/aix/library/au-threadingpython/
2. http://stackoverflow.com/questions/10525185/python-threading-how-do-i-lock-a-thread

一個爬蟲的簡單框架

一個簡單的爬蟲框架，主要就是處理網絡請求，Scrapy使用的是Twisted（一個事件驅動網絡框架，以非阻塞的方式對網絡I/O進行異步處理），這里不使用異步處理，等以后再研究這個框架。如果使用的是Python3.4及其以上版本，到可以使用asyncio這個標准庫。

這個簡單的爬蟲使用多線程來處理網絡請求，使用線程來處理URL隊列中的url，然后將url返回的結果保存在另一個隊列中，其它線程在讀取這個隊列中的數據，然后寫到文件中去。

該爬蟲主要用下面幾個部分組成。

1 URL隊列和結果隊列

將將要爬去的url放在一個隊列中，這里使用標准庫Queue。訪問url后的結果保存在結果隊列中

初始化一個URL隊列

from Queue import Queue
urls_queue = Queue()
out_queue = Queue()

2 請求線程

使用多個線程，不停的取URL隊列中的url，並進行處理：

import threading

class ThreadCrawl(threading.Thread):
    def __init__(self, queue, out_queue):
        threading.Thread.__init__(self)
        self.queue = queue
        self.out_queue = out_queue

    def run(self):
        while True:
            item = self.queue.get()
            self.queue.task_down()

下面是部分標准庫Queue的使用方法:

Queue.get([block[, timeout]])
Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available.
Queue.task_done()
Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.

如果隊列為空，線程就會被阻塞，直到隊列不為空。處理隊列中的一條數據后，就需要通知隊列已經處理完該條數據。

處理線程

處理結果隊列中的數據，並保存到文件中。如果使用多個線程的話，必須要給文件加上鎖。

lock = threading.Lock()
f = codecs.open('out.txt', 'w', 'utf8')

當線程需要寫入文件的時候，可以這樣處理：

with lock:
    f.write(something)

程序的執行結果

運行狀態：

抓取結果：

源碼

代碼還不完善，將會持續修改中。

# coding: utf-8
'''
Author mr_zys

Email [e-mail] 
'''

from Queue import Queue
import threading
import urllib2
import time
import json
import codecs
from bs4 import BeautifulSoup

urls_queue = Queue()
data_queue = Queue()
lock = threading.Lock()
f = codecs.open('out.txt', 'w', 'utf8')


class ThreadUrl(threading.Thread):

    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        pass


class ThreadCrawl(threading.Thread):

    def __init__(self, url, queue, out_queue):
        threading.Thread.__init__(self)
        self.url = url
        self.queue = queue
        self.out_queue = out_queue

    def run(self):
        while True:
            item = self.queue.get()
            data = self._data_post(item)
            try:
                req = urllib2.Request(url=self.url, data=data)
                res = urllib2.urlopen(req)
            except urllib2.HTTPError, e:
                raise e.reason
            py_data = json.loads(res.read())
            res.close()
            item['first'] = 'false'
            item['pn'] = item['pn'] + 1
            success = py_data['success']
            if success:
                print 'Get success...'
            else:
                print 'Get fail....'
            print 'pn is : %s' % item['pn']
            result = py_data['content']['result']
            if len(result) != 0:
                self.queue.put(item)
            print 'now queue size is: %d' % self.queue.qsize()
            self.out_queue.put(py_data['content']['result'])
            self.queue.task_done()

    def _data_post(self, item):
        pn = item['pn']
        first = 'false'
        if pn == 1:
            first = 'true'
        return 'first=' + first + '&pn=' + str(pn) + '&kd=' + item['kd']

    def _item_queue(self):
        pass


class ThreadWrite(threading.Thread):

    def __init__(self, queue, lock, f):
        threading.Thread.__init__(self)
        self.queue = queue
        self.lock = lock
        self.f = f

    def run(self):
        while True:
            item = self.queue.get()
            self._parse_data(item)
            self.queue.task_done()

    def _parse_data(self, item):
        for i in item:
            l = self._item_to_str(i)
            with self.lock:
                print 'write %s' % l
                self.f.write(l)

    def _item_to_str(self, item):
        positionName = item['positionName']
        positionType = item['positionType']
        workYear = item['workYear']
        education = item['education']
        jobNature = item['jobNature']
        companyName = item['companyName']
        companyLogo = item['companyLogo']
        industryField = item['industryField']
        financeStage = item['financeStage']
        companyShortName = item['companyShortName']
        city = item['city']
        salary = item['salary']
        positionFirstType = item['positionFirstType']
        createTime = item['createTime']
        positionId = item['positionId']
        return positionName + ' ' + positionType + ' ' + workYear + ' ' + education + ' ' + \
            jobNature + ' ' + companyLogo + ' ' + industryField + ' ' + financeStage + ' ' + \
            companyShortName + ' ' + city + ' ' + salary + ' ' + positionFirstType + ' ' + \
            createTime + ' ' + str(positionId) + '\n'


def main():
    for i in range(4):
        t = ThreadCrawl(
            'http://www.lagou.com/jobs/positionAjax.json', urls_queue, data_queue)
        t.setDaemon(True)
        t.start()
    datas = [
        {'first': 'true', 'pn': 1, 'kd': 'Java'}
        #{'first': 'true', 'pn': 1, 'kd': 'Python'}
    ]
    for d in datas:
        urls_queue.put(d)
    for i in range(4):
        t = ThreadWrite(data_queue, lock, f)
        t.setDaemon(True)
        t.start()

    urls_queue.join()
    data_queue.join()

    with lock:
        f.close()
    print 'data_queue siez: %d' % data_queue.qsize()
main()

總結

主要是熟悉使用Python的多線程編程，以及一些標准庫的使用Queue、threading。