Python多线程爬虫详解

网络爬虫程序是一种 IO 密集型程序，程序中涉及了很多网络 IO 以及本地磁盘 IO 操作，这些都会消耗大量的时间，从而降低程序的执行效率，而 Python 提供的多线程能够在一定程度上提升 IO 密集型程序的执行效率。

如果想学习 Python 多进程、多线程以及 Python GIL 全局解释器锁的相关知识，可参考《Python并发编程教程》。

多线程使用流程

Python 提供了两个支持多线程的模块，分别是 _thread 和 threading。其中 _thread 模块偏底层，它相比于 threading 模块功能有限，因此推荐大家使用 threading 模块。 threading 中不仅包含了 _thread 模块中的所有方法，还提供了一些其他方法，如下所示：

threading.currentThread() 返回当前的线程变量。
threading.enumerate() 返回一个所有正在运行的线程的列表。
threading.activeCount() 返回正在运行的线程数量。

线程的具体使用方法如下所示：

复制纯文本复制

from threading import Thread
#线程创建、启动、回收
t = Thread(target=函数名) # 创建线程对象
t.start() # 创建并启动线程
t.join()  # 阻塞等待回收线程

from threading import Thread
#线程创建、启动、回收
t = Thread(target=函数名) # 创建线程对象
t.start() # 创建并启动线程
t.join()  # 阻塞等待回收线程

创建多线程的具体流程:

复制纯文本复制

t_list = []
for i in range(5):
    t = Thread(target=函数名)
    t_list.append(t)
    t.start()
for t in t_list:
    t.join()

t_list = []
for i in range(5):
    t = Thread(target=函数名)
    t_list.append(t)
    t.start()
for t in t_list:
    t.join()

除了使用该模块外，您也可以使用 Thread 线程类来创建多线程。

在处理线程的过程中要时刻注意线程的同步问题，即多个线程不能操作同一个数据，否则会造成数据的不确定性。通过 threading 模块的 Lock 对象能够保证数据的正确性。

比如，使用多线程将抓取数据写入磁盘文件，此时，就要对执行写入操作的线程加锁，这样才能够避免写入的数据被覆盖。当线程执行完写操作后会主动释放锁，继续让其他线程去获取锁，周而复始，直到所有写操作执行完毕。具体方法如下所示：

复制纯文本复制

from threading import Lock
lock = Lock()
# 获取锁
lock.acquire()
wirter.writerows("线程锁问题解决")
# 释放锁
lock.release()

from threading import Lock
lock = Lock()
# 获取锁
lock.acquire()
wirter.writerows("线程锁问题解决")
# 释放锁
lock.release()

Queue队列模型

对于 Python 多线程而言，由于 GIL 全局解释器锁的存在，同一时刻只允许一个线程占据解释器执行程序，当此线程遇到 IO 操作时就会主动让出解释器，让其他处于等待状态的线程去获取解释器来执行程序，而该线程则回到等待状态，这主要是通过线程的调度机制实现的。

由于上述原因，我们需要构建一个多线程共享数据的模型，让所有线程都到该模型中获取数据。queue（队列，先进先出）模块提供了创建共享数据的队列模型。比如，把所有待爬取的 URL 地址放入队列中，每个线程都到这个队列中去提取 URL。queue 模块的具体使用方法如下：

复制纯文本复制

# 导入模块
from queue import Queue
q = Queue() #创界队列对象
q.put(url) 向队列中添加爬取一个url链接
q.get() # 获取一个url，当队列为空时，阻塞
q.empty() # 判断队列是否为空，True/False

# 导入模块
from queue import Queue
q = Queue() #创界队列对象
q.put(url) 向队列中添加爬取一个url链接
q.get() # 获取一个url，当队列为空时，阻塞
q.empty() # 判断队列是否为空，True/False

多线程爬虫案例

下面通过多线程方法抓取小米应用商店（https://app.mi.com/）中应用分类一栏，所有类别下的 APP 的名称、所属类别以及下载详情页 URL 。如下图所示：

图1：小米应用商城

抓取下来的数据 demo 如下所示：

三国杀,棋牌桌游,http://app.mi.com/details?id=com.bf.sgs.hdexp.mi

1) 案例分析

通过搜索关键字可知这是一个动态网站，因此需要抓包分析。

刷新网页来重新加载数据，可得知请求头的 URL 地址，如下所示：

https://app.mi.com/categotyAllListApi?page=0&categoryId=1&pageSize=30

其中查询参数 pageSize 参数值不变化，page 会随着页码的增加而变化，而类别 Id 通过查看页面元素，如下所示

复制纯文本复制

<ul class="category-list">
<li><a class="current" href="/category/15">游戏</a></li>
<li><a href="/category/5">实用工具</a></li>
<li><a href="/category/27">影音视听</a></li>
<li><a href="/category/2">聊天社交</a></li>
<li><a href="/category/7">图书阅读</a></li>
<li><a href="/category/12">学习教育</a></li>
<li><a href="/category/10">效率办公</a></li>
<li><a href="/category/9">时尚购物</a></li>
<li><a href="/category/4">居家生活</a></li>
<li><a href="/category/3">旅行交通</a></li>
<li><a href="/category/6">摄影摄像</a></li>
<li><a href="/category/14">医疗健康</a></li>
<li><a href="/category/8">体育运动</a></li>
<li><a href="/category/11">新闻资讯</a></li>
<li><a href="/category/13">娱乐消遣</a></li>
<li><a href="/category/1">金融理财</a></li>
</ul>

