一个简单的多线程Python爬虫（一）

2022-07-14 21:40:35

一个简单的多线程Python爬虫

最近想要抓取拉勾网的数据，最开始是使用Scrapy的，但是遇到了下面两个问题:

前端页面是用JS模板引擎生成的
接口主要是用POST提交参数的

目前不会处理使用JS模板引擎生成的HTML页面，用POST的提交参数的话，接口统一，也没有必要使用Scrapy，所以就萌生了自己写一个简单的Python爬虫的想法。

本文中的部分链接可能需要*。

参考资料：

一个爬虫的简单框架

一个简单的爬虫框架，主要就是处理网络请求，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 myzysv5@sina.com

'''

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。

码农公寓