程序员经验分享-hwhale

Python并发编程之线程池/进程池

2023-11-08

转载

http://python.jobbole.com/87272/

引言

Python标准库为我们提供了threading和multiprocessing模块编写相应的多线程/多进程代码，但是当项目达到一定的规模，频繁创建/销毁进程或者线程是非常消耗资源的，这个时候我们就要编写自己的线程池/进程池，以空间换时间。但从Python3.2开始，标准库为我们提供了concurrent.futures模块，它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，实现了对threading和multiprocessing的进一步抽象，对编写线程池/进程池提供了直接的支持。

Executor和Future

concurrent.futures模块的基础是Exectuor，Executor是一个抽象类，它不能被直接使用。但是它提供的两个子类ThreadPoolExecutor和ProcessPoolExecutor却是非常有用，顾名思义两者分别被用来创建线程池和进程池的代码。我们可以将相应的tasks直接放入线程池/进程池，不需要维护Queue来操心死锁的问题，线程池/进程池会自动帮我们调度。

Future这个概念相信有java和nodejs下编程经验的朋友肯定不陌生了，你可以把它理解为一个在未来完成的操作，这是异步编程的基础，传统编程模式下比如我们操作queue.get的时候，在等待返回结果之前会产生阻塞，cpu不能让出来做其他事情，而Future的引入帮助我们在等待的这段时间可以完成其他的操作。关于在Python中进行异步IO可以阅读完本文之后参考我的Python并发编程之协程/异步IO。

p.s: 如果你依然在坚守Python2.x，请先安装futures模块。

pip install futures

1

pip install futures

 

使用submit来操作线程池/进程池

我们先通过下面这段代码来了解一下线程池的概念

Python

1

2

3

4

5

6

7

8

9

10

11

12

13

14

# example1.py

from concurrent.futures import ThreadPoolExecutor

import time

def return_future_result(message):

    time.sleep(2)

    return message

pool = ThreadPoolExecutor(max_workers=2)  # 创建一个最大可容纳2个task的线程池

future1 = pool.submit(return_future_result, ("hello"))  # 往线程池里面加入一个task

future2 = pool.submit(return_future_result, ("world"))  # 往线程池里面加入一个task

print(future1.done())  # 判断task1是否结束

time.sleep(3)

print(future2.done())  # 判断task2是否结束

print(future1.result())  # 查看task1返回的结果

print(future2.result())  # 查看task2返回的结果

我们根据运行结果来分析一下。我们使用submit方法来往线程池中加入一个task，submit返回一个Future对象，对于Future对象可以简单地理解为一个在未来完成的操作。在第一个print语句中很明显因为time.sleep(2)的原因我们的future1没有完成，因为我们使用time.sleep(3)暂停了主线程，所以到第二个print语句的时候我们线程池里的任务都已经全部结束。

1

2

3

4

5

6

7

8

9

10

ziwenxie :: ~ » python example1.py

False

True

hello

world

# 在上述程序执行的过程中，通过ps命令我们可以看到三个线程同时在后台运行

ziwenxie :: ~ » ps -eLf | grep python

ziwenxie      8361  7557  8361  3    3 19:45 pts/0    00:00:00 python example1.py

ziwenxie      8361  7557  8362  0    3 19:45 pts/0    00:00:00 python example1.py

ziwenxie      8361  7557  8363  0    3 19:45 pts/0    00:00:00 python example1.py

 

上面的代码我们也可以改写为进程池形式，api和线程池如出一辙，我就不罗嗦了。

Python

1

2

3

4

5

6

7

8

9

10

11

12

13

14

# example2.py

from concurrent.futures import ProcessPoolExecutor

import time

def return_future_result(message):

    time.sleep(2)

    return message

pool = ProcessPoolExecutor(max_workers=2)

future1 = pool.submit(return_future_result, ("hello"))

future2 = pool.submit(return_future_result, ("world"))

print(future1.done())

time.sleep(3)

print(future2.done())

print(future1.result())

print(future2.result())

下面是运行结果

 

1

2

3

4

5

6

7

8

9

10

11

ziwenxie :: ~ » python example2.py

False

True

hello

world

ziwenxie :: ~ » ps -eLf | grep python

ziwenxie      8560  7557  8560  3    3 19:53 pts/0    00:00:00 python example2.py

ziwenxie      8560  7557  8563  0    3 19:53 pts/0    00:00:00 python example2.py

ziwenxie      8560  7557  8564  0    3 19:53 pts/0    00:00:00 python example2.py

ziwenxie      8561  8560  8561  0    1 19:53 pts/0    00:00:00 python example2.py

ziwenxie      8562  8560  8562  0    1 19:53 pts/0    00:00:00 python example2.py

 

使用map/wait来操作线程池/进程池

除了submit，Exectuor还为我们提供了map方法，和内建的map用法类似，下面我们通过两个例子来比较一下两者的区别。

使用submit操作回顾

 

Python

 

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

# example3.py

import concurrent.futures

import urllib.request

URLS = ['http://httpbin.org', 'http://example.com/', 'https://api.github.com/']

