多线程实现之线程池

1 引入

• 系统启动一个新线程的成本是比较高的，因为它涉及与操作系统的交互。

  • 在这种情形下，使用线程池可以很好地提升性能，
  • 尤其是当程序中需要创建大量生存期很短暂的线程时，更应该考虑使用线程池。

• 线程池在系统启动时即创建大量空闲的线程，程序只要将一个函数提交给线程池，线程池就会启动一个空闲的线程来执行它。

  • 当该函数执行结束后，该线程并不会死亡，而是再次返回到线程池中变成空闲状态，等待执行下一个函数。

• 此外，使用线程池可以有效地控制系统中并发线程的数量。

  • 当系统中包含有大量的并发线程时，会导致系统性能急剧下降，甚至导致解释器崩溃，而线程池的最大线程数参数可以控制系统中并发线程的数量不超过此数。

2 模块

• 线程池的基类是 concurrent.futures 模块中的 Executor，Executor 提供了两个子类，
  • 即 ThreadPoolExecutor 和 ProcessPoolExecutor，
    • ThreadPoolExecutor 用于创建线程池，
    • ProcessPoolExecutor 用于创建进程池。

3 创建步骤

• 使用线程池来执行线程任务的步骤如下：

  • 调用 ThreadPoolExecutor 类的构造器创建一个线程池。

  • 定义一个普通函数作为线程任务。

  • 调用 ThreadPoolExecutor 对象的 submit() 方法来提交线程任务。

  • 调用 ThreadPoolExecutor 对象的 shutdown(wait = True) 方法来关闭线程池。

  def fun(index):
  	print(index)
  	time.sleep(3)
  	
  from concurrent.futures import ThreadPoolExecutor
  theard_pool = ThreadPoolExecutor(max_workers=2)
  for i in range(1000):
  	thread_pool.submit(fun, i)
  thread_pool.shutdown(wait= True)
  

import time
from concurrent.futures import ThreadPoolExecutor


def task(i):
    print(f'任务{i}开始！')
    time.sleep(i)
    print(f'任务{i}结束！')
    return i


start = time.time()
pool = ThreadPoolExecutor(3)

future01 = pool.submit(task, 1)
# print("future01是否结束", future01.done())
# 当程序使用 Future 的 result() 方法来获取结果时，该方法会阻塞当前线程，如果没有指定 timeout 参数，当前线程将一直处于阻塞状态，直到 Future 代表的任务返回。
# print("future01的结果", future01.result())  # 同步等待
future02 = pool.submit(task, 2)
future03 = pool.submit(task, 3)
pool.shutdown()  # 阻塞等待
print(f"程序耗时{time.time() - start}秒钟")

print("future01的结果", future01.result())
print("future02的结果", future02.result())
print("future03的结果", future03.result())

使用线程池来执行线程任务的步骤如下：

1. 调用 ThreadPoolExecutor 类的构造器创建一个线程池。
2. 定义一个普通函数作为线程任务。
3. 调用 ThreadPoolExecutor 对象的 submit() 方法来提交线程任务。
4. 当不想提交任何任务时，调用 ThreadPoolExecutor 对象的 shutdown() 方法来关闭线程池。

4 应用案例

import requests
from lxml import etree
import os
import asyncio
import time
import threading


def get_img_urls():
    res = requests.get("https://www.pkdoutu.com/photo/list/", headers={
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36"
    })
    
    selector = etree.HTML(res.text)
    img_urls = selector.xpath('//li[@class="list-group-item"]/div/div/a/img[@data-backup]/@data-backup')

    print(img_urls)
    return img_urls


def save_img(url):
    res = requests.get(url)
    name = os.path.basename(url)
    with open("imgs/" + name, "wb") as f:
        f.write(res.content)
    print(f"{name}下载完成！")


def main():
    img_urls = get_img_urls()
    # 串行
    [save_img(url) for url in img_urls]
    # 协程并发
    t_list = []
    for url in img_urls:
        t = threading.Thread(target=save_img, args=(url,))
        t.start()
        t_list.append(t)

    for t in t_list:
        t.join()


if __name__ == '__main__':
    start = time.time()
    main()
    end = time.time()
    print(end - start)

针对IO密集型任务，Python多线程可以发挥出不错的并发作用