多线程爬虫实战：糗事百科

参考上一节课糗事百科单进程项目

python下多线程的思考

Queue是python中的标准库，可以直接import Queue引用;队列是线程间最常用的交换数据的形式

对于资源，加锁是个重要的环节。因为python原生的list,dict等，都是not thread safe的。而Queue，是线程安全的，因此在满足使用条件下，建议使用队列

Python Queue模块有三种队列及构造函数:

1、Python Queue模块的FIFO队列先进先出。 class Queue.Queue(maxsize)

2、LIFO类似于堆，即先进后出。 class Queue.LifoQueue(maxsize)

3、还有一种是优先级队列级别越低越先出来。 class Queue.PriorityQueue(maxsize)

Queue（队列对象）

  • 初始化： class Queue.Queue(maxsize) FIFO 先进先出

  • 包中的常用方法:

    Queue.qsize() 返回队列的大小

    Queue.empty() 如果队列为空，返回True,反之False

    Queue.full() 如果队列满了，返回True,反之False

    Queue.full 与 maxsize 大小对应

    Queue.get([block[, timeout]])获取队列，timeout等待时间 

  1.   调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。
  3.   如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。
  5.   如果队列为空且block为False，队列将引发Empty异常。

  • 创建一个“队列”对象

    import Queue myqueue = Queue.Queue(maxsize = 10)

  • 将一个值放入队列中

    myqueue.put(10)

  • 将一个值从队列中取出

    myqueue.get()

多线程项目案例

多线程示意图

多线程爬虫实战：糗事百科 - 图1

  1. # -*- coding:utf-8 -*-
  2. import requests
  3. from lxml import etree
  4. from Queue import Queue
  5. import threading
  6. import time
  7. import json
  10. class thread_crawl(threading.Thread):
  11.     '''
  12.     抓取线程类
  13.     '''
  15.     def __init__(self, threadID, q):
  16.         threading.Thread.__init__(self)
  17.         self.threadID = threadID
  18.         self.q = q
  20.     def run(self):
  21.         print "Starting " + self.threadID
  22.         self.qiushi_spider()
  23.         print "Exiting ", self.threadID
  25.     def qiushi_spider(self):
  26.         # page = 1
  27.         while True:
  28.             if self.q.empty():
  29.                 break
  30.             else:
  31.                 page = self.q.get()
  32.                 print 'qiushi_spider=', self.threadID, ',page=', str(page)
  33.                 url = 'http://www.qiushibaike.com/hot/page/' + str(page) + '/'
  34.                 headers = {
  35.                     'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36',
  36.                     'Accept-Language': 'zh-CN,zh;q=0.8'}
  37.                 # 多次尝试失败结束、防止死循环
  38.                 timeout = 4
  39.                 while timeout > 0:
  40.                     timeout -= 1
  41.                     try:
  42.                         content = requests.get(url, headers=headers)
  43.                         data_queue.put(content.text)
  44.                         break
  45.                     except Exception, e:
  46.                         print 'qiushi_spider', e
  47.                 if timeout < 0:
  48.                     print 'timeout', url
  51. class Thread_Parser(threading.Thread):
  52.     '''
  53.     页面解析类；
  54.     '''
  56.     def __init__(self, threadID, queue, lock, f):
  57.         threading.Thread.__init__(self)
  58.         self.threadID = threadID
  59.         self.queue = queue
  60.         self.lock = lock
  61.         self.f = f
  63.     def run(self):
  64.         print 'starting ', self.threadID
  65.         global total, exitFlag_Parser
  66.         while not exitFlag_Parser:
  67.             try:
  68.                 '''
  69.                 调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。
  70.                 如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。
  71.                 如果队列为空且block为False，队列将引发Empty异常。
  72.                 '''
  73.                 item = self.queue.get(False)
  74.                 if not item:
  75.                     pass
  76.                 self.parse_data(item)
  77.                 self.queue.task_done()
  78.                 print 'Thread_Parser=', self.threadID, ',total=', total
  79.             except:
  80.                 pass
  81.         print 'Exiting ', self.threadID
  83.     def parse_data(self, item):
  84.         '''
  85.         解析网页函数
  86.         :param item: 网页内容
  87.         :return:
  88.         '''
  89.         global total
  90.         try:
  91.             html = etree.HTML(item)
  92.             result = html.xpath('//div[contains(@id,"qiushi_tag")]')
  93.             for site in result:
  94.                 try:
  95.                     imgUrl = site.xpath('.//img/@src')[0]
  96.                     title = site.xpath('.//h2')[0].text
  97.                     content = site.xpath('.//div[@class="content"]')[0].text.strip()
  98.                     vote = None
  99.                     comments = None
  100.                     try:
  101.                         vote = site.xpath('.//i')[0].text
  102.                         comments = site.xpath('.//i')[1].text
  103.                     except:
  104.                         pass
  105.                     result = {
  106.                         'imgUrl': imgUrl,
  107.                         'title': title,
  108.                         'content': content,
  109.                         'vote': vote,
  110.                         'comments': comments,
  111.                     }
  113.                     with self.lock:
  114.                         # print 'write %s' % json.dumps(result)
  115.                         self.f.write(json.dumps(result, ensure_ascii=False).encode('utf-8') + "\n")
  117.                 except Exception, e:
  118.                     print 'site in result', e
  119.         except Exception, e:
  120.             print 'parse_data', e
  121.         with self.lock:
  122.             total += 1
  124. data_queue = Queue()
  125. exitFlag_Parser = False
  126. lock = threading.Lock()
  127. total = 0
  129. def main():
  130.     output = open('qiushibaike.json', 'a')
  132.     #初始化网页页码page从1-10个页面
  133.     pageQueue = Queue(50)
  134.     for page in range(1, 11):
  135.         pageQueue.put(page)
  137.     #初始化采集线程
  138.     crawlthreads = []
  139.     crawlList = ["crawl-1", "crawl-2", "crawl-3"]
  141.     for threadID in crawlList:
  142.         thread = thread_crawl(threadID, pageQueue)
  143.         thread.start()
  144.         crawlthreads.append(thread)
  146.     #初始化解析线程parserList
  147.     parserthreads = []
  148.     parserList = ["parser-1", "parser-2", "parser-3"]
  149.     #分别启动parserList
  150.     for threadID in parserList:
  151.         thread = Thread_Parser(threadID, data_queue, lock, output)
  152.         thread.start()
  153.         parserthreads.append(thread)
  155.     # 等待队列清空
  156.     while not pageQueue.empty():
  157.         pass
  159.     # 等待所有线程完成
  160.     for t in crawlthreads:
  161.         t.join()
  163.     while not data_queue.empty():
  164.         pass
  165.     # 通知线程是时候退出
  166.     global exitFlag_Parser
  167.     exitFlag_Parser = True
  169.     for t in parserthreads:
  170.         t.join()
  171.     print "Exiting Main Thread"
  172.     with lock:
  173.         output.close()
  176. if __name__ == '__main__':
  177.     main()