def load_url(url, timeout):

    with urllib.request.urlopen(url, timeout=timeout) as conn:

        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:

    # Start the load operations and mark each future with its URL

    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}

    for future in concurrent.futures.as_completed(future_to_url):

        url = future_to_url[future]

        try:

            data = future.result()

        except Exception as exc:

            print('%r generated an exception: %s' % (url, exc))

        else:

            print('%r page is %d bytes' % (url, len(data)))

从运行结果可以看出，as_completed不是按照URLS列表元素的顺序返回的。

 

1

2

3

4

ziwenxie :: ~ » python example3.py

'http://example.com/' page is 1270 byte

'https://api.github.com/' page is 2039 bytes

'http://httpbin.org' page is 12150 bytes

 

使用map

 

Python

1

2

3

4

5

6

7

8

9

10

11

# example4.py

import concurrent.futures

import urllib.request

URLS = ['http://httpbin.org', 'http://example.com/', 'https://api.github.com/']

def load_url(url):

    with urllib.request.urlopen(url, timeout=60) as conn:

        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:

    for url, data in zip(URLS, executor.map(load_url, URLS)):

        print('%r page is %d bytes' % (url, len(data)))

从运行结果可以看出，map是按照URLS列表元素的顺序返回的，并且写出的代码更加简洁直观，我们可以根据具体的需求任选一种。

1

2

3

4

ziwenxie :: ~ » python example4.py

'http://httpbin.org' page is 12150 bytes

'http://example.com/' page is 1270 bytes

'https://api.github.com/' page is 2039 bytes

 

第三种选择wait

wait方法接会返回一个tuple(元组)，tuple中包含两个set(集合)，一个是completed(已完成的)另外一个是uncompleted(未完成的)。使用wait方法的一个优势就是获得更大的自由度，它接收三个参数FIRST_COMPLETED, FIRST_EXCEPTION 和ALL_COMPLETE，默认设置为ALL_COMPLETED。

我们通过下面这个例子来看一下三个参数的区别

Python

1

2

3

4

5

6

7

8

9

10

11

12

from concurrent.futures import ThreadPoolExecutor, wait, as_completed

from time import sleep

from random import randint

def return_after_random_secs(num):

    sleep(randint(1, 5))

    return "Return of {}".format(num)

pool = ThreadPoolExecutor(5)

futures = []

for x in range(5):

    futures.append(pool.submit(return_after_random_secs, x))

print(wait(futures))

# print(wait(futures, timeout=None, return_when='FIRST_COMPLETED'))

如果采用默认的ALL_COMPLETED，程序会阻塞直到线程池里面的所有任务都完成。

1

2

3

4

5

6

7

ziwenxie :: ~ » python example5.py

DoneAndNotDoneFutures(done={

<Future at 0x7f0b06c9bc88 state=finished returned str>,

<Future at 0x7f0b06cbaa90 state=finished returned str>,

<Future at 0x7f0b06373898 state=finished returned str>,

<Future at 0x7f0b06352ba8 state=finished returned str>,

<Future at 0x7f0b06373b00 state=finished returned str>}, not_done=set())

如果采用FIRST_COMPLETED参数，程序并不会等到线程池里面所有的任务都完成。

1

2

3

4

5

6

7

ziwenxie :: ~ » python example5.py

DoneAndNotDoneFutures(done={

<Future at 0x7f84109edb00 state=finished returned str>,

<Future at 0x7f840e2e9320 state=finished returned str>,

<Future at 0x7f840f25ccc0 state=finished returned str>},

not_done={<Future at 0x7f840e2e9ba8 state=running>,

<Future at 0x7f840e2e9940 state=running>})

 

思考题

写一个小程序对比multiprocessing.pool(ThreadPool)和ProcessPollExecutor(ThreadPoolExecutor)在执行效率上的差距，结合上面提到的Future思考为什么会造成这样的结果。

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

python

并发

Python并发编程之线程池/进程池 的相关文章

