Python多线程中线程数量如何控制
作者:Python热爱者 发布时间:2021-01-12 04:05:05
标签: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个线程,完成后再启动下一批
来源:https://blog.csdn.net/qdPython/article/details/121907420
0
投稿
猜你喜欢
- python代码生成API接口如果要将我们写好的Python代码生成API接口时,我们需要借助Flask框架1. 安装Flaskpip in
- 问题: 1. 后台管理员只有一个用户: admin, 密码: admin 2. 当管理员登陆成功后, 可以管理前台会员信
- 用下列代码可在控制弹出NTLM验证窗口:<% Response.Status = "401 Unauthorized&quo
- 使用 Beanstalkd 作为消息队列服务,然后结合 Python 的装饰器语法实现一个简单的异步任务处理工具.最终效果定义任务:from
- xlabel(‘time',‘FontSize',12);如果没有设置位置,默认是在中间在xlabel中也有position
- 目录一.定义二.命名方法2.1小驼峰命名法2.2大驼峰命名法2.3下划线命名法三.命名规则3.1标识符3.2关键字四.使用方法4.1单变量赋
- -----最近从github上找了一个代码跑,但是cpu训练的时间实在是太长,所以想用gpu训练一下,经过了一天的折腾终于可以用gpu进行训
- 前言:不用改掉系统python2.7 ,原来是python2.7,我们还进行python2.7的保留1.编译前准备其他库的安装(使用sudo
- 用于逐行分析文本的代码示例fileIN = open(sys.argv[1], "r")line = fileIN.re
- 说明本文根据https://github.com/liuchengxu/blockchain-tutorial的内容,用python实现的,
- 在Oracle 8i中,往往会出现要在存储过程中运行操作系统命令的情况。一般来说,利用Oracle Enterprise Manager设定
- python中format的使用format函数这是一种字符串格式化的方法,用法如str.format()。基本语法是通过 {} 和 : 来
- 对于请求一些网站,我们需要加上请求头才可以完成网页的抓取,不然会得到一些错误,无法返回抓取的网页。下面,介绍两种添加请求头的方法。方法一:借
- 1.C语言实现1.1代码说明a 创建双向链表:在创建哈夫曼树的过程中,需要不断对结点进行更改和删除,所以选用双向链表的结构更容易
- python字典中,值可任意更改;但键是唯一的,不支持直接修改。若真的需要修改字典中的键,可通过几种间接方式实现。新建空白字典。info =
- 前言在一个分布式环境中,每台机器上可能需要启动和停止多个进程,使用命令行方式一个一个手动启动和停止非常麻烦,而且查看每个进程的状态也很不方便
- 切片从list或tuple中取部分元素。list = [1, 2, 3, 4]list[0 : 3] # [1, 2, 3]list[-2
- 如下所示:import time首先导入时间模块在程序开始执行的地方写入:start=time.clock()在程序末尾写入:end=tim
- [概 要] 这篇文章讨论常用的"sql注入"技术的细节,应用于流行的Ms IIS/ASP/SQL-Server平台。这里
- 本文为大家分享了python实现学生管理系统的具体代码,供大家参考,具体内容如下1.0版本学生管理系统''' 1.添