~~并发编程（十五）：进程池线程池~~

进击のpython

*****

并发编程——进程池线程池

---

本小节是对进程池线程池做一个了解同时也对回调函数有一个清晰的认识最后再提一下异步与同步

但是在提到这三个知识点之前,我们有必要基于线程的知识点进行一个练习

目的是为了能够对以前的知识有个印象,对于接下来学这两个知识点也有好处

---

练习

还是选择套接字来进行练习,在上一部分的时候,基于套接字,我们没有做到并发的概念

那么现在我们基于多线程来做一个并发的套接字通信:

简单写一个套接字通信:

# 服务端
import socket

server = socket.socket()
server.bind(('127.0.0.1', 8080))
server.listen(5)

while 1:
    conn, address = server.accept()
    while 1:
        try:
            msg = conn.recv(1024)
            conn.send(msg.upper())
        except:
            break
conn.close()
server.close()

# 客户端
import socket

client = socket.socket()
client.connect(('127.0.0.1', 8080))

while 1:
    msg = input()
    client.send(msg.encode('utf-8'))
    print(client.recv(1024).decode('utf-8'))

conn.close()

client.close()

然后我们尝试着用多线程进行操作

import socket
from threading import Thread


def A():
    server.bind(('127.0.0.1', 8080))
    server.listen(5)
    while 1:
        conn, address = server.accept()
        t = Thread(target=B, args=(conn,))
        t.start()


def B(conn):
    while 1:
        try:
            msg = conn.recv(1024)
            conn.send(msg.upper())
        except:
            break
    conn.close()
    server.close()


if __name__ == '__main__':
    server = socket.socket()
    A()

很好,问题随之就来了!

---

线程池进程池

根据上面的代码,我们看,没什么问题

但是随着客户端开的越来越多,就会导致客户端的线程开的越来越多

要是成千上万个呢?是不是电脑就死机了

所以说我们应该有一个办法,让服务端开启的线程控制在一个范围内,使电脑稳定运行

线程池进程池,我们还是可以镜像来学,学一个,剩下的就会了

线程池开启借助模块里的的一个方法ThreadPoolExecutor

使用方法都是一样的,毕竟python遵循鸭子类型

先进行实例化,参数是最大可开的线程池pool = ThreadPoolExecutor(3)

接着提交就不再是用Thread了,而是实例化对象的submit方法,第一个参数是函数,第二个是传参变量

示例如下：

import os
import random
import time
from concurrent.futures.thread import ThreadPoolExecutor
from threading import current_thread


def func():
    print('%s is runing' % current_thread().ident)
    time.sleep(random.randint(1, 3))


if __name__ == '__main__':
    pool = ThreadPoolExecutor(3)
    for i in range(10):
        t = pool.submit(func)
    print('主')

注意调用的模块concurrent.futures.thread

在线程中有join方法，其实在线程池也有一个类似于join的方法

shutdown()里面有一个wait的默认参数True，代表着等待子线程结束的

你可以试着运行一下下面的代码，再把注释打开，看看是什么样的执行结果

import os
import random
import time
from concurrent.futures.thread import ThreadPoolExecutor
from threading import current_thread


def func():
    print('%s is runing' % current_thread().ident)
    time.sleep(random.randint(1, 3))


if __name__ == '__main__':
    pool = ThreadPoolExecutor(3)
    for i in range(10):
        t = pool.submit(func)
    # pool.shutdown()
    print('主')

进程池的使用和线程池是一样的，除了ThreadPoolExecutor→ProcessPoolExecutor

那什么时候用进程池，什么时候用线程池呢？

其实就是跟什么时候用进程什么时候用线程一样

当I/O密集的时候，用线程池；当计算密集的时候，用进程池

多提一嘴，在代码的这部分：

 for i in range(10):
        t = pool.submit(func)

如果觉得麻烦，可以用

map(func,range(10))

但是这里有个陷阱，func函数也需要有一个接受值，可以不用，但是得有

import random
import time
from concurrent.futures.thread import ThreadPoolExecutor
from threading import current_thread


def func(n): # 注意这个n
    print('%s is runing' % current_thread().ident)
    time.sleep(random.randint(1, 3))


