线程池

模拟网页下载的任务

def get_page_html(seconds, url):
    # sleep模拟下载耗时
    time.sleep(seconds)
    logger.info(f"URL:{url}下载网页成功，耗时{seconds}秒")
    # 返回页面源码+耗时
    return url, seconds

实例化线程池，并提交下载任务

from concurrent.futures import ThreadPoolExecutor

# 实例化线程池，并指定线程池的容量
executor = ThreadPoolExecutor(max_workers=1)

# submit方法:向线程池中提交任务：函数名+参数列表
task_1 = executor.submit(get_page_html, 3, "http://3.com")
task_2 = executor.submit(get_page_html, 2, "http://2.com")

任务执行完成后，返回一个Future对象

# 未来对象，或者是任务执行结果的容器
class Future(object):
    """Represents the result of an asynchronous computation."""

Future对象里的常用方法

1、获取线程的执行状态

# 获取线程任务的执行状态
logger.info(f"task_1是否执行完毕: {task_1.done()}")
logger.info(f"task_2是否执行完毕: {task_2.done()}")

2、获取线程任务的执行结果

# 获取线程异步执行完成后的结果,这里会阻塞主线程，等待子线程的执行结果
logger.info(task_1.result())

result()是一个阻塞的方法

3、取消线程任务的执行

# 在线程未开启前，即没有进到线程池中前，可以关闭该线程任务
task_2.cancel()

4、添加回调函数

# 线程任务执行完成，可以指定回调函数，接受一个future作为参数
def callback(future):
    logger.info(future.result())
# 添加回调函数
task_1.add_done_callback(callback)

批量提交线程任务

# 实例化线程池对象
executor = ThreadPoolExecutor(max_workers=3)

# 使用列表生成器批量提交任务到线程池
all_task = [executor.submit(get_page_html, seconds, url) for seconds, url in [(1, "http://1"), (2, "http://2")]]

线程任务执行完毕后立即接收返回值

from concurrent.futures import ThreadPoolExecutor, as_completed, wait, ALL_COMPLETED

# as_completed是一个生成器，会把执行完成的Future yield回来
for future in as_completed(fs=all_task):
    logger.info(future.result())

等待线程任务全部执行完毕

# 主线程阻塞，等待所有子线程执行完毕
wait(all_task, return_when=ALL_COMPLETED)
logger.info("主线程执行完毕")

使用线程池执行Selenium程序

# _*_coding:utf-8_*_

"""
@Time   : 2020/11/27 10:10
@Author : CarpLi
@File   : thread_pool_selenium.py
@Desc   : 使用线程池启动Chrome访问百度
"""
import time
from loguru import logger
from concurrent.futures import ThreadPoolExecutor, as_completed
from selenium.webdriver import Chrome


# chrome浏览器访问url
def open_url_page(url):
    chrome = Chrome()
    chrome.get(url)
    time.sleep(2)
    title = chrome.title
    chrome.quit()
    return title


executor = ThreadPoolExecutor(max_workers=5)
all_task = [executor.submit(open_url_page, f"https://www.baidu.com/s?wd={url}") for url in range(1, 20)]

for future in as_completed(all_task):
    # 有线程任务执行成功就直接返回
    logger.info(future.result())


if __name__ == '__main__':
    pass