随机推荐

  • 2023 年程序员的最佳工作角色

    当今价值数十亿美元的计算机编程市场正受到下一代应用程序和产品的日益普及的推动 例如自动编码 统计计算 数据分析 ML 机器学习 和 AI 人工智能 根据MarketsandMarkets的一项研究 到11年 全球CAC 计算机辅助编码 行业

  • Redis三种集群架构

    一 主从架构 搭建主从结构 从节点配置步骤 1 复制一份redis conf文件 2 将相关配置修改为如下值 port 6380 pidfile var run redis 6380 pid 把pid进程号写入pidfile配置的文件 lo

  • 统计学第九周：参数估计python实现

    统计学第九周 参数估计复习 参数估计 根据从总体中随机取样获得样本 根据取样样本来估计总体分布中参数的过程 方法 估计形式上分 点估计与区间估计 估计的方法有矩法估计 最小二乘法估计 似然估计 贝叶斯估计等等 问题一般有 未知参数的估计量

  • 期货方法（期货方法很简单 只用MACD）

    期货法 在期货市场上 95 的人都是赔钱的 对于期货市场的新手来说 一定要有一个清晰的认识 95 的数字告诉我们什么 它只是告诉我们 在获得稳定的利润之前 我们不应该投入太多的钱 对于初学者来说 投入的钱越多 损失的钱就越多 所以一定要控制

  • 渗透之Aircrack—wifi破解

    环境 kali 工具使用 无线网卡 型号雷凌RT3070L 注意型号选择的使用 不同型号可能导致kali无法识别 步骤总结 1 kali连接外接无线网卡 2 ifconfig查看网卡信息 新出现的wlan0网卡 3 开启网卡监听模式 air

  • 实践积累：用Vue3简单写一个单行横向滚动组件

    目录 效果图 需求分析 实现分析 样式展示分析 变量分析 方法分析 实现步骤 1 实现模板 2 实现css 3 首先获取list 4 页面挂载后监听groupBoxRef的scroll事件并获取当前的滚动位置 5 计算展示的宽度显隐箭头 当

  • 讲透JVM类加载机制，向高手进阶！

    目录 前言 JVM在什么情况下会加载一个类 从实用角度出发 来看看验证 准备和初始化的过程 核心阶段 初始化 类加载器和双亲委派机制 1 前言 先来看一下JVM整体的一个运行原理 我们首先从 java 代码文件 编译成 class 字节码文

  • 运放的虚短和虚断以及分类

    放大器定义 能实现信号 功率放大的器件 称为放大器 英文为Amplifier 以放大器为核心 能实现放大功能的电路组合 称为放大电路 放大器的种类 全部放大器被分为三种 晶体管放大器 运算放大器和功能放大器 晶体管及其放大电路的复杂 从静态

  • 哈希表（散列表）原理详解

    什么是哈希表 哈希表 Hash table 也叫散列表 是根据关键码值 Key value 而直接进行访问的数据结构 也就是说 它通过把关键码值映射到表中一个位置来访问记录 以加快查找的速度 这个映射函数叫做散列函数 存放记录的数组叫做散列

  • 华中科技大学操作系统实验课 实验四

    一 实验目的 1 理解设备是文件的概念 2 掌握Linux模块 驱动的概念和编程流程 3 Windows Linux下掌握文件读写基本操作 二 实验内容 1 编写一个Linux内核模块 并完成模块的安装 卸载等操作 2 编写Linux驱动程

  • MySQL多表查询(8.0)

    文章目录 多表查询 1 多表关系 1 1 一对多 1 2 多对多 1 3 一对一 2 多表查询概述 2 1 数据准备 2 2 概述 2 3 分类 3 内连接 4 外连接 5 自连接 5 1 自连接查询 5 2 联合查询 6 子查询 6 1

  • chatGLM-Windows环境安装

    Windows系统下环境安装 一 概要 不同安装方式 安装python 安装Nvidia驱动 安装cuda与cuddn 安装PyTorch与TensorFlow 二 安装文件 百度网盘链接 https pan baidu com s 1lb

  • Prometheus部分监控项

    Metrics Chinese explanation English explanation node arp entries device的ARP表项 HELP node arp entries ARP entries by devic

  • Kaldi 入门详解

    train mono sh 是音素训练脚本 下面详细介绍各个功能 这部分是训练用参数 调用mono sh时可以通过 name value的方式改变这些参数 nj 4 并行个数 cmd run pl 处理程序 scale opts trans

  • 已知年月日利用公式求星期几模板

    在本文中 我们将使用C语言实现基于已知的年月日计算星期几的公式 这个公式被称为 蔡勒公式 Zeller s Congruence 是一种快速求解星期几的方法 代码分析 首先 我们需要对月份进行调整 如果月份小于3 即1月或2月 则将其视为上

  • 12个医学公共数据库

    01 NCDB 网址 https www facs org quality programs cancer ncdb 美国国家癌症数据库 National Cancer Database NCDB 是经国家认证的 由美国外科医师学会和美国癌

  • DBA_Oracle Table Partition表分区概念汇总（概念）

    一 摘要 有关表分区的一些维护性操作 注 分区根据具体情况选择 表分区有以下优点 1 数据查询 数据被存储到多个文件上 减少了I O负载 查询速度提高 2 数据修剪 保存历史数据非常的理想 3 备份 将大表的数据分成多个文件 方便备份和恢复

  • MySQL5.7保姆级安装教程

    环境 Linux版本 Mysql版本 待安装 CentOS 7 5 7 1 配置YUM源 在MySQL官网中下载YUM源rpm安装包 http dev mysql com downloads repo yum 目前MySQL官网下载的MyS

  • C++ 的string类学习

    一 string类型变量构造赋值方法 1 构造 1 用等号直接赋值S0 2 定义一个空白变量S1 3 定义一个新变量S2 内容完全等于S0 4 定义一个新变量S3 内容是S0从第八个字符开始的三个字符 5 定义一个新变量S4 用括号赋值 和

  • Python并发编程之线程池/进程池

    转载 http python jobbole com 87272 引言 Python标准库为我们提供了threading和multiprocessing模块编写相应的多线程 多进程代码 但是当项目达到一定的规模 频繁创建 销毁进程或者线程是

热门标签