参考网址

www.cnblogs.com/venvive/p/1…

1/ThreadPoolExecutor多线程

<1>为什么需求线程池呢?

关于io密集型，提高履行的效率。
线程的创立是需求消耗系统资源的。
所以线程池的思维便是：每个线程各自分配一个使命，剩余的使命排队等候，当某个线程完结了使命的时分，排队使命就能够安排给这个线程继续履行。

<2>规范库concurrent.futures模块

   它供给了ThreadPoolExecutor和ProcessPoolExecutor两个类，
   分别完结了对threading模块和multiprocessing模块的进一步抽象。
   不只能够帮咱们主动调度线程，还能够做到:
    - 主线程能够获取某一个线程(或者使命)的状况，以及回来值
    - 当一个线程完结的时分，主线程能够当即知道
    - 让多线程和多进程的编码接口一致

<3>简单运用

# -*-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和task2是使命句柄
task1 = executor.submit( get_html, (2) )
task2 = executor.submit( get_html, (3) )
# done()办法用于判别某个使命是否完结，bool型，完结回来True，没有完结回来False
print( task1.done() )
# cancel()办法用于撤销某个使命，该使命没有放到线程池中才干被撤销，如果已经放进线程池子中，则不能被撤销
# bool型，成功撤销了回来True，没有撤销回来False
print( task2.cancel() )
# result()办法能够获取task的履行成果，条件是get_html()函数有回来值
print( task1.result() )
print( task2.result() )
# 成果：
# get page 3s finished
# get page 2s finished
# True
# False
# 2
# 3
# ThreadPoolExecutor类在构造实例的时分，传入max_workers参数来设置线程池中最多能一起运转的线程数目
# 运用submit()函数来提交线程需求履行使命(函数名和参数)到线程池中，并回来该使命的句柄，
# 注意submit()不是堵塞的，而是当即回来。
# 通过submit()函数回来的使命句柄，能够运用done()办法判别该使命是否完毕，运用cancel()办法来撤销，
# 运用result()办法能够获取使命的回来值，查看内部代码，发现该办法是堵塞的

<4>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
# 创立线程池子
# 设置最多2个线程运转，其他等候
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
# 一次性把一切的使命都放进线程池，得到一个句柄，可是最多只能一起履行2个使命
all_task = [ executor.submit(get_html,(each_url)) for each_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
# 从成果能够看到，并不是先传入哪个url，就先履行哪个url，没有先后顺序

<5>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
# 创立线程池子
# 设置最多2个线程运转，其他等候
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
for result in executor.map(get_html, urls):
    print("in main:get page {}s success".format(result))
# 成果
# 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

<6>wait()办法

wait办法能够让主线程堵塞，直到满足设定的要求。
wait办法接收3个参数，等候的使命序列、超时时间以及等候条件。
等候条件return_when默以为ALL_COMPLETED,标明要等候一切的使命都借宿。
能够看到运转成果中，确实是一切使命都完结了，主线程才打印出main，
等候条件还能够设置为FIRST_COMPLETED,表明第一个使命完结就停止等候。
超时时间参数能够不设置。
wait()办法和as_completed(), map()没有关系。
不管你是用as_completed()，仍是用map()办法，你都能够在履行主线程之前运用wait()。
as_completed()和map()是二选一的。

# -*-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
# 创立线程池子
# 设置最多2个线程运转，其他等候
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

2/ProcessPoolExecutor多进程

<1>同步调用方法: 调用，然后等回来值，能解耦，可是速度慢

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("主")
################
################
# 另一个同步调用的demo
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

<2>异步调用方法：只调用，不等回来值，或许存在耦合，可是速度快

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('主')
##################
##################
# 另一个异步调用的demo
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

<3>怎样运用异步调用方法，但一起防止耦合的问题？

(1)进程池：异步 + 回调函数，，cpu密集型，一起履行，每个进程有不同的解释器和内存空间，互不搅扰

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

(2)线程池：异步 + 回调函数，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

3/总结

1、线程不是越多越好，会涉及cpu上下文的切换（会把上一次的记载保存）。
2、进程比线程消耗资源，进程相当于一个工厂，工厂里有很多人，里边的人一起享受着福利资源，，一个进程里默许只要一个主线程，比方：敞开程序是进程，里边履行的是线程，线程仅仅一个进程创立多个人一起去作业。
3、线程里有GIL大局解锁器：不允许cpu调度
4、核算密度型适用于多进程
5、线程：线程是核算机中作业的最小单元
6、进程：默许有主线程 (帮作业)能够多线程共存
7、协程：一个线程，一个进程做多个使命,运用进程中一个线程去做多个使命，微线程
8、GIL大局解释器锁：确保同一时间只要一个线程被cpu调度