if __name__ == '__main__':
    pool = ThreadPoolExecutor(3)
    pool.map(func, range(5))
    print('主')

---

异步调用

跟异步调用的概念相对的还有同步调用这个概念

同步调用和异步调用其实是提交任务的两种方式

假设我有一筐包子，然后以一帮人来吃，同步和异步就是两种操作

同步调用

同步的特点是执行完任务之后就在原地等待，等待程序执行完毕，然后拿到结果在执行下一个任务：

import random
import time
from concurrent.futures.thread import ThreadPoolExecutor


def baozi(name):
    time.sleep(random.randint(1, 3))
    return {"name": name}


def zhanshi(res):
    name = res["name"]
    print(f'{name}吃了包子！')


if '__main__' == __name__:
    t = ThreadPoolExecutor(3)

    t1 = t.submit(baozi, '可乐').result()
    zhanshi(t1)
    t2 = t.submit(baozi, '雪碧').result()
    zhanshi(t2)
    t3 = t.submit(baozi, '柠檬茶').result()
    zhanshi(t3)

所以这个吃包子就是这样的：可乐过来吃，吃完了，雪碧过来吃，吃完了，柠檬茶，吃完了，程序结束

就可以发现同步调用，其实就变成串行了

异步调用

异步的特点是执行完任务之后不再等待：

import random
import time
from concurrent.futures.thread import ThreadPoolExecutor


def baozi(name):
    time.sleep(random.randint(1, 3))
    return {"name": name}


def zhanshi(res):
    name = res["name"]
    print(f'{name}吃了包子！')


if '__main__' == __name__:
    t = ThreadPoolExecutor(3)

    t1 = t.submit(baozi, '可乐')
    t2 = t.submit(baozi, '雪碧')
    t3 = t.submit(baozi, '柠檬茶')

执行了，也开了线程，但是拿不到结果，线程执行完才会有结果！那谁能知道线程执行是否完毕？

线程本身会知道！进程执行到了最后一行，就代表着线程执行完毕

此时就应该让他去执行zhanshi()这个函数

那要怎么做呢？可能会有这样的想法：我直接传过去就可以了

import random
import time
from concurrent.futures.thread import ThreadPoolExecutor


def baozi(name):
    time.sleep(random.randint(1, 3))
    zhanshi({"name": name}) # 改动在这里


def zhanshi(res):
    name = res["name"]
    print(f'{name}吃了包子！')


if '__main__' == __name__:
    t = ThreadPoolExecutor(3)

    t1 = t.submit(baozi, '可乐')
    t2 = t.submit(baozi, '雪碧')
    t3 = t.submit(baozi, '柠檬茶')

这么做是没有问题，但是在开发的考虑上来说，这两个方法的耦合性增强了

不是说耦合增强一定不好，但是，能解耦写就解耦写

所以说我们应该有其他的办法来解决这个问题：回调函数！

回调函数就是在线程执行完毕后自动执行另一个函数的方法add_done_callback()

import random
import time
from concurrent.futures.thread import ThreadPoolExecutor


def baozi(name):
    time.sleep(random.randint(1, 3))
    return {"name": name}


def zhanshi(res):
    res = res.result()
    name = res["name"]
    print(f'{name}吃了包子！')


if '__main__' == __name__:
    t = ThreadPoolExecutor(3)

    t1 = t.submit(baozi, '可乐').add_done_callback(zhanshi)
    t2 = t.submit(baozi, '雪碧').add_done_callback(zhanshi)
    t3 = t.submit(baozi, '柠檬茶').add_done_callback(zhanshi)

当线程执行完了，自动执行add_done_callback()里面的函数，并将结果的对象传给这个函数

注意！传的是对象！而不是结果！，所以才会在函数里面有一句res = res.result()

---

单列出来一块是为了解决一个问题：有的同学认为同步调用就是阻塞

首先你要知道什么是阻塞？I/O阻塞是吧，程序遇到I/O就会阻塞

那会不会有这种情况，程序的运行的是纯计算的，这是不是就不存在I/O阻塞了？

但是同步提交任务，该等结果，还是得等结果！这就不是阻塞了

所以这种说法是不正确的！

---

***** *****