码迷,mamicode.com
首页 > 编程语言 > 详细

ThreadPoolExecutor线程池和ProcessPoolExecutor进程池

时间:2019-09-28 09:13:46      阅读:73      评论:0      收藏:0      [点我收藏+]

标签:多线程   设定   roc   sina   range   first   log   baidu   最小   

ProcessPoolExecutor线程池

1、为什么需要线程池呢,如果创建了20个线程,而同时只允许3个线程在运行,但是20个线程都需要创建和销毁,线程的创建是需要消耗系统资源的,所以线程池的思想就是:每个线程各分配一个任务,剩下的任务皮队等待,当某个线程完成了任务的时候,排队任务就可以安排给这个线程继续执行

2、标准库concurrent.futures模块,它提供了 ProcessPoolExecutor和ThreadPoolExecutor两个类,实现了对threading和multiprocessing的进一步抽象(这里主要关注线程池),不仅可以帮我们自动调度线程,还可以做到:

<ul>
    <li>主线程可以获取某一个线程(或者任务)的状态,以及返回值</li>
    <li>当一个线程完成的时候,主线程能够立即知道</li>
    <li>让多线程和多进程的编码接口一致</li>
</ul>

简单使用

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor
import time
#参数times用来模拟网络请求时间
def get_html(times):
    print("get page {}s finished".format(times))
    return times
executor = ThreadPoolExecutor(max_workers=2)
#通过submit函数提交执行的函数到线程池中,submit函数立即返回,不阻塞
task1 = executor.submit(get_html,(3))
task2 = executor.submit(get_html,(2))
#done方法用于判断某个任务是否完成
print(task1.done())
#cancel方法用于取消某个任务,该任务没有放到线程池中才能被取消
print(task2.cancel())
print(task1.done())
#result方法可以获取task的执行结果
print(task1.result())

#结果:
# get page 3s finished
# get page 2s finished
# True
# False
# True
# 3
  • ThreadPoolExecutor构造实例的时候,传入max_workers参数来设置线程中最多能同时运行的线程数目
  • 使用submit函数来提交线程需要执行任务(函数名和参数)到线程池中,并返回该任务的句柄(类似于文件、画图),注意submit()不是阻塞的,而是立即返回。
  • 通过submit函数返回的任务句柄,能够使用done()方法判断该任务是否结束
  • 使用result()方法可以获取任务的返回值,查看内部代码,发现这个方法是阻塞的

as_completed

上面虽然提供了判断任务是否结束的方法,但是不能在主线程中一直判断,有时候我们是得知某个任务结束了,就去获取结果,而不是一直判断每个任务有没有结束。这是就可以使用as_completed方法一次取出所有任务的结果。

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,as_completed
import time
#参数times用来模拟网络请求时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
all_task = [executor.submit(get_html,(url)) for url in urls]
for future in as_completed(all_task):
    data = future.result()
    print("in main:get page {}s success".format(data))

#结果
# get page 2s finished
# in main:get page 2s success
# get page 3s finished
# in main:get page 3s success
# get page 4s finished
# in main:get page 4s success

map

除了上面的as_completed方法,还可以使用execumap方法,但是有一点不同,使用map方法,不需提前使用submit方法,map方法与python标准库中的map含义相同,都是将序列中的每个元素都执行同一个函数。上面的代码就是对urls的每个元素都执行get_html函数,并分配各线程池。可以看到执行结果与上面的as_completed方法的结果不同,输出顺序和urls列表的顺序相同,就算2s的任务先执行完成,也会先打印出3s的任务先完成,再打印2s的任务完成

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,as_completed
import time
#参数times用来模拟网络请求时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
for data in executor.map(get_html,urls):
    print("in main:get page {}s success".format(data))

#结果
# get page 2s finished
# get page 3s finished
# in main:get page 3s success
# in main:get page 2s success
# get page 4s finished
# in main:get page 4s success

wait

wait方法可以让主线程阻塞,直到满足设定的要求。wait方法接收3个参数,等待的任务序列、超时时间以及等待条件。等待条件return_when默认为ALL_COMPLETED,表明要等待所有的任务都借宿。可以看到运行结果中,确实是所有任务都完成了,主线程才打印出main,等待条件还可以设置为FIRST_COMPLETED,表示第一个任务完成就停止等待

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,wait,ALL_COMPLETED,FIRST_COMPLETED
import time
#参数times用来模拟网络请求时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
all_task = [executor.submit(get_html,(url)) for url in urls]
wait(all_task,return_when=ALL_COMPLETED)
print("main")
#结果
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main

ProcessPoolExecutor

方式一、同步调用方式:提交任务,原地等待任务执行结束,拿到任务返回结果。再执行下一行代码会导致任务串行执行

# -*- coding:utf-8 -*-
# 方式一、同步调用方式:提交任务,原地等待任务执行结束,拿到任务返回结果。再执行下一行代码会导致任务串行执行

# 进程池的两种任务提交方式

import datetime
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread
import time, random, os
import requests
def task(name):
    print('%s %s is running'%(name,os.getpid()))
    #print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
if __name__ == '__main__':
    p=ProcessPoolExecutor(4) #设置进程池内进程数
    for i in range(10):
        #同步调用方式,调用和等值
        obj = p.submit(task,"进程pid:")#传参方式(任务名,参数),参数使用位置或者关键字参数
        res =obj.result()
    p.shutdown(wait=True) #关闭进程池的入口,等待池内任务运行结束
    print("主")

