进程池和线程池

一、什么是池

• 无论是开设进程还是开设线程，都需要消耗资源

• 只不过开始线程消耗的资源比开始进程相对少一些

• 硬件的开发速度永远赶不上软件开发速度

• 我们的宗旨是保证计算机硬件正常工作的情况下最大程度的利用它

• 池是用来保证计算机硬件安全的情况下最大限度的利用计算机

• 池降低了程序的运行效率，但是保证了计算机硬件的安全，从而保证程序的正常运行

• 池是用来保证计算机硬件安全的情况下最大限度的利用计算机

• 池降低了程序的运行效率，但是保证了计算机硬件的安全，从而保证程序的正常运行

二、进程池

• 在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。
  • 多进程是实现并发的手段之一，需要注意的问题是：
    • 很明显需要并发执行的任务通常要远大于核数
    • 一个操作系统不可能无限开启进程，通常有几个核就开几个进程
    • 进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行）
  • 例如当被操作对象数目不大时
    • 可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个。。。手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。
• 我们就可以通过维护一个进程池来控制进程数目
  • 比如httpd的进程模式，规定最小进程数和最大进程数…

PS：对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。

创建进程池的类：如果指定进程总数为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务，不会开启其他进程

[1]语法

from concurrent.futures import ProcessPoolExecutor

# 可以指定进程总数
pool = ProcessPoolExecutor(5)

pool.submit(进程要执行函数，要传入的参数)

[2]原理

• 池子造出来后 里面固定存在五个进程
• 这五个进程不会存在出现重复创建和销毁的过程

[3]优点

• 避免了重复创建五个进程的资源开销

[4]使用方法

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time


def p_run(i):
    print(f'进程{i}开始！')
    time.sleep(1)
    print(f'进程{i}结束！')


def p_main():
    p_pool = ProcessPoolExecutor(5)
    for i in range(1, 10):
        p_pool.submit(p_run, i)


if __name__ == '__main__':
    p_main()
# 最多同时运行5个进程，进程6想要运行必须等待前面的进程结束
"""
进程1开始！
进程2开始！
进程3开始！
进程4开始！
进程5开始！
进程1结束！
进程6开始！
进程3结束！
进程2结束！
进程7开始！
进程8开始！
进程5结束！
进程4结束！
进程9开始！
进程6结束！
进程8结束！进程7结束！

进程9结束！
"""

[5]异步返回值

（1）查看异步回调对象和返回值

• 如果直接打印返回对象，是异步，如果对返回对象进行进一步处理就会变成同步

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time


def p_run(i):
    print(f'进程{i}开始！')
    time.sleep(1)
    print(f'进程{i}结束！')
    return i ** 2


def p_main():
    p_pool = ProcessPoolExecutor(5)
    for i in range(1, 10):
        res = p_pool.submit(p_run, i)
        # 返回值对象
        print(f'这是pool的返回对象：{res}')
        
if __name__ == '__main__':
    p_main()
    
 """
这是pool的返回对象：<Future at 0x207361e07c0 state=running>
这是pool的返回对象：<Future at 0x207361e27a0 state=running>
这是pool的返回对象：<Future at 0x207361e2a70 state=pending>
这是pool的返回对象：<Future at 0x207361e2b90 state=pending>
这是pool的返回对象：<Future at 0x207361e2920 state=pending>
这是pool的返回对象：<Future at 0x207361e2ef0 state=pending>
这是pool的返回对象：<Future at 0x207361e2e60 state=pending>
这是pool的返回对象：<Future at 0x207361e2fe0 state=pending>
这是pool的返回对象：<Future at 0x207361e30a0 state=pending>
进程1开始！
进程2开始！
进程3开始！
进程4开始！
进程5开始！
进程1结束！
进程6开始！
进程2结束！
进程7开始！
进程4结束！
进程3结束！
进程8开始！
进程9开始！
进程5结束！
进程6结束！
进程7结束！
进程9结束！进程8结束！
"""

(2)直接在循环中打印返回对象并获取返回值然后打印会变成同步

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time


def p_run(i):
    print(f'进程{i}开始！')
    time.sleep(1)
    print(f'进程{i}结束！')
    return i ** 2


def p_main():
    p_pool = ProcessPoolExecutor(5)
    for i in range(1, 10):
        res = p_pool.submit(p_run, i)
        # 返回值对象
        print(f'这是pool的返回对象：{res}')
        # 返回值
        print(f'这是pool的返回值：{res.result()}')4
        
