python —— 池

Posted 2022-04-26 yangjie0906

1. 池

池分为：进程池、线程池

池：预先的开启固定个数的进程数/线程数，当任务来临的时候，直接提交给已经开好的进程 / 线程，让这个进程 / 线程去执行就可以了。

池节省了进程、线程的开启、关闭、切换需要的时间，并且减轻了操作系统调度的负担。

concurrent.futures模块中：ProcessPoolExcutor类（进程池）、ThreadPoolExcutor类（线程池）

1.1 进程池

进程池缺点：

• 开销大
• 一个池中的任务个数限制了我们程序的并发个数

# 没有参数和返回值
import os
import time
import random
from concurrent.futures import ProcessPoolExecutor
# submit + shutdown
def func():
    print('start',os.getpid())
    time.sleep(random.randint(1,3))
    print('end', os.getpid())
if __name__ == '__main__':
    p = ProcessPoolExecutor(5)
    for i in range(10):
        p.submit(func)
    p.shutdown()   # 关闭池之后就不能继续提交任务，并且会阻塞，直到已经提交的任务完成
    print('main',os.getpid())

# 任务的参数 + 返回值
def func(i,name):
    print('start',os.getpid())
    time.sleep(random.randint(1,3))
    print('end', os.getpid())
    return '%s * %s'%(i,os.getpid())
if __name__ == '__main__':
    p = ProcessPoolExecutor(5)
    ret_l = []
    for i in range(10):
        ret = p.submit(func,i,'alex')
        ret_l.append(ret)
    for ret in ret_l:
        print('ret-->',ret.result())  # ret.result() 同步阻塞
    print('main',os.getpid())

1.2 线程池

# 示例
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print('start', os.getpid())
    time.sleep(random.randint(1,3))
    print('end', os.getpid())
    return '%s * %s'%(i,os.getpid())
tp = ThreadPoolExecutor(20)

# 方法一：
ret = tp.map(func,range(20))  # a
for i in ret:  # b
    print(i)  # c
    
# a,b,c三行 相当于 下面1，2，3，4，5，6六行

# 方法二：
# ret_l = []  # 1
# for i in range(20):  # 2
#     ret = tp.submit(func,i)  # 3
#     ret_l.append(ret)  # 4
tp.shutdown()
print('main')
# for ret in ret_l:  # 5
#     print(ret.result())  # 6

1..3 回调函数

对象.add_done_callback（子线程执行完毕之后要执行的代码对应的函数名）

效率高

执行完子线程任务之后直接调用对应的回调函数

爬取网页：需要等待数据传输和网络上的响应高IO操作的 — 交子线程完成

分析网页：没有什么IO操作 — 这个操作没必要在子线程完成，交给回调函数完成

add_done_callback
# 示例：爬虫
import requests
from concurrent.futures import ThreadPoolExecutor
def get_page(url):
    res = requests.get(url)
    return 'url':url,'content':res.text

def parserpage(ret):
    dic = ret.result()
    print(dic['url'])
tp = ThreadPoolExecutor(5)
url_lst = [
    'http://www.baidu.com',
    'http://www.cnblogs.com', 
    'http://www.douban.com', 
    'http://www.tencent.com',
    'http://www.xinhuanet.com/',
    'https://www.toutiao.com/',
]
ret_l = []
for url in url_lst:
    ret = tp.submit(get_page,url)
    ret_l.append(ret)
    ret.add_done_callback(parserpage)

1.4 总结

ThreadPoolExcutor类
ProcessPoolExcutor类

创建一个池子
tp = ThreadPoolExcutor(池中线程（CPU个数*5）/进程（CPU个数）的个数)
异步提交任务
ret = tp.submit(需要在子线程执行的函数名,参数1，参数2....)
获取返回值
ret.result() 是一个阻塞方法
在异步的执行完所有任务之后，主线程/主进程才开始执行的代码
tp.shutdown() 阻塞 直到所有的任务都执行完毕
map方法
ret = tp.map(需要在子线程执行的函数名（如：func）,iterable) 迭代获取iterable中的内容，作为func的参数，让子线程来执行对应的任务
for i in ret: 每一个都是任务的返回值
绑定回调函数
ret.add_done_callback(子线程执行完毕之后要执行的代码对应的函数名)
要在ret对应的任务执行完毕之后，直接继续执行add_done_callback绑定的函数中的内容，并且ret的结果会作为参数返回给绑定的函数

1.做一些操作时是单独开启线程、进程还是池？

• 1.如果只是开启一个子线程做一件事情，就可以单独开线程
• 2.有大量的任务等待程序去做，要达到一定的并发数，就开启线程池
• 3.根据你程序的io操作也可以判定是用池还是不用池？
  • socket的server端：大量的阻塞io —recv、recvfrom、socketserver —— 不用池。
  • 爬虫的时候 —— 用池

2.进程 和 线程都有锁：

• 所有在线程中能工作的基本都不能在进程中工作
• 在进程中能够使用的基本在线程中也可以使用