Python多线程爬虫
精华
小牛编辑
136浏览
2023-03-14
网络爬虫程序是一种 IO 密集型程序,程序中涉及了很多网络 IO 以及本地磁盘 IO 操作,这些都会消耗大量的时间,从而降低程序的执行效率,而 Python 提供的多线程能够在一定程度上提升 IO 密集型程序的执行效率。
线程的具体使用方法如下所示:
在处理线程的过程中要时刻注意线程的同步问题,即多个线程不能操作同一个数据,否则会造成数据的不确定性。通过 threading 模块的 Lock 对象能够保证数据的正确性。
比如,使用多线程将抓取数据写入磁盘文件,此时,就要对执行写入操作的线程加锁,这样才能够避免写入的数据被覆盖。当线程执行完写操作后会主动释放锁,继续让其他线程去获取锁,周而复始,直到所有写操作执行完毕。具体方法如下所示:
由于上述原因,我们需要构建一个多线程共享数据的模型,让所有线程都到该模型中获取数据。queue(队列,先进先出) 模块提供了创建共享数据的队列模型。比如,把所有待爬取的 URL 地址放入队列中,每个线程都到这个队列中去提取 URL。queue 模块的具体使用方法如下:
图1:小米应用商城 抓取下来的数据 demo 如下所示:
刷新网页来重新加载数据,可得知请求头的 URL 地址,如下所示:
如果想学习 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() # 阻塞等待回收线程创建多线程的具体流程:
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()
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
多线程爬虫案例
下面通过多线程方法抓取小米应用商店( https://app.mi.com/)中应用分类一栏,所有类别下的 APP 的名称、所属类别以及下载详情页 URL 。如下图所示:图1:小米应用商城
三国杀,棋牌桌游,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>因此,可以使用 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))运行上述程序后,打开存储文件,其内容如下:
在我们之间-单机版,休闲创意,http://app.mi.com/details?id=com.easybrain.impostor.gtx 粉末游戏,模拟经营,http://app.mi.com/details?id=jp.danball.powdergameviewer.bnn 三国杀,棋牌桌游,http://app.mi.com/details?id=com.bf.sgs.hdexp.mi 腾讯欢乐麻将全集,棋牌桌游,http://app.mi.com/details?id=com.qqgame.happymj 快游戏,休闲创意,http://app.mi.com/details?id=com.h5gamecenter.h2mgc 皇室战争,战争策略,http://app.mi.com/details?id=com.supercell.clashroyale.mi 地铁跑酷,跑酷闯关,http://app.mi.com/details?id=com.kiloo.subwaysurf ... ...