8.15并发编程（四）

时间：2019-08-16 23:54:29 阅读：164 评论：0 收藏：0 [点我收藏+]

一、进程池和线程池

1.什么是池

　　池可以理解是一种容器，它其实是降低了程序的运行效率但是提高了计算机的硬件安全，因为硬件的发展跟不上软件的发展速度

2.池的作用是什么

　　池的作用就是在保证计算机硬件安全的前提下，最大限度的利用计算机

3.进程池和线程池

　　进程池：我们所允许创建的最大进程数

　　线程池：我们在一个进程内所允许创建的最大的线程数

4.注意：

　　开进程和开线程都会消耗资源，但是两者对比来说开线程所消耗的资源最小，开销最小

　　进程池和线程池的本质都是在计算机能够承受的范围内最大限度的利用计算机

二、进程池、线程池的创建和异步回调

1.使用方法

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

#创建池
pool = ThreadPoolExecutor(5)  # 5 线程个数，也可以不传，默认是CPU个数乘以5
#pool = ProcessPoolExecutor()
# 进程池不传参数，默认是当前cpu的个数

def task(n):
    print(n)
    time.sleep()
    
#提交任务:异步提交
# pool.submit(task, 1)  朝线程池提交任务
#回调机制：pool.submit(task, 1).add_done_callback(callback回调函数)

# 定义一个列表，实现线程的异步提交的并发
t_list = []
# 线程池只有五个，所以五个五个的接，类似缓冲区域
for i in range(20):
    res = pool.submit(task, i) # res 是任务 task 的返回值
    print(res.result())  # 原地等待任务返回结果，将并发变成了串行。
    t_list.append(res)
   
# 需求，让20个线程运行结束，再拿结果
pool.shutdown()  # 关闭池子，等待所有任务完成之后，才会往下运行。
for t in t_list:
    print(‘>>>‘, t.result())

2.线程池用法

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

pool = ThreadPoolExecutor(5)  # 默认线程数是cpu * 5

def task(n):
    print(n)
    time.sleep(3)
    return n**2     # 给task函数添加返回结果
‘‘‘
引出提交任务的方式：
同步：提交任务之后，原地等待任务的返回结果，期间不做任何事
异步：提交任务之后，不等待任务的返回结果（但是结果会以另一种方式获得。。。），直接运行下一行代码
‘‘‘
# pool.submit(task,1)  # 向池子里面提交任务  异步提交  参数：任务函数，及其所需参数
# print(‘主。。‘)  # 由结果可知，pool.submit(task,1)  是异步提交的形式。

‘‘‘
思考：异步提交任务的结果要怎么拿？

‘‘‘
# 多创建几个线程来看看效果
# for i in range(20):
#     pool.submit(task,i)  # 思考：我们怎么拿到异步提交任务的返回结果？
‘‘‘
首先我们尝试看一下 submit 源码 看到submit 会有一个返回值 f(是一类的对象)
‘‘‘
# for i in range(20):
#     res =pool.submit(task,i)
#     # print(res)  # 打印看看返回的对象
#     print(res.result())  # 我们打印对象res 结果，用方法 .result()
‘‘‘
执行结果：返回结果是 None,且打印结果是串行的形式
1 串行原因：
print(res.result())   这句代码在for循环中表示：原地等待任务结果，变成同步提价任务，将并发变成串行了
2 返回结果为None 
因为同步提交任务，我们要原地等待任务的返回结果，任务就是task函数，但是函数task没有return结果，所以就是None

‘‘‘
"""
思考：
以上代码我们虽然拿到了返回结果但是却变成同步提交的形式，
我们如何实现并发将20 个线程全部启动，再拿到他们的结果？
解决方法：
先将20个线程全部启动，并放进一个 列表里面，我们在循环这个列表，

"""
# 将上述代码进行修改
t_list = []
for i in range(20):
    res =pool.submit(task,i)
    # print(res)  # 打印看看返回的对象
    # print(res.result())  # 我们打印对象res 结果，用方法 .result()
    t_list.append(res)

# for t in t_list:
#     print(‘>>>:‘,t.result())  # 返回的结果是有顺序的，因为我们将线程添加到list里面是有顺序添加的，所以for循环打印是有序的

"""
思考：以上代码的执行结果 线程和返回结果都是错乱的，
我们希望所有的任务等待池子里的线程全部执行完毕，再获取结果
方法：pool.shutdown()
"""
# 修改以上代码
pool.shutdown()  # 关闭池子，等待池子里的任务全部执行完毕再往下走
for t in t_list:
    print(‘>>>:‘,t.result())

　　以上代码是通过线程池实现并发，并拿到异步提交任务的结果

