Python多线程中线程数量如何控制
前言
前段时间学习了python的多线程爬虫,当时爬取一个图片网站,开启多线程后,并没有限制线程的数量,也就是说,如果下载1000张图片,会一次性开启1000个子线程同时进行下载
现在希望控制线程数量:例如每次只下载5张,当下载完成后再下载另外5张,直至全部完成
查了一些资料,发现在python中,threading 模块有提供 Semaphore类 和 BoundedSemaphore类来限制线程数
官网给出例子如下:
信号量通常用于保护容量有限的资源,例如数据库服务器。在资源大小固定的任何情况下,都应使用有界信号量。在产生任何工作线程之前,您的主线程将初始化信号量:
maxconnections = 5 # ... pool_sema = BoundedSemaphore(value=maxconnections)
产生后,工作线程在需要连接到服务器时会调用信号量的获取和释放方法:
改造之前的多线程爬虫
首先贴出原来的代码
# -*- 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
if __name__ == '__main__': start = time.time() print('这是主线程:{}'.format(threading.current_thread().name)) urls = get_all_image_url(5) # 获取所有图片url列表 thread_list = [] # 定义一个列表,向里面追加线程 # 更多Python相关视频、资料加群778463939免费获取 max_connections = 5 # 定义最大线程数 pool_sema = threading.BoundedSemaphore(max_connections) # 或使用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)
2、修改main()函数
(1)方法一:通过with语句实现,第9行添加 with pool_sema
使用 with 语句来获得一个锁、条件变量或信号量,相当于调用 acquire();离开 with 块后,会自动调用 release()
def main(url, image_name): """ 主函数:实现下载图片功能 :param url: 图片url :param image_name: 图片名称 :return: """ with pool_sema: 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)) 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)
(2)方法二:直接使用 acquire()和 release()
下面的第8行调用 acquire(),第24行调用release()
def main(url, image_name): """ 主函数:实现下载图片功能 :param url: 图片url :param image_name: 图片名称 :return: """ pool_sema.acquire() # 加锁,限制线程数 # with pool_sema: 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)) pool_sema.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)
最终效果是一样的,每次启用5个线程,完成后再启动下一批
关于Python多线程中线程数量如何控制的文章就介绍至此,更多相关Python 线程数量控制内容请搜索编程宝库以前的文章,希望以后支持编程宝库!
1、基本原理访问网站扫码登录页,网站给浏览器返回一个二维码和一个唯一标志KEY浏览器开启定时轮询服务器,确认KEY对应的扫码结果用户使用app扫码二维码,app获取KEY并告知服务器已扫码 ...