<ul class="category-list">
<li><a class="current" href="/category/15">游戏</a></li>
<li><a href="/category/5">实用工具</a></li>
<li><a href="/category/27">影音视听</a></li>
<li><a href="/category/2">聊天社交</a></li>
<li><a href="/category/7">图书阅读</a></li>
<li><a href="/category/12">学习教育</a></li>
<li><a href="/category/10">效率办公</a></li>
<li><a href="/category/9">时尚购物</a></li>
<li><a href="/category/4">居家生活</a></li>
<li><a href="/category/3">旅行交通</a></li>
<li><a href="/category/6">摄影摄像</a></li>
<li><a href="/category/14">医疗健康</a></li>
<li><a href="/category/8">体育运动</a></li>
<li><a href="/category/11">新闻资讯</a></li>
<li><a href="/category/13">娱乐消遣</a></li>
<li><a href="/category/1">金融理财</a></li>
</ul>

因此，可以使用 Xpath 表达式匹配 href 属性，从而提取类别 ID 以及类别名称，表达式如下：

基准表达式：xpath_bds = '//ul[@class="category-list"]/li'
提取 id 表达式：typ_id = li.xpath('./a/@href')[0].split('/')[-1]
类型名称：typ_name = li.xpath('./a/text()')[0]

点击开发者工具的 response 选项卡，查看响应数据，如下所示：