异步提交方式

def task(name):
    print("%s %s is running" %(name,os.getpid()))
    time.sleep(random.randint(1,3))

if __name__ == '__main__':
    p = ProcessPoolExecutor(4) #设置进程池内进程
    for i in range(10):
        #异步调用方式,只调用,不等值
        p.submit(task,'进程pid:') #传参方式(任务名,参数),参数使用位置参数或者关键字参数
    p.shutdown(wait=True)
    print('主')

进程池的同步调用方式

同步调用优点解耦合,缺点:速度慢

def get(url):
    print('%s GET %s' % (os.getpid(),url))
    time.sleep(3)
    response = requests.get(url)
    if response.status_code == 200:
        res = response.text
    else:
        res = "下载失败"
    return res

def parse(res):
    time.sleep(1)
    print("%s 解析结果为%s" %(os.getpid(),len(res)))

if __name__ == "__main__":
    urls = [
        'https://www.baidu.com',
        'https://www.sina.com.cn',
        'https://www.tmall.com',
        'https://www.jd.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',

    ]
    p=ProcessPoolExecutor(9)
    l=[]
    start = time.time()
    for url in urls:
        future = p.submit(get,url)
        l.append(future)
    p.shutdown(wait=True)

    for future in l:
        parse(future.result())
    print('完成时间:',time.time()-start)
    #完成时间: 13.209137678146362

异步调用方式

异步调用方式:缺点存在耦合,优点速度快

def get(url):
    print('%s GET %s' % (os.getpid(),url))
    time.sleep(3)
    reponse = requests.get(url)
    if reponse.status_code == 200:
        res = reponse.text
    else:
        res = "下载失败"
    parse(res)

def parse(res):
    time.sleep(1)
    print('%s 解析结果为%s' %(os.getpid(),len(res)))

if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.sina.com.cn',
        'https://www.tmall.com',
        'https://www.jd.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',

    ]

    p = ProcessPoolExecutor(9)
    start = time.time()
    for url in urls:
        future = p.submit(get,url)
    p.shutdown(wait=True)
    print("完成时间",time.time()-start)#完成时间 6.293345212936401

2-1-4进程池,异步调用+回调函数:解决耦合,速度慢,(异步调用+回调)

def get(url):
    print('%s GET %s' % (os.getpid(), url))
    time.sleep(3)
    response = requests.get(url)
    if response.status_code == 200:
        res = response.text
    else:
        res = '下载失败'
    return res

def parse(future):
    time.sleep(1)
    # 传入的是个对象,获取返回值 需要进行result操作
    res = future.result()
    print("res",)
    print('%s 解析结果为%s' % (os.getpid(), len(res)))



if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.sina.com.cn',
        'https://www.tmall.com',
        'https://www.jd.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
    ]
    p = ProcessPoolExecutor(9)
    start = time.time()
    for url in urls:
        future = p.submit(get,url)
        #模块内的回调函数方法,parse会使用future对象的返回值,对象返回值是执行任务的返回值
        #回调应该是相当于parse(future)
        future.add_done_callback(parse)

    p.shutdown(wait=True)
    print("完成时间",time.time()-start)#完成时间 33.79998469352722

线程池:异步+回调 ----IO密集型主要使用方式,线程池:执行操作为谁有空谁执行

def get(url):
    print("%s GET %s" %(current_thread().name,url))
    time.sleep(3)
    reponse = requests.get(url)
    if reponse.status_code == 200:
        res = reponse.text
    else:
        res = "下载失败"
    return res

def parse(future):
    time.sleep(1)
    res = future.result()
    print("%s 解析结果为%s" %(current_thread().name,len(res)))

if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.sina.com.cn',
        'https://www.tmall.com',
        'https://www.jd.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
    ]
    p = ThreadPoolExecutor(4)
    start = time.time()
    for url in urls:
        future = p.submit(get,url)
        future.add_done_callback(parse)
    p.shutdown(wait=True)
    print("主",current_thread().name)
    print("完成时间",time.time()-start)#完成时间 32.52604126930237

1、线程不是越多越好,会涉及cpu上下文的切换(会把上一次的记录保存)。

2、进程比线程消耗资源,进程相当于一个工厂,工厂里有很多人,里面的人共同享受着福利资源,,一个进程里默认只有一个主线程,比如:开启程序是进程,里面执行的是线程,线程只是一个进程创建多个人同时去工作。

3、线程里有GIL全局解锁器:不允许cpu调度

4、计算密度型适用于多进程

5、线程:线程是计算机中工作的最小单元

6、进程:默认有主线程 (帮工作)可以多线程共存

7、协程:一个线程,一个进程做多个任务,使用进程中一个线程去做多个任务,微线程

8、GIL全局解释器锁:保证同一时刻只有一个线程被cpu调度

来自:https://www.jianshu.com/p/b9b3d66aa0be

来自:https://blog.csdn.net/qq_33961117/article/details/82587873

ThreadPoolExecutor线程池和ProcessPoolExecutor进程池

标签:多线程   设定   roc   sina   range   first   log   baidu   最小   

原文地址:https://www.cnblogs.com/venvive/p/11601190.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!