if __name__ == '__main__':
    p_main()
"""
这是pool的返回对象：<Future at 0x15308d00a90 state=running>
进程1开始！
进程1结束！
这是pool的返回值：1
这是pool的返回对象：<Future at 0x15308d00c10 state=pending>
进程2开始！
进程2结束！
这是pool的返回值：4
这是pool的返回对象：<Future at 0x15308d01b40 state=pending>
进程3开始！
进程3结束！
这是pool的返回值：9
这是pool的返回对象：<Future at 0x15308d02b30 state=pending>
.............
"""   

（3）将返回值的获取和打印与打印返回对象分开就又回到异步

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time

def t_run(i):
    print(f'线程{i}开始！')
    time.sleep(1)
    print(f'线程{i}结束！')
    return i ** 2

def p_main():
    res_list = []
    p_pool = ProcessPoolExecutor(5)
    for i in range(1, 10):
        res = p_pool.submit(p_run, i)
        res_list.append(res)
    for res in res_list:
        print(f'这是p_poll的返回值:{res.result()}')
if __name__ == '__main__':
    p_main()
    
"""
进程1开始！
进程2开始！
进程3开始！
进程4开始！
进程5开始！
进程1结束！
进程6开始！
这是p_poll的返回值:1
进程4结束！进程3结束！
进程2结束！

进程7开始！
进程8开始！
进程9开始！
这是p_poll的返回值:4
这是p_poll的返回值:9
这是p_poll的返回值:16
进程5结束！
这是p_poll的返回值:25
进程6结束！
这是p_poll的返回值:36
进程7结束！进程8结束！进程9结束！


这是p_poll的返回值:49
这是p_poll的返回值:64
这是p_poll的返回值:81
"""

[6]pool.shutdown()

• 等待所有子线程结束后在打印程序运行结果

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time

def t_run(i):
    print(f'线程{i}开始！')
    time.sleep(1)
    print(f'线程{i}结束！')
    return i ** 2

def p_main():
    res_list = []
    p_pool = ProcessPoolExecutor(5)
    for i in range(1, 10):
        res = p_pool.submit(p_run, i)
        res_list.append(res)
    p_pool.shutdown()
    for res in res_list:
        print(f'这是p_poll的返回值:{res.result()}')
if __name__ == '__main__':
    p_main()
"""
进程1开始！
进程2开始！
进程3开始！
进程4开始！
进程5开始！
进程1结束！
进程6开始！
进程3结束！
进程2结束！
进程7开始！
进程8开始！
进程4结束！进程5结束！

进程9开始！
进程6结束！
进程7结束！
进程8结束！
进程9结束！
这是p_poll的返回值:1
这是p_poll的返回值:4
这是p_poll的返回值:9
这是p_poll的返回值:16
这是p_poll的返回值:25
这是p_poll的返回值:36
这是p_poll的返回值:49
这是p_poll的返回值:64
这是p_poll的返回值:81
"""

[7]异步回调函数

• add_done_callback(将返回值进行下一步处理的函数) 会将进程池中的进程的返回值交由括号内的函数进行再一次处理（不会有返回值）
• 原来的返回值也会因为回调而变为None

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time

def t_run(i):
    print(f'线程{i}开始！')
    time.sleep(1)
    print(f'线程{i}结束！')
    return i ** 2


def write_results(data):
    with open(r"D:\project\python\project_study\网络并发\并发编程\results_data.txt", 'a', 					  encoding='utf-8') as fp:
        fp.write(str(data.result()) + '\n')
 

def p_main():
    res_list = []
    p_pool = ProcessPoolExecutor(5)
    for i in range(1, 10):
        res = p_pool.submit(p_run, i).add_done_callback(write_results)
        res_list.append(res)
    p_pool.shutdown()
    for res in res_list:
        print(f'这是p_poll的返回对象:{res()}')
if __name__ == '__main__':
    p_main()
    
"""
进程1开始！
进程2开始！
进程3开始！
进程4开始！
进程5开始！
进程1结束！
进程6开始！
进程3结束！
进程4结束！进程2结束！
进程5结束！

进程7开始！
进程8开始！
进程9开始！
进程6结束！
进程8结束！
进程9结束！进程7结束！

这是poll的返回对象:None
这是poll的返回对象:None
这是poll的返回对象:None
这是poll的返回对象:None
这是poll的返回对象:None
这是poll的返回对象:None
这是poll的返回对象:None
这是poll的返回对象:None
这是poll的返回对象:None
"""