进程池、线程池、异步调用（取返回值）

进程池vs线程池

为什么要用“池”：
池子使用来限制并发的任务数目，限制我们的计算机在一个自己可承受的范围内去并发地执行任务

池子内什么时候装进程：并发的任务属于计算密集型
池子内什么时候装线程：并发的任务属于IO密集型
（

concurrent:并发的，一致的，同时发生的  Executor执行者 

）

'''
#1、阻塞与非阻塞指的是程序的两种运行状态
    阻塞：遇到IO就发生阻塞，程序一旦遇到阻塞操作就会停在原地，并且立刻释放CPU资源
    非阻塞（就绪态或运行态）：没有遇到IO操作，或者通过某种手段让程序即便是遇到IO操作也不会停在原地，执行其他操作，力求尽可能多的占有CPU


#2、同步与异步指的是提交任务的两种方式：
    同步调用：提交完任务后，就在原地等待，直到任务运行完毕后，拿到任务的返回值，才继续执行下一行代码
    异步调用：提交完任务后，不在原地等待，直接执行下一行代码，等到任务有返回值后自动触发后调函数

'''

#进程池
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random

def task(x):
    print('%s 接客' %os.getpid())
    time.sleep(random.randint(2,5))
    return x**2


if __name__ == '__main__':
    # p=ProcessPoolExecutor(3) # 不指定默认开启的进程数是cpu的核数
    # print(os.cpu_count())查看CPU的核数
    p=ProcessPoolExecutor(max_workers=2) # 默认开启的进程数是cpu的核数,也就是干活的人数

    #简写成for循环模拟6个任务执行
    for i in range(6):
        p.submit(task,i)  #submit只是负责往池子里面丢任务，刚开始2个人同时接客


#1.submit往里面丢任务干活的人就是你指定的ProcessPoolExecutor(2)后面的人数
#2.刚一造个进程池，立马给你预备2个人干活，无论你丢多少任务，干活的永远是那2个人（谁先完了腾出手再接下一个）

#线程池
def task(x):
    print('%s is running'%x)
    time.sleep(random.randint(2, 5))
    # return x**2

if __name__ == '__main__':
    # p=ProcessPoolExecutor(max_workers=2)
    p=ThreadPoolExecutor(20)   #默认不写开启的是CPU核数*5

    for i in range(25):
        p.submit(task,i)

result功能（可以通过每次线程或者进程对象.result(),拿到返回值）

#线程池（result功能）
def task(x):
    print('%s is running'%x)
    time.sleep(random.randint(2, 5))
    return x**2

if __name__ == '__main__':
    # p=ProcessPoolExecutor(max_workers=2)
    p=ThreadPoolExecutor(3)   #默认不写开启的是CPU核数*5

    for i in range(6):
        obj=p.submit(task,i)
        print(obj.result())     #重点：每个线程打印 is running后，obj.result()都回去等着拿return x**2的返回值，所以输出结果会是串行的输出

    print('主')

'''结果：因为每次
1 is running
1
2 is running
4
3 is running
9
4 is running
16
5 is running
25
主


'''

 线程池、进程池下异步调用分析：

版本1：

#线程池下分析异步调用（取返回值），分析版本1
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time,os,random
#线程池为例：
def task(x):
    print('%s is running'%x)
    time.sleep(random.randint(2, 5))
    return x**2

if __name__ == '__main__':
    异步调用：submit只负责提交任务，不会等待拿结果不会拿return
    p=ThreadPoolExecutor(4)
    obj_l=[]
    for i in range(10):
        obj=p.submit(task,i)   #每个返回obj对象里面，其实都封装了一个result方法
        # print(obj.result()) #可以通过result拿到每次return的返回值
        obj_l.append(obj)
        # print(obj_l)
    # p.close()
    # p.join() #
    p.shutdown(wait=True)  #关闭进程池入口 等着取一个数目就减1

    #实现异步调用拿结果：如何取结果：
    print(obj_l[0].result())  #第一个任务的结果 即return 0**2
    print(obj_l[1].result())  #第二个：1**2
    print('主')

    同步调用：每次都拿到返回结果
    p=ThreadPoolExecutor(4)
    for i in range(10):
        obj=p.submit(task,i)
        print(obj.result())   #每次等待返回值后再执行下一个
    print('主')

版本2：

#线程池下分析异步调用（取返回值），分析版本2
def task(x):
    print('%s is running' % x)
    time.sleep(10)
    return x ** 2

def parse(res):
    print('.......')

if __name__ == '__main__':
    # 异步调用：
    p=ThreadPoolExecutor(4)
    obj_l=[]
    for i in range(10):
        obj=p.submit(task,i)
        obj_l.append(obj)

    #实现目的：为了保证进程池里面任务全部运行完，后再统一拿返回值
    p.shutdown(wait=True)  #1.关闭进程池入口（确保统计时进程池不会再有数据进入），等着取一个数目就减1
    # print(obj_l)  #第一个任务的结果 即return 0**2
    for future in obj_l:
        # print(future.result())
        parse(future.result())  #4个任务一次10s + 4个任务（每个1s）=共14s
    print('主')

最终版本（引用回调函数）

#线程池、进程池异步调用取返回值最终版（引用回调函数add_done_callback()）
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time, os, random
import os

def task(x):
    print('%s is running' %x)
    # time.sleep(1)
    return x ** 2
    # parse(x**2)  #产生结果后立马被处理

def parse(future):
    # time.sleep(1)
    res=future.result()
    print('%s 处理了 %s'%(os.getpid(),res))  #统一是主进程的一个Pid
if __name__ == '__main__':
    poo1=ProcessPoolExecutor(4)  #进程为9s左右
    # poo1=ThreadPoolExecutor(4)   #线程为6s左右，线程效率更高
    start=time.time()
    for i in range(1,5):
        future=poo1.submit(task,i)
        #回调函数就是有结果之后立马执行结果，回调是主进程去处理运行结果，所以parse函数os.getpid()pid都是主进程一个PID
        #提交完之后给对象绑定了回调函数，parse会在future有返回值是立即触发，并且将future当做参数传给parse
        future.add_done_callback(parse) #绑定回调函数
    poo1.shutdown(wait=True) #关闭进程池入口，确保进程池里面不会再有数据进入
    stop=time.time()

    print('主',(stop-start))

'''
总结：根据以上执行
    回调函数：
        1.在线程池环境下，处理回调函数是线程池里面的运行的，是线程池里面的线程去执行的
        （还是I/O密集型）
        
        2.在进程池环境下，处理回调函数是主进程去处理的
        
        
'''

线程池与进程池里面没有join（）这个方法，