进程池、线程池、协程

进程池与线程池

硬件是有极限的，我们不可能一直在一台计算机上无限的创建新的进程和线程，虽然软件逻辑上我们可以无限的创建，但是一旦这么做了，我们的计算机可能到达承受不了的极限，最后崩溃到无法执行任何程序的地步。

所以为了限制进程和线程的无限创建（如我们写过的来一个客户端就开个进程去服务他），有了池的概念。

• 进程池/线程池：提前创建好固定数量的进程/线程供后续程序的调用，超出数量则等待

代码实现进程池和线程池

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import os
import time
import random

# 1.产生含有固定数量线程的线程池
# pool = ThreadPoolExecutor(10)
pool = ProcessPoolExecutor(5)  # 产生进程池
# 查看源码可以看见这个类可以不写数字产生对象，这个数字是池的容量
# 如果不指定就根据cpu的参数来自动获取对应合适的数量


def task(n):
    print('task is running')
    time.sleep(random.randint(1, 3))

if __name__ == '__main__':
    # 2.将任务提交给线程池即可
    for i in range(20):
        pool.submit(task, 123)  # 朝线程池提交任务
        
"""
运行的结果会成批的打印，按照原本的开进程线程的方式，相当于同时开启了20个进程或线程
但现在当提交的任务达到进程或线程达到池的上限，新的任务就会等待某个任务线程执行完再去争夺线程\进程
"""

回调机制

回调机制指当提交到进程\线程池的任务结束时可以触发另一个函数的运行，并且将自己的任务对象传入函数以便它进行处理，如可以通过任务对象.result()的方式拿到任务函数的返回值。

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import os
import time
import random

# 1.产生含有固定数量线程的线程池
# pool = ThreadPoolExecutor(10)
pool = ProcessPoolExecutor(5)


def task(n):
    print('task is running')
    time.sleep(random.randint(1, 3))
    return '我是task函数的返回值'  # 拿到返回值


def func(self):
    print('from func', self.result())  # 在这里仍然异步的处理返回值。

if __name__ == '__main__':
    # 2.将任务提交给线程池即可
    for i in range(20):
        res = pool.submit(task, 123)  # 朝线程池提交任务，返回任务对象
        # print(res.result())  # 不能直接获取，相当于变成串行
        res.add_done_callback(func)  # 提交的任务可以调用这个函数来触发回调，当任务结束时调用func函数

协程

是独立于操作系统的进程线程外开发的实现并发的方式，也就是说可以通过单线程实现并发。

我们知道进程和线程是通过碰到IO时切换加保存程序状态的方式实现异步的，现在我们让程序帮我们检测IO，当碰到IO时就去做当前线程的其他代码，当那个IO结束时就处理相应的代码段，这个过程也一样实现了异步并发。这样做让操作系统永远不知道我们进入了IO，也可以提升我们程序对CPU的占用率。

协程代码实现

import time
from gevent import monkey;

monkey.patch_all()  # 固定编写 用于检测所有的IO操作(猴子补丁)
from gevent import spawn


def func1():
    print('func1 running')
    time.sleep(3)
    print('func1 over')


def func2():
    print('func2 running')
    time.sleep(5)
    print('func2 over')


if __name__ == '__main__':
    start_time = time.time()
    # func1()
    # func2()
    s1 = spawn(func1)  # 检测代码 一旦有IO自动切换(执行没有io的操作 变向的等待io结束)
    s2 = spawn(func2)
    s1.join()
    s2.join()
    print(time.time() - start_time)  # 串行8.01237154006958   协程 5.015487432479858

协程实现并发服务端

import socket
from gevent import monkey;monkey.patch_all()  # 固定编写 用于检测所有的IO操作(猴子补丁)
from gevent import spawn


def communication(sock):
    while True:
        data = sock.recv(1024)  # IO操作
        print(data.decode('utf8'))
        sock.send(data.upper())


def get_server():
    server = socket.socket()
    server.bind(('127.0.0.1', 8080))
    server.listen(5)
    while True:
        sock, addr = server.accept()  # IO操作
        spawn(communication, sock)  # 异步的接收用户输入

s1 = spawn(get_server)  # 创建socket套接字
s1.join()

上述代码中，如果任务中含IO操作，则就可以将其传入spwan让其进行协程处理。

ps:如果我们想要尽可能的提高程序的并发量，则要多开进程，进程下开多线程，线程下开协程。