3.进程池的用法和异步回调机制

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

pool = ProcessPoolExecutor(5)  # 创建进程池
"""
理论：
池子中创建的进程/线程创建一次就不会再创建了
至始至终用的都是最初的那几个
这样的话节省了反复开辟进程/线程的资源
"""
def task(n):
    print(n,os.getpid())  # 通过os.getpid()来获得进程id 验证理论
    time.sleep(3)
    return n**2
"""
怎样拿到异步提交任务的结果？
异步回调机制：当异步提交的任务有结果之后，会自动触发回调函数的执行
"""
def call_back(n):
    print(‘异步回调结果：‘,n.result())  # 这里的n 是task 返回结果的个数
if __name__ == ‘__main__‘:
    for i in range(20):
        # 给提交的每一个任务绑定回调函数，当任务有结果的时候，会执行对应的回调函数
        res = pool.submit(task,i).add_done_callback(call_back)

技术分享图片

三、异步回调机制

1.异步回调机制

　　异步回调机制就是我们定义一个回调函数，当异步提交的任务有返回结果时会自动触发回调函数的执行

2.异步回调机制的作用

　　获得异步提交任务的结果

四、协程

进程：资源单位

线程：执行单位

协程：单线程下实现并发

并发：

　　切换 + 保存状态

　　PS：看起来像同时执行的就可以称之为并发

协程：完全是程序员自己意淫出来的名词

　　单线程下实现并发

并发的条件：

　　多道技术

　　空间上的复用

　　时间上的复用

　　切换 + 保存状态

1.什么是协程

　　协程就是在单线程下实现并发

2.实现并发的条件是什么

　　多道技术

　　空间上的复用和时间上的复用

　　其中时间上的复用，即：切换 + 保存状态

3.因此我们可以这样来理解协程

　　程序员自己通过代码自己检测程序中的IO

　　一旦遇到IO自己通过代码切换

　　给操作系统的感觉是你这个线程没有任何的IO

　　PS：欺骗操作系统让它误认为你这个程序一直没有IO

　　　　从而保证程序在运行态和就绪态来回切换

　　　　提升代码的运行效率

4.切换 + 保存状态就一定能提高效率吗

　　当你的任务是io密集型：提高效率

　　当你的任务是计算密集型：降低效率

5.协程如何实现并发，即协程如何实现：切换 + 保存状态

　　保存状态：yield 保存上一次的结果，保存状态

　　切换：碰到 I/O操作切换状态

"""
需要找到一个能够识别IO的一个工具
gevent模块
"""
from gevent import monkey;monkey.patch_all()  # 由于该模块经常被使用 所以建议写成一行
from gevent import spawn
import time
"""
注意gevent模块没办法自动识别time.sleep等IO情况
需要你手动再配置一个参数
"""

　　代码示例协程的状态切换

from gevent import monkey; monkey.patch_all()  # 为了识别time.sleep() 手动添加参数
from gevent import spawn
import time

def sing():
    print(‘唱歌‘)
    time.sleep(2)
    print(‘唱歌‘)

def jump():
    print(‘跳舞‘)
    time.sleep(3)
    print(‘跳舞‘)

def rap():
    print(‘说唱‘)
    time.sleep(5)
    print(‘说唱‘)

start = time.time()
g1 = spawn(sing)  # spawn() 的作用是传入一个函数名，它自动帮你加括号调用
g2 = spawn(jump)
g3 = spawn(rap)
g1.join()
g2.join()
g3.join()
print(time.time() - start)
"""
协程一直在这3个函数之间切换状态，碰到I/O操作就切换去执行别的程序
"""

五、协程实现TCP服务端并发

1.服务端

from gevent import monkey;monkey.patch_all()
import socket
from gevent import spawn

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

def talk(conn):
    while True:
        try:
            data = conn.recv(1024)
            if len(data) == 0:break
            print(data.decode(‘utf-8‘))
            conn.send(data.upper())
        except ConnectionResetError as e:
            print(e)
            break
    conn.close()

def server1():
    while True:
        conn, addr = server.accept()
        spawn(talk,conn)

if __name__ == ‘__main__‘:
    g1 = spawn(server1)
    g1.join()

2.客户端

import socket
from threading import Thread,current_thread

def client():
    client = socket.socket()
    client.connect((‘127.0.0.1‘,8080))
    n = 0
    while True:
        data = ‘%s %s‘%(current_thread().name,n)
        client.send(data.encode(‘utf-8‘))
        res = client.recv(1024)
        print(res.decode(‘utf-8‘))
        n += 1

for i in range(400):
    t = Thread(target=client)
    t.start()