七.并发编程 (线程池,返回值，回调函数)

一 .线程池(ThreadPoolExecutor)

https://www.cnblogs.com/nickchen121/p/11141751.html#autoid-3-0-0

#1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

#2 基本方法
#submit(fn, *args, **kwargs)
异步提交任务

#map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

#shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数

# done()
判断某一个线程是否完成

# cancle()
取消某个任务

1. 线程池异步

import  time
from concurrent.futures import ThreadPoolExecutor
def fun(n):
    time.sleep(1)
    print("用线程池启动的个数为%s"%n)
po=ThreadPoolExecutor(max_workers=5) #  默认不超过cpu个数*5
for i in  range(5):
     po.submit(fun,i)     #  线程池是用submit来提交  传递参数  异步提交任务

print("主线程")

主线程
用线程池启动的个数为3
用线程池启动的个数为0
用线程池启动的个数为2
用线程池启动的个数为1
用线程池启动的个数为4
进程已结束，退出代码 0

import  time

from concurrent.futures import ThreadPoolExecutor

def fun(n):
    time.sleep(1)
    print("启动线程的个数为%s"%n)
po=ThreadPoolExecutor(max_workers=5) #  默认不超过cpu个数*5
for i in  range(6):
     po.submit(fun,i)     #  线程池是用submit来提交  传递参数  异步提交任务
po.shutdown()    # 这里里面自带了close  join
print("主线程")

启动线程的个数为0
启动线程的个数为4
启动线程的个数为2
启动线程的个数为3
启动线程的个数为1
启动线程的个数为5
主线程

进程已结束，退出代码 0

 2.线程返回值

import  time
from concurrent.futures import ThreadPoolExecutor

def fun(n):
    time.sleep(1)
    return  n*n

po=ThreadPoolExecutor(max_workers=5) #  默认不超过cpu个数*5
ll=[]
for i in  range(6):
     po.submit(fun,i)     #  线程池是用submit来提交  传递参数  异步提交任务
     ll.append(po)
po.shutdown()  #  里面自带close json
print("主线程")
for p in ll:print(p.result())   #     result() 来获取返回值

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s线程的id号' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':
    executor=ProcessPoolExecutor(max_workers=3)
    futures=[]
    for i in range(5):
        future=executor.submit(task,i)
        futures.append(future)
    executor.shutdown(True)

    print('主线程!!!!!!!!!!!!!!!!!!11')
    for future in futures:
        print(future.result())

3256线程的id号
1228线程的id号
7608线程的id号
7608线程的id号
3256线程的id号
主线程!!!!!!!!!!!!!!!!!!11
0
1
4
9
16

进程已结束，退出代码 0

map是获取不到返回值

import  time
from concurrent.futures import ThreadPoolExecutor
def fun(n):
    time.sleep(1)
    return  n*n

po=ThreadPoolExecutor(max_workers=5) #  默认不超过cpu个数*5
po.map(fun,range(5))  #  map是拿不到返回值的
print("主线程")

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ThreadPoolExecutor(max_workers=3)

    # for i in range(11):
    #     future=executor.submit(task,i)

    executor.map(task,range(1,12)) #map取代了for+submit
    
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
4468 is runing
进程已结束，退出代码 0

3. 线程回调函数(和进程使用方式一样)

import  time,os
from concurrent.futures import ThreadPoolExecutor

def fun(n):
    time.sleep(1)
    return  n*5

def show(a):
    print(a)
    print(a.result())  # 获取结果

    print("线程号",os.getpid())   

po=ThreadPoolExecutor(max_workers=5)

for i in  range(2):
     po.submit(fun,5).add_done_callback(show)
print("主线程",os.getpid())

主进程 9808
<Future at 0x2466fe4b5c0 state=finished returned int>
25
线程号 9808
<Future at 0x2466fb42b00 state=finished returned int>
25
线程号 9808

进程已结束，退出代码 0

2	需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程： 我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数

# 先执行fun1函数 fun1函数的返回值  作为回调函数的参数  然后去执行回调函数