一招教你控制python多线程的线程数量

大家早好、午好、晚好吖 ❤ ~欢迎光临本文章

如果有什么疑惑/资料需要的可以点击文章末尾名片领取源码

在使用python的多线程爬虫，当时爬取一个图片网站，开启多线程后，并没有限制线程的数量，

也就是说，如果下载1000张图片，会一次性开启1000个子线程同时进行下载

现在希望控制线程数量：例如每次只下载5张，当下载完成后再下载另外5张，直至全部完成

查了一些资料，发现在python中，threading 模块有提供 Semaphore类 和 BoundedSemaphore类来限制线程数

官网给出例子如下：

信号量通常用于保护容量有限的资源，例如数据库服务器。

在资源大小固定的任何情况下，都应使用有界信号量。

在产生任何工作线程之前，您的主线程将初始化信号量：

maxconnections = 5
# ...
pool_sema = BoundedSemaphore(value=maxconnections)
1
2
3

产生后，工作线程在需要连接到服务器时会调用信号量的获取和释放方法：

with pool_sema:
    conn = connectdb()
    try:
        # ... use connection ...
    finally:
        conn.close()
1
2
3
4
5
6

改造之前的多线程爬虫

首先贴出原来的代码

# -*- coding:utf-8 -*-
'''
遇到问题没人解答？小编创建了一个Python学习交流QQ群：702813599
寻找有志同道合的小伙伴，互帮互助,群里还有不错的视频学习教程和PDF电子书！
'''
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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194

将代码进行改造

1、下面的第8、9行表示调用 threading 的 BoundedSemaphore类，初始化信号量为5，把结果赋给变量 pool_sema

'''
遇到问题没人解答？小编创建了一个Python学习交流QQ群：702813599
寻找有志同道合的小伙伴，互帮互助,群里还有不错的视频学习教程和PDF电子书！
'''
if __name__ == '__main__':
    start = time.time()
    print('这是主线程：{}'.format(threading.current_thread().name))

    urls = get_all_image_url(5)  # 获取所有图片url列表
    thread_list = []  # 定义一个列表，向里面追加线程

    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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

（2）方法二：直接使用 acquire()和 release()

下面的第8行调用 acquire()，第24行调用release()

'''
遇到问题没人解答？小编创建了一个Python学习交流QQ群：702813599
寻找有志同道合的小伙伴，互帮互助,群里还有不错的视频学习教程和PDF电子书！
'''
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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

最终效果是一样的，每次启用5个线程，完成后再启动下一批

尾语

好了，今天的分享就差不多到这里了！

对下一篇大家想看什么，可在评论区留言哦！看到我会更新哒(ง •_•)ง

喜欢就关注一下博主，或点赞收藏评论一下我的文章叭！！！

最后，宣传一下呀~👇👇👇 更多源码、资料、素材、解答、交流 皆点击下方名片获取呀👇👇👇