{
count: 2000,
data: [
{
appId: 1348407,
displayName: "天气暖暖-关心Ta从关心天气开始",
icon: "http://file.market.xiaomi.com/thumbnail/PNG/l62/AppStore/004ff4467a7eda75641eea8d38ec4d41018433d33",
level1CategoryName: "居家生活",
packageName: "com.xiaowoniu.WarmWeather"
},
{
appId: 1348403,
displayName: "贵斌同城",
icon: "http://file.market.xiaomi.com/thumbnail/PNG/l62/AppStore/0e607ac85ed9742d2ac2ec1094fca3a85170b15c8",
level1CategoryName: "居家生活",
packageName: "com.gbtc.guibintongcheng"
},
...
...

通过上述响应内容，我们可以从中提取出 APP 总数量（count）和 APP （displayName）名称，以及下载详情页的 packageName。由于每页中包含了 30 个 APP，所以总数量（count）可以计算出每个类别共有多少页。

pages = int(count) // 30 + 1

下载详情页的地址是使用 packageName 拼接而成，如下所示：

link = 'http://app.mi.com/details?id=' + app['packageName']

2) 完整程序

完整程序如下所示：

复制纯文本复制

# -*- coding:utf8 -*-
import requests
from threading import Thread
from queue import Queue
import time
from fake_useragent import UserAgent
from lxml import etree
import csv
from threading import Lock
import json
 
class XiaomiSpider(object):
  def __init__(self):
    self.url = 'http://app.mi.com/categotyAllListApi?page={}&categoryId={}&pageSize=30'
    # 存放所有URL地址的队列
    self.q = Queue()
    self.i = 0
    # 存放所有类型id的空列表
    self.id_list = []
    # 打开文件
    self.f = open('XiaomiShangcheng.csv','a',encoding='utf-8')
    self.writer = csv.writer(self.f)
    # 创建锁
    self.lock = Lock()
 
  def get_cateid(self):
    # 请求
    url = 'http://app.mi.com/'
    headers = { 'User-Agent': UserAgent().random}
    html = requests.get(url=url,headers=headers).text
    # 解析
    parse_html = etree.HTML(html)
    xpath_bds = '//ul[@class="category-list"]/li'
    li_list = parse_html.xpath(xpath_bds)
    for li in li_list:
      typ_name = li.xpath('./a/text()')[0]
      typ_id = li.xpath('./a/@href')[0].split('/')[-1]
      # 计算每个类型的页数
      pages = self.get_pages(typ_id)
      #往列表中添加二元组
      self.id_list.append( (typ_id,pages) )
 
    # 入队列
    self.url_in()
 
  # 获取count的值并计算页数
  def get_pages(self,typ_id):
    # 获取count的值，即app总数
    url = self.url.format(0,typ_id)
    html = requests.get(
      url=url,
      headers={'User-Agent':UserAgent().random}
    ).json()
    count = html['count']
    pages = int(count) // 30 + 1
    return pages
 
  # url入队函数，拼接url，并将url加入队列
  def url_in(self):
    for id in self.id_list:
      # id格式：('4',pages)
      for page in range(1,id[1]+1):
        url = self.url.format(page,id[0])
        # 把URL地址入队列
        self.q.put(url)
 
  # 线程事件函数: get() -请求-解析-处理数据,三步骤
  def get_data(self):
    while True:
       # 判断队列不为空则执行，否则终止
      if not self.q.empty():
        url = self.q.get()
        headers = {'User-Agent':UserAgent().random}
        html = requests.get(url=url,headers=headers)
        res_html = html.content.decode(encoding='utf-8')
        html=json.loads(res_html)
        self.parse_html(html)
      else:
        break
  # 解析函数
  def parse_html(self,html):
    # 写入到csv文件
    app_list = []
 
    for app in html['data']:
      # app名称 + 分类 + 详情链接
      name = app['displayName']
      link = 'http://app.mi.com/details?id=' + app['packageName']
      typ_name = app['level1CategoryName']
      # 把每一条数据放到app_list中,并通过writerows()实现多行写入
      app_list.append([name,typ_name,link])
 
      print(name,typ_name)
      self.i += 1
 
    # 向CSV文件中写入数据
    self.lock.acquire()
    self.writer.writerows(app_list)
    self.lock.release()
 
  # 入口函数
  def main(self):
    # URL入队列
    self.get_cateid()
    t_list = []
    # 创建多线程
    for i in range(1):
      t = Thread(target=self.get_data)
      t_list.append(t)
      # 启动线程
      t.start()
 
    for t in t_list:
        # 回收线程   
        t.join()
 
    self.f.close()
    print('数量:',self.i)
 
if __name__ == '__main__':
  start = time.time()
  spider = XiaomiSpider()
  spider.main()
  end = time.time()
  print('执行时间:%.1f' % (end-start))

# -*- coding:utf8 -*-
import requests
from threading import Thread
from queue import Queue
import time
from fake_useragent import UserAgent
from lxml import etree
import csv
from threading import Lock
import json

class XiaomiSpider(object):
  def __init__(self):
    self.url = 'http://app.mi.com/categotyAllListApi?page={}&categoryId={}&pageSize=30'
    # 存放所有URL地址的队列
    self.q = Queue()
    self.i = 0
    # 存放所有类型id的空列表
    self.id_list = []
    # 打开文件
    self.f = open('XiaomiShangcheng.csv','a',encoding='utf-8')
    self.writer = csv.writer(self.f)
    # 创建锁
    self.lock = Lock()

  def get_cateid(self):
    # 请求
    url = 'http://app.mi.com/'
    headers = { 'User-Agent': UserAgent().random}
    html = requests.get(url=url,headers=headers).text
    # 解析
    parse_html = etree.HTML(html)
    xpath_bds = '//ul[@class="category-list"]/li'
    li_list = parse_html.xpath(xpath_bds)
    for li in li_list:
      typ_name = li.xpath('./a/text()')[0]
      typ_id = li.xpath('./a/@href')[0].split('/')[-1]
      # 计算每个类型的页数
      pages = self.get_pages(typ_id)
      #往列表中添加二元组
      self.id_list.append( (typ_id,pages) )

    # 入队列
    self.url_in()

  # 获取count的值并计算页数
  def get_pages(self,typ_id):
    # 获取count的值，即app总数
    url = self.url.format(0,typ_id)
    html = requests.get(
      url=url,
      headers={'User-Agent':UserAgent().random}
    ).json()
    count = html['count']
    pages = int(count) // 30 + 1
    return pages

  # url入队函数，拼接url，并将url加入队列
  def url_in(self):
    for id in self.id_list:
      # id格式：('4',pages)
      for page in range(1,id[1]+1):
        url = self.url.format(page,id[0])
        # 把URL地址入队列
        self.q.put(url)

  # 线程事件函数: get() -请求-解析-处理数据,三步骤
  def get_data(self):
    while True:
       # 判断队列不为空则执行，否则终止
      if not self.q.empty():
        url = self.q.get()
        headers = {'User-Agent':UserAgent().random}
        html = requests.get(url=url,headers=headers)
        res_html = html.content.decode(encoding='utf-8')
        html=json.loads(res_html)
        self.parse_html(html)
      else:
        break
  # 解析函数
  def parse_html(self,html):
    # 写入到csv文件
    app_list = []

    for app in html['data']:
      # app名称 + 分类 + 详情链接
      name = app['displayName']
      link = 'http://app.mi.com/details?id=' + app['packageName']
      typ_name = app['level1CategoryName']
      # 把每一条数据放到app_list中,并通过writerows()实现多行写入
      app_list.append([name,typ_name,link])

      print(name,typ_name)
      self.i += 1

    # 向CSV文件中写入数据
    self.lock.acquire()
    self.writer.writerows(app_list)
    self.lock.release()

  # 入口函数
  def main(self):
    # URL入队列
    self.get_cateid()
    t_list = []
    # 创建多线程
    for i in range(1):
      t = Thread(target=self.get_data)
      t_list.append(t)
      # 启动线程
      t.start()

    for t in t_list:
        # 回收线程   
        t.join()

    self.f.close()
    print('数量:',self.i)

if __name__ == '__main__':
  start = time.time()
  spider = XiaomiSpider()
  spider.main()
  end = time.time()
  print('执行时间:%.1f' % (end-start))

运行上述程序后，打开存储文件，其内容如下：