Python并发编程进阶：从初学者到高手的进程与模块开发指南

concurrent.futures.thread.ThreadPoolExecutor #线程池
concurrent.futures.process.ProcessPoolExecutor #进程池
#构造函数
　　def __init__(self, max_workers=None, mp_cOntext=None,
　　initializer=None, initargs=()):

add_done_callback(self,fn)
cancel(self)
cancelled(self)
done(self)
exception(self,timeout=None)
result(self,timeout=None)
running(self)
set_exception(self,exception)
set_result(self,result)
set_running_or_notify_cancel(self)

from concurrent.futures import ProcessPoolExecutor
import os
import time
def task(i):
print("第"+str(i)+"个在执行任务id:"+str(os.getpid()))
time.sleep(1)
if __name__ == \'__main__\':
start=time.time()
pool = ProcessPoolExecutor(4) # 进程池里又4个进程
for i in range(5): # 5个任务
pool.submit(task,i)# 进程池里当前执行的任务i，池子里的4个进程一次一次执行任务
pool.shutdown()
print("耗时：",time.time()-start)

from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
import time
def task(i):
print("第"+str(i)+"个在执行任务id:"+currentThread().name)
time.sleep(1)
if __name__ == \'__main__\':
start = time.time()
pool = ThreadPoolExecutor(4) # 进程池里又4个线程
for i in range(5): # 5个任务
pool.submit(task,i)# 线程池里当前执行的任务i，池子里的4个线程一次一次执行任务
pool.shutdown()
print("耗时：",time.time()-start)

import time
from concurrent.futures import ProcessPoolExecutor,as_completed,ThreadPoolExecutor
def get_html(i):
times=1
time.sleep(times)
print("第 NO.{i} get page {times} finished".format(i=i,times=times))
return "第 NO.{i} ".format(i=i)
start=time.time()
executor = ThreadPoolExecutor(max_workers=2)
# executor = ProcessPoolExecutor(max_workers=2) #进程池会导致后面的all_task报错

all_task = [executor.submit(get_html,(i)) for i in range(5)]
for future in as_completed(all_task):
data = future.result()
print("in main:get page {} success".format(data))
print(\'主进程结束--耗时\',time.time()-start)

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os,time,random
def task(i):
print("第"+str(i)+"个在执行任务id:"+str(os.getpid()))
time.sleep(1)
if __name__ == \'__main__\':
start=time.time()
pool=ProcessPoolExecutor(max_workers=3) #也可以换成ThreadPoolExecutor
pool.map(task,range(1,5)) #map取代了for+submit
pool.shutdown()
print("耗时：",time.time()-start)

import time
from random import random
from concurrent.futures import ProcessPoolExecutor,as_completed,ThreadPoolExecutor
def get_html(i):
times=1+random()/100
time.sleep(times)
print("第 NO.{i} get page {times}s finished".format(i=i,times=times))
return "第 NO.{i} ".format(i=i)
start=time.time()
executor = ThreadPoolExecutor(max_workers=2)
# executor = ProcessPoolExecutor(max_workers=2) #进程池会导致后面的executor.map报错
res=executor.map(get_html, range(5))
#
for future in res: #直接返回结果，不需要get
print("in main:get page {} success".format(future))
print(\'主进程结束--耗时\',time.time()-start)

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process
import time
n = 1
def task(i):
global n
time.sleep(1)
print(f\'{current_process().name} 在执行任务{i}\')
n += i
return f\'得到 {current_process().name} 任务{i} 的结果\'
if __name__ == \'__main__\':
start=time.time()
pool = ProcessPoolExecutor(2) # 进程池里又4个线程
pool_lis = []
for i in range(5): # 20个任务
future = pool.submit(task,i)# 进程池里当前执行的任务i，池子里的4个线程一次一次执行任务

pool_lis.append(future.result()) #等待我执行任务得到的结果,如果一直没有结果，则阻塞。这里会导致我们所有任务编程了串行
# 在这里就引出了下面的pool.shutdown()方法
pool.shutdown(wait=True) # 关闭了池的入口，不允许在往里面添加任务了，会等带所有的任务执行完，结束阻塞
for res in pool_lis:
print(res)
print(n)# 这里肯定是拿到0的
print("主进程---耗时",time.time()-start)
# 可以用join去解决，等待每一个进程结束后，拿到他的结果

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process
import time
n = 1
def task(i):
global n
time.sleep(1)
print(f\'{current_process().name} 在执行任务{i}\')
n += i
return f\'得到 {current_process().name} 任务{i} 的结果\'
if __name__ == \'__main__\':
start=time.time()
pool = ProcessPoolExecutor(2) # 进程池里又4个线程
pool_lis = []
for i in range(5): # 20个任务
future = pool.submit(task,i)# 进程池里当前执行的任务i，池子里的4个线程一次一次执行任务
# print(future.result()) # 这是在等待我执行任务得到的结果，如果一直没有结果，这里会导致我们所有任务编程了串行
# 在这里就引出了下面的pool.shutdown()方法
pool_lis.append(future)
pool.shutdown(wait=True) # 关闭了池的入口，不允许在往里面添加任务了，会等带所有的任务执行完，结束阻塞
for p in pool_lis:
print(p.result())
print(n)# 这里肯定是拿到0的
print("主进程---耗时",time.time()-start)
# 可以用join去解决，等待每一个进程结束后，拿到他的结果

from multiprocessing import current_process
import time
from random import random
from concurrent.futures import ProcessPoolExecutor
def task(i):
print(f\'{current_process().name} 在执行{i}\')
time.sleep(1+random())
return i
# parse 就是一个回调函数
def parse(future):
# 处理拿到的结果
print(f\'{current_process().name} 拿到结果{future.result()} 结束了当前任务\')
if __name__ == \'__main__\':
start=time.time()
pool = ProcessPoolExecutor(2)
for i in range(5):
future = pool.submit(task,i)
\'\'\'
给当前执行的任务绑定了一个函数，在当前任务结束的时候就会触发这个函数（称之为回调函数）
会把future对象作为参数传给函数
注：这个称为回调函数，当前任务处理结束了，就回来调parse这个函数
\'\'\'
future.add_done_callback(parse)
# add_done_callback (parse) parse是一个回调函数
# add_done_callback () 是对象的一个绑定方法，他的参数就是一个函数
pool.shutdown()
print(\'主线程耗时：\',time.time()-start)

from concurrent.futures import ThreadPoolExecutor,wait,ALL_COMPLETED,FIRST_COMPLETED
import time
#参数times用来模拟网络请求时间
from random import random
def get_html(i):
times=1+random()*10
time.sleep(times)
print("第 NO.{i} get page {times}s finished".format(i=i,times=times))
return "第 NO.{i} ".format(i=i)
executor = ThreadPoolExecutor(max_workers=2)
urls = range(5)
all_task = [executor.submit(get_html,(url)) for url in urls]
wait(all_task,return_when=ALL_COMPLETED)
print("main")