python多线程:控制线程数量
背景
前段时间学习了python的多线程爬虫,当时爬取一个图片网站,开启多线程后,并没有限制线程的数量,也就是说,如果下载1000张图片,会一次性开启1000个子线程同时进行下载
现在希望控制线程数量:例如每次只下载5张,当下载完成后再下载另外5张,直至全部完成
查了一些资料,发现在python中,threading 模块有提供 Semaphore类 和 BoundedSemaphore类来限制线程数
详细说明可以看看下面几篇文章,写的很棒:
https://docs.python.org/3.5/library/threading.html?highlight=threading#semaphore-objects
https://www.liujiangblog.com/course/python/79
https://my.oschina.net/u/3524921/blog/920303
https://zhuanlan.zhihu.com/p/34159519
官网给出例子如下:
信号量通常用于保护容量有限的资源,例如数据库服务器。在资源大小固定的任何情况下,都应使用有界信号量。在产生任何工作线程之前,您的主线程将初始化信号量:
maxconnections = 5 # ... pool_sema = BoundedSemaphore(value=maxconnections)
产生后,工作线程在需要连接到服务器时会调用信号量的获取和释放方法:
with pool_sema: conn = connectdb() try: # ... use connection ... finally: conn.close()
改造之前的多线程爬虫
首先贴出原来的代码
# -*- coding:utf-8 -*- import requests from requests.exceptions import RequestException import os, time import re from lxml import etree import threading lock = threading.Lock() def get_html(url): """ 定义一个方法,用于获取一个url页面的响应内容 :param url: 要访问的url :return: 响应内容 """ response = requests.get(url, timeout=10) # print(response.status_code) try: if response.status_code == 200: # print(response.text) return response.text else: return None except RequestException: print("请求失败") # return None def parse_html(html_text): """ 定义一个方法,用于解析页面内容,提取图片url :param html_text: :return:一个页面的图片url集合 """ html = etree.HTML(html_text) if len(html) > 0: img_src = html.xpath("//img[@class='photothumb lazy']/@data-original") # 元素提取方法 # print(img_src) return img_src else: print("解析页面元素失败") def get_image_pages(url): """ 获取所查询图片结果的所有页码 :param url: 查询图片url :return: 总页码数 """ html_text = get_html(url) # 获取搜索url响应内容 # print(html_text) if html_text is not None: html = etree.HTML(html_text) # 生成XPath解析对象 last_page = html.xpath("//div[@class='pages']//a[last()]/@href") # 提取最后一页所在href链接 print(last_page) if last_page: max_page = re.compile(r'(\d+)', re.S).search(last_page[0]).group() # 使用正则表达式提取链接中的页码数字 print(max_page) print(type(max_page)) return int(max_page) # 将字符串页码转为整数并返回 else: print("暂无数据") return None else: print("查询结果失败") def get_all_image_url(page_number): """ 获取所有图片的下载url :param page_number: 爬取页码 :return: 所有图片url的集合 """ base_url = 'https://imgbin.com/free-png/naruto/' image_urls = [] x = 1 # 定义一个标识,用于给每个图片url编号,从1递增 for i in range(1, page_number): url = base_url + str(i) # 根据页码遍历请求url try: html = get_html(url) # 解析每个页面的内容 if html: data = parse_html(html) # 提取页面中的图片url # print(data) # time.sleep(3) if data: for j in data: image_urls.append({ 'name': x, 'value': j }) x += 1 # 每提取一个图片url,标识x增加1 except RequestException as f: print("遇到错误:", f) continue # print(image_urls) return image_urls def get_image_content(url): """请求图片url,返回二进制内容""" # print("正在下载", url) try: r = requests.get(url, timeout=15) if r.status_code == 200: return r.content return None except RequestException: return None def main(url, image_name): """ 主函数:实现下载图片功能 :param url: 图片url :param image_name: 图片名称 :return: """ semaphore.acquire() # 加锁,限制线程数 print('当前子线程: {}'.format(threading.current_thread().name)) save_path = os.path.dirname(os.path.abspath('.')) + '/pics/' try: file_path = '{0}/{1}.jpg'.format(save_path, image_name) if not os.path.exists(file_path): # 判断是否存在文件,不存在则爬取 with open(file_path, 'wb') as f: f.write(get_image_content(url)) f.close() print('第{}个文件保存成功'.format(image_name)) else: print("第{}个文件已存在".format(image_name)) semaphore.release() # 解锁imgbin-多线程-重写run方法.py except FileNotFoundError as f: print("第{}个文件下载时遇到错误,url为:{}:".format(image_name, url)) print("报错:", f) raise except TypeError as e: print("第{}个文件下载时遇到错误,url为:{}:".format(image_name, url)) print("报错:", e) class MyThread(threading.Thread): """继承Thread类重写run方法创建新进程""" def __init__(self, func, args): """ :param func: run方法中要调用的函数名 :param args: func函数所需的参数 """ threading.Thread.__init__(self) self.func = func self.args = args def run(self): print('当前子线程: {}'.format(threading.current_thread().name)) self.func(self.args[0], self.args[1]) # 调用func函数 # 因为这里的func函数其实是上述的main()函数,它需要2个参数;args传入的是个参数元组,拆解开来传入 if __name__ == '__main__': start = time.time() print('这是主线程:{}'.format(threading.current_thread().name)) urls = get_all_image_url(5) # 获取所有图片url列表 thread_list = [] # 定义一个列表,向里面追加线程 semaphore = threading.BoundedSemaphore(5) # 或使用Semaphore方法 for t in urls: # print(i) m = MyThread(main, (t["value"], t["name"])) # 调用MyThread类,得到一个实例 thread_list.append(m) for m in thread_list: m.start() # 调用start()方法,开始执行 for m in thread_list: m.join() # 子线程调用join()方法,使主线程等待子线程运行完毕之后才退出 end = time.time() print(end-start) # get_image_pages("https://imgbin.com/free-png/Naruto")
将代码进行改造
1、下面的第8、9行表示调用 threading 的 BoundedSemaphore类,初始化信号量为5,把结果赋给变量 pool_sema
1 if __name__ == '__main__': 2 start = time.time() 3 print('这是主线程:{}'.format(threading.current_thread().name)) 4 5 urls = get_all_image_url(5) # 获取所有图片url列表 6 thread_list = [] # 定义一个列表,向里面追加线程 7 8 max_connections = 5 # 定义最大线程数 9 pool_sema = threading.BoundedSemaphore(max_connections) # 或使用Semaphore方法 10 for t in urls: 11 # print(i) 12 13 m = MyThread(main, (t["value"], t["name"])) # 调用MyThread类,得到一个实例 14 15 thread_list.append(m) 16 17 for m in thread_list: 18 19 m.start() # 调用start()方法,开始执行 20 21 for m in thread_list: 22 m.join() # 子线程调用join()方法,使主线程等待子线程运行完毕之后才退出 23 24 25 end = time.time() 26 print(end-start)
2、修改main()函数
(1)方法一:通过with语句实现,第9行添加 with pool_sema
使用 with 语句来获得一个锁、条件变量或信号量,相当于调用 acquire();离开 with 块后,会自动调用 release()
1 def main(url, image_name): 2 """ 3 主函数:实现下载图片功能 4 :param url: 图片url 5 :param image_name: 图片名称 6 :return: 7 """ 8 9 with pool_sema: 10 print('当前子线程: {}'.format(threading.current_thread().name)) 11 save_path = os.path.dirname(os.path.abspath('.')) + '/pics/' 12 try: 13 file_path = '{0}/{1}.jpg'.format(save_path, image_name) 14 if not os.path.exists(file_path): # 判断是否存在文件,不存在则爬取 15 with open(file_path, 'wb') as f: 16 f.write(get_image_content(url)) 17 f.close() 18 19 print('第{}个文件保存成功'.format(image_name)) 20 21 else: 22 print("第{}个文件已存在".format(image_name)) 23 24 25 26 except FileNotFoundError as f: 27 print("第{}个文件下载时遇到错误,url为:{}:".format(image_name, url)) 28 print("报错:", f) 29 raise 30 31 except TypeError as e: 32 print("第{}个文件下载时遇到错误,url为:{}:".format(image_name, url)) 33 print("报错:", e)
(2)方法二:直接使用 acquire()和 release()
下面的第8行调用 acquire(),第24行调用release()
1 def main(url, image_name): 2 """ 3 主函数:实现下载图片功能 4 :param url: 图片url 5 :param image_name: 图片名称 6 :return: 7 """ 8 pool_sema.acquire() # 加锁,限制线程数 9 # with pool_sema: 10 print('当前子线程: {}'.format(threading.current_thread().name)) 11 save_path = os.path.dirname(os.path.abspath('.')) + '/pics/' 12 try: 13 file_path = '{0}/{1}.jpg'.format(save_path, image_name) 14 if not os.path.exists(file_path): # 判断是否存在文件,不存在则爬取 15 with open(file_path, 'wb') as f: 16 f.write(get_image_content(url)) 17 f.close() 18 19 print('第{}个文件保存成功'.format(image_name)) 20 21 else: 22 print("第{}个文件已存在".format(image_name)) 23 24 pool_sema.release() # 解锁imgbin-多线程-重写run方法.py 25 26 except FileNotFoundError as f: 27 print("第{}个文件下载时遇到错误,url为:{}:".format(image_name, url)) 28 print("报错:", f) 29 raise 30 31 except TypeError as e: 32 print("第{}个文件下载时遇到错误,url为:{}:".format(image_name, url)) 33 print("报错:", e)
最终效果是一样的,每次启用5个线程,完成后再启动下一批
相关文章
- python的引用计数分析(二)
- Python任意参数的数量/预习课python基础
- Python随机数生成方法
- Python多线程之线程创建和终止
- Python 卸载python
- Eclipse Kepler SR2 + Python 3.4 + JDK7+Pydev3.4 搭建 python 开发环境(MAC)
- Python初学者如何系统的学习python————Python入门学习指南--内附学习路径
- python学了真的很有用吗?新手小白有必要学吗
- 零基础自学Python需要多长时间从入门到精通?学python能兼职挣钱吗?怎么挣钱?
- Python分享:python爬虫可以用来做什么?
- 利用Python绘制 3D 体素色温图
- 57 python - 异常
- Python网络编程中的服务器架构(负载均衡、单线程、多线程和同步、异步等)
- Python爬虫系列(三)多线程爬取斗图网站(皮皮虾,我们上车)
- Python编程-基础知识-python项目包和文件的管理以及如何引用相对路径的包和模块
- python做中学(五)多线程的用法
- Python map() 函数
- 《Python算法教程》——1.7 练习题
- 神仙级python入门教程(非常详细),从零基础入门到精通,从看这篇开始!
- Python数据处理Tips数据缺失值的8种常用处理方法和可视化
- Linux环境下 多线程下载 (Python 实现版)
- 【Python分布式服务框架】容器服务 Docker
- python之多线程举例
- Python爬虫练习三-多线程爬虫
- 小学生python游戏编程arcade----坦克大战2
- Unity 工具 之 报错 Jenkins 执行/调用 Python 脚本,报错提示 ‘python‘ 不是内部或外部命令,也不是可运行的程序或批处理文件
- python五十六课——正则表达式(常用函数之findall)
- Python脚本完美解决Linux环境解压.zip文件乱码问题
- 【python百度智能云】:Python — 三种获取__VIEWSTATE、__VIEWSTATEGENERATOR、__EVENTVALIDATION方法。
- 【python 学习】——pycharm终端解释器和Python解释器配置
- 解决kolla ansible docker api client python TypeError: load_config() got an unexpected keyword argument
- Python 分部分项 数据处理2 桂柳
- Python 从字符串中取出日期