Python学习-并发编程之线程池

Python学习-并发编程之线程池

利用concurrent.futures模块创建线程池

ThreadPoolExecutor类概述

class ThreadPoolExecutor(_base.Executor):
def __init__(self, max_workers=None):
'''max_workers线程池中最大线程数'''
pass

def submit(self, fn, *args, **kwargs):
'''
向线程池中提交线程，默认是异步的，
fn：线程处理的对象，函数的地址，
*args：以位置传参的方式给处理对象传递参数，
**kwargs：以关键字传参的方式给对象传递参数,
返回一个Future类型的对象。
'''
pass

def map(self, fn, *iterables, timeout=None, chunksize=1):
'''
向线程池中提交一个类似如map()函数功能的进程
fn：进程处理的对象，函数的地址，
*iterables：传递一个可迭代的对象，
map()方法返回值为一个迭代器。
'''
pass

def shutdown(self, wait=True):
'''关闭线程池'''
pass

ThreadPoolExecutorr类创建线程池的方法

# 用ThreadPoolExecutor类创建的线程池默认都是异步的
import os,time,random
from concurrent.futures import  ThreadPoolExecutor

def Calculation(value, count):		# 创建一个处理函数
if value == 1:
print("pid:%s count:%d" %((os.getpid(), count)))
return count
if value % 2 == 0:
count += 1
value = value / 2
time.sleep(1)				# 模拟执行时间
else:
value = value * 3 + 1
return Calculation(value, count)

if __name__ == '__main__':
# ThreadPoolExecutor类支持with上下文管理，因此可以省略shutdown()方法
with ThreadPoolExecutor(4) as T_Pool:
T_List = []
count = 0
for i in range(10):
ret = T_Pool.submit(Calculation, random.randint(i+1, i+10), count)
# 如果想实现同步线程池，可用下述方法，返回的就直接就是线程的执行结果
# ret = T_Pool.submit(Calculation, random.randint(i+1, i+10), count).result()
T_List.append(ret)
# 同Pool类获取返回值的方法不同，这里使用result()方法，而不是get()方法
print([ret.result() for ret in T_List])

ThreadPoolExecutor实现线程池的回调函数

from concurrent.futures import ThreadPoolExecutor
import os,requests

def get_page(url):				# 定义一个获取网页的函数
print('<进程%s> get %s' %(os.getpid(), url))
respone = requests.get(url)		# 获取网页
if respone.status_code == 200:	# 查看获取状态
return {'url':url,'text':respone.text}	# 返回获取的内容

def pasrse_page(res):		# 定义处理网页的函数
res = res.result()      # 注意：此时接受的res是Future类型的对象，通过result() 方法拿到值
print('<进程%s> parse %s' %(os.getpid(), res['url']))
parse_res = 'url:<%s> size:[%s]\n' %(res['url'], len(res['text']))
with open('db.txt','a') as f:
f.write(parse_res)		# 保存至文件

if __name__ == '__main__':

urls=[
'https://www.baidu.com',
'https://www.python.org',
'https://www.openstack.org',
'https://help.github.com/',
'http://www.sina.com.cn/'
]

with ThreadPoolExecutor(4) as T_Pool:
T_List = []
for url in urls:
# 注意：add_done_callback()方法接受线程返回的Future类型的对象，而不线程的结果值
res = T_Pool.submit(get_page, url).add_done_callback(pasrse_page)
T_List.append(res)