0x0 多任务编程
意义:充分利用计算机多核资源,提高程序的运行效率。
实现方案:多进程,多线程
并行与并发
并发:同时处理多个任务,内核在任务间不断的切换达到好像多个任务被同时执行的效果,实际每个时刻只有一个任务占用内核
注意:程序在运行中会有IO阻塞的存在,当一个程序被阻塞后会被立刻cpu踢出,等待阻塞结束后,再次排队等待cpu执行
并行:多个任务利用计算机多核资源在同时执行,此时多个任务间并行关系。
1×0 进程(process)
1×1 进程理论基础
定义:程序在计算机中的一次运行
程序是一个可执行的文件,是静态的占有磁盘
进程是一个动态的过程描述,占有计算机的运行资源,有一定的生命周期
系统中如何产生一个进程
- 用户控件通过调用程序接口或者命令发起请求
- 操作系统接受用户请求,开始创建进程
- 操作系统调配计算机资源,确定进程的状态
- 操作系统将创建的进程提供给用户使用
进程的基本概念
cpu时间片:如果一个进程占有cpu内核则称这个进程在cpu时间片上。
PCB(进程控制块):在内存中开辟的一块空间,用于存放进程的基本信息,也用于系统查找识别进程。
进程ID(PID): 系统为每个进程分配的一个大于0的整数,作为进程ID。每个进程ID不重复。 Linux查看进程ID : ps -aux
父子进程 : 系统中每一个进程(除了系统初始化进程)都有唯一的父进程,可以有0个或多个子进程。父子进程关系便于进程管理。
查看进程树: pstree
进程状态:
就绪态 | 进程具备执行条件,等待分配CPU资源 |
运行态 | 进程占有CPU时间片正在运行 |
等待态 | 进程暂时停止运行,让出cpu |
五态 (在三态基础上增加新建和终止)
新建 : 创建一个进程,获取资源的过程
终止 : 进程结束,释放资源的过程
状态查看命令:ps -aux –> STAT列
进程的运行特征
【1】 进程可以使用计算机多核资源
【2】 进程是计算机分配资源的最小单位
【3】 进程之间的运行互不影响,各自独立
【4】 每个进程拥有独立的空间,各自使用自己空间资源
面试要求
- 什么是进程,进程和程序有什么区别
- 进程有哪些状态,状态之间如何转化
2×0 基于fork的多进程编程
2×1 fork使用
语法:pid = os.fork()
- 功能:创建新的进程
- 返回值:整数,如果创建进程失败返回一个负数,如果成功则在原有进程中返回新进程的PID,在新进程中返回0
注意:
- 子进程会复制父进程全部内存空间,从fork下一句开始执行。
- 父子进程各自独立运行,运行顺序不一定。
- 利用父子进程fork返回值的区别,配合if结构让父子进程执行不同的内容几乎是固定搭配。
- 父子进程有各自特有特征比如PID PCB 命令集等。
- 父进程fork之前开辟的空间子进程同样拥有,父子进程对各自空间的操作不会相互影响。
""" fork函数演示 """ import os from time import sleep pid = os.fork() if pid < 0: print("Create process failed") elif pid == 0: sleep(5) print("The new process") else: sleep(6) print("The old process") print("Fork test over")
import os print("==============") a = 1 pid = os.fork() if pid < 0: print("Error") elif pid == 0: print("Child process") print("a = %d" %a) a = 10000 else: print("Parent process") print("a:",a) print("ALL a",a)
2×2 进程相关函数
os.getpid()
- 功能:获取一个进程的PID值
- 返回值:返回父进程的PID
os.getppid()
- 功能:获取父进程的PID号
- 返回值:返回父进程PID
os_exit([status])
- 功能:结束一个进程
- 参数:进程的终止状态,可以传任意数值,在没有约定的情况下没有意义,在约定的情况下就会拥有意义
- status会做一个乘以256的操作
sys.exit([status])
- 功能:退出进程
- 参数:整数 表示退出状态(为可选参数,不选默认为0)。可以写字符串,表示退出时打印的内容
- 父子进程的退出是相互独立的
2×3 孤儿和僵尸
孤儿进程
父进程先于子进程退出,此时子进程成为孤儿进程
特点:孤儿进程会被系统进程收养,此时系统进程就会成为孤儿进程新的父进程,孤儿进程退出该进程会自动处理。并不会被爷爷进程所接受
僵尸进程
子进程先于父进程退出,父进程又没有处理子进程的退出状态,此时子进程就会称为僵尸进程
特点:僵尸进程虽然结束,但是会存留部分PCB(进程信息)在内存中,大量的僵尸进程会浪费系统的内存资源。
孤儿进程不会变成僵尸进程,因为孤儿进程退出,该系统进程会自动处理
如何避免僵尸进程产生
使用wati函数处理子进程退出
""" wait处理僵尸进程 """ import os pid = os.fork() if pid < 0: print("Error") elif pid == 0: print("child process", os.getpid()) os._exit(2) # 这个参数会乘以256 else: # pid, status = os.wait() # 等待处理僵尸 pid, status = os.waitpid(-1, os.WNOHANG) # 非阻塞处理僵尸 print(pid) print(status, os.WEXITSTATUS(status)) while True: pass
创建二级子进程处理僵尸
- 父进程创建子进程,等待回收子进程
- 子进程创建二级子进程然后退出
- 二级子进程称为孤儿,和原本父进程一同执行事件
""" 创建二级子进程防止僵尸进程 """ import os from time import * def f1(): for i in range(4): sleep(2) print("写代码...") def f2(): for i in range(5): sleep(1) print("测代码...") pid = os.fork() if pid < 0: print("Error") elif pid == 0: p = os.fork() # 二级子进程 if p == 0: f2() # 二级子进程执行 else: os._exit(0) else: os.wait() f1()
通过信号处理子进程退出
原理:子进程退出时会发送信号给父进程,如果父进程忽略子进程信号,则系统就会自动处理子进程退出。
方法:使用signal模块在父进程创建子进程前写如下语句:
import signal signal.signal(signal.SIGCHLD,signal.SIG_IGN)
特点:非阻塞,不会影响父进程运行,可以处理所有子进程退出
2×4 群聊聊天室
功能 : 类似qq群功能
【1】 有人进入聊天室需要输入姓名,姓名不能重复
【2】 有人进入聊天室时,其他人会收到通知:xxx 进入了聊天室
【3】 一个人发消息,其他人会收到:xxx : xxxxxxxxxxx
【4】 有人退出聊天室,则其他人也会收到通知:xxx退出了聊天室
【5】 扩展功能:服务器可以向所有用户发送公告:管理员消息: xxxxxxxxx
思路分析
1.技术点的确认 转发模型,客户端-->服务端-->转发给其他客户端 网络模型,udp通信 保存用户信息:字典 收发关系处理:采用多进程分别进行收发操作 2.结构设计 采用什么样的封装结构:函数 编写一个功能,测试一个功能 注意注释和结构的设计
3.分析功能模块,指定具体编写流程 搭建网络连接 进入聊天室 客户端: 输入新名 将新名发送给服务端 接受返回的结果 如果不允许则重复输入姓名 服务端: 接受姓名 判断姓名是否存在 将结果返回给客户端 如果允许进入聊天室,增加用户信息 通知其他用户 聊天 客户端 创建新的进程 一个进程循环发送消息 一个进程循环接受消息 服务端 接受请求,判断请求类型 将消息转发给其他用户 退出聊天室 客户端 输入quit或者ctrl-c退出 将请求发送给服务端 结束进程 接收进程收到EXIT退出进程 服务端 接收消息 将突出消息告知其他人 给该用户发送"EXIT" 删除用户 管理员消息 4.协议 如果允许进入聊天室,那么服务端发送"OK"给客户端 如果不允许进入聊天室,则服务端发送"不允许的原因" 请求类别: L --> 进入聊天室 C --> 聊天信息 Q --> 退出聊天室 用户存储结构:{name:addr....} 客户端如果输入quit或者ctrl-c,点击esc表示退出
""" Chat room env:python3.7 socket fork 练习 """ from socket import * import os import sys # 服务器地址 BIND = ("0.0.0.0", 8888) # 用户列表 USER = {super: ("0.0.0.0", 8888)} # 聊天 def do_chat(s, name, text): msg = "%s : %s" % (name, text) for i in USER: s.sendto(msg.encode(), USER[i]) # 退出聊天室 def do_quit(s, name): msg = "%s退出了聊天室" % name print(msg) for i in USER: if i != name: s.sendto(msg.encode(), USER[i]) else: s.sendto("EXIT".encode(), USER[i]) # 删除用户 del USER[name] def do_login(s, name, addr): if name in USER or "管理员" in name: s.sendto("该用户已存在".encode(), addr) return s.sendto(b"OK", addr) # 通知其他人 msg = "欢迎%s进入聊天室" % name print(msg) for i in USER: if i != name: s.sendto(msg.encode(), USER[i]) # 将用户加入 USER[name] = addr def do_request(s): """ 接受数据函数 :param s: obj类型,socket类型 :return: """ while True: data, addr = s.recvfrom(1024) msg = data.decode().split(" ") # 区分请求类型 if msg[0] == "L": do_login(s, msg[1], addr) elif msg[0] == "C": do_chat(s, msg[1], " ".join(msg[2:])) elif msg[0] == "Q": do_quit(s, msg[1]) def main(): # 套接字 s = socket(AF_INET, SOCK_DGRAM) s.bind(BIND) pid = os.fork() if pid < 0: return elif pid == 0: while True: msg = input("管理员消息:") msg = "C super" + msg s.sendto(msg.encode(), BIND) else: # 请求处理 do_request(s) def udp_connect(): """ 建立udp连接选项 :return: object类型,socket对象 """ soc = socket(AF_INET, SOCK_DGRAM) soc.bind(BIND) if __name__ == '__main__': main()
""" chat room client """ from socket import * import os import sys # 服务器地址 BIND = ("0.0.0.0", 8888) # 发送消息 def send_msg(s, name): while True: try: text = input(">>") except KeyboardInterrupt: text = "quit" if text == "quit": msg = "Q " + name s.sendto(msg.encode(), BIND) sys.exit("退出聊天室") msg = "C %s %s" % (name, text) s.sendto(msg.encode(), BIND) # 接受消息 def recv_msg(s): while True: data, addr = s.recvfrom(2048) # 服务端发送EXIT表示让客户端退出 if data.decode() == "EXIT": sys.exit() print(data.decode()) # 创建网络连接 def main(): s = socket(AF_INET, SOCK_DGRAM) while True: name = input("请输入姓名") msg = "L " + name s.sendto(msg.encode(), BIND) data, addr = s.recvfrom(1024) if data.decode() == "OK": print("进入聊天室") break else: print(data.decode()) # 创建新的进程 pid = os.fork() if pid < 0: sys.exit("Error!") elif pid == 0: send_msg(s, name) else: recv_msg(s) if __name__ == "__main__": main()
3×0 multiprocessing模块创建进程
- 流程特点
【1】 将需要子进程执行的事件封装为函数
【2】 通过模块的Process类创建进程对象,关联函数
【3】 可以通过进程对象设置进程信息及属性
【4】 通过进程对象调用start启动进程
【5】 通过进程对象调用join回收进程 - 基本接口使用
Process(target,*args,*kwargs)
- args元组,用于给target函数位置传参
- kwargs字典,给target函数键值传参
p.start()
- 功能:启动进程
- 注意:启动进程时target绑定函数开始执行,该函数作为子进程执行内容,此时进程真正被创建
p.join([timeout])
- 功能:阻塞等待回收进程
- 参数:超时时间
注意:
- 使用multiprocessing创建进程同样是子进程复制父进程空间代码段,父子进程运行互不影响。
- 子进程只运行target绑定的函数部分,其余内容均是父进程执行内容。
- multiprocessing中父进程往往只用来创建子进程回收子进程,具体事件由子进程完成。
- multiprocessing创建的子进程中无法使用标准输入
3. 进程对象属性
p.name
进程名称
p.pid
对应子进程的PID号
p.is_alive()
查看子进程是否在生命周期
p.daemon
设置父子进程的退出关系
- 如果设置为True则子进程会随父进程的退出而结束
- 要求必须在start()前设置
- 如果daemon设置成True 通常就不会使用 join()
3×1 进程池实现
3x1x1 必要性
- 进程的创建和销毁过程消耗的资源较多
- 当任务量众多,每个任务在很短时间内完成时,需要频繁的创建和销毁进程。此时对计算机压力较大
- 进程池技术很好的解决了以上问题
3x1x2 原理
创建一定数量的进程来处理事件,事件处理完进程不退出而是继续处理其他事件,直到所有事件全都处理完毕统一销毁。增加进程的重复利用,降低资源消耗。
3x1x3 进程池实现
1 创建进程池对象,放入适当的进程
from multiprocessing import Pool Pool(processes) 功能: 创建进程池对象 参数: 指定进程数量,默认根据系统自动判定
2 将事件加入进程池队列执行
pool.apply_async(func,args,kwds) 功能: 使用进程池执行 func事件 参数: func 事件函数 args 元组 给func按位置传参 kwds 字典 给func按照键值传参 返回值: 返回函数事件对象
3 关闭进程池
pool.close() 功能: 关闭进程池
4 回收进程池中进程
pool.join() 功能: 回收进程池中进程
5 获取进程执行函数的返回值
get()
4×0 进程间通信(IPC)
必要性:进程间空间独立,资源不共享,此时在需要进程间数据传输时就需要特定的手段进行数据通信。
常用进程间通信方法:
4×1 管道通信(Pipe)
4x1x1 通信原理:
在内存中开辟管道空间,生成管道操作对象,多个进程使用同一个管道对象进行读写即可实现通信
4x1x2 实现方法
from multiprocessing import Pipe fd1,fd2 = Pipe(duplex = True) # 功能: 创建管道 # 参数:默认表示双向管道 # 如果为False 表示单向管道 # 返回值:表示管道两端的读写对象 # 如果是双向管道均可读写 # 如果是单向管道fd1只读 fd2只写 fd.recv() # 功能 : 从管道获取内容 # 返回值:获取到的数据 fd.send(data) # 功能: 向管道写入内容 # 参数: 要写入的数据
# 管道通信 from multiprocessing import Process, Pipe import os, time # 创建管道,参数为默认(True)双向通信管道 fd1, fd2 = Pipe() def fun(name): time.sleep(3) # 子进程向管道写入内容 fd1.send({name: os.getpid()}) # 用于保存子进程对象 jobs = [] for i in range(5): # 创建进程 p = Process(target=fun, args=(i,)) jobs.append(p) p.start() for i in range(5): # 父进程读取管道 data = fd2.recv() print(data) for i in jobs: i.join()
4×2 消息队列
4x2x1 通信原理
在内存中建立队列模型,进程通过队列将消息存入,或者从队列去除完成进程间通信。
4x2x2 实现方法
from multiprocessing import Queue q = Queue(maxsize=0) # 功能: 创建队列对象 # 参数:最多存放消息个数 # 返回值:队列对象 q.put(data,[block,timeout]) # 功能:向队列存入消息 # 参数:data 要存入的内容 # block 设置是否阻塞 False为非阻塞 # timeout 超时检测 q.get([block,timeout]) # 功能:从队列取出消息 # 参数:block 设置是否阻塞 False为非阻塞,默认为True,一直等到有数据为止 # timeout 超时检测,当过时间还没有数据抛出异常 # 返回值: 返回获取到的内容 q.get_nowait() # 相当于q.get(False) 如果空 立刻抛出异常 q.full() # 判断队列是否为满 q.empty() # 判断队列是否为空 q.qsize() # 获取队列中消息个数 q.close() # 关闭队列
# 消息队列通信 from multiprocessing import Queue, Process from time import sleep from random import randint # 创建消息队列 q = Queue(3) def request(): for i in range(20): x = randint(0, 100) y = randint(0, 100) # 向消息队列中写入信息 q.put((x, y)) def handle(): while True: sleep(0.5) try: # 取出消息队列的数据,延迟等待3秒,3秒过后抛出异常 x, y = q.get(timeout=3) except: break else: print(x + y) p1 = Process(target=request) p2 = Process(target=handle) p1.start() p2.start() p1.join() p2.join()
4×3 共享内存
4x3x1 通信原理
在内存中开辟一块空间,进程可以写入内容和读取内容完成通信,但是每次写入内容会覆盖之前内容
4x3x2 实现方法
from multiprocessing import Value,Array obj = Value(ctype,data) # 功能 : 开辟共享内存 # 参数 : ctype 表示共享内存空间类型 'i' 'f' 'c' # data 共享内存空间初始数据 # 返回值:共享内存对象 obj.value # 对该属性的修改查看即对共享内存读写 obj = Array(ctype,data) # 功能: 开辟共享内存空间 # 参数: ctype 表示共享内存数据类型 # data 整数则表示开辟空间的大小,其他数据类型 # 表示开辟空间存放的初始化数据 # 返回值:共享内存对象 # Array共享内存读写: 通过遍历obj可以得到每个值,直接可以通过索引序号修改任意值。 # 可以使用obj.value直接打印共享内存中的字节串
value
from multiprocessing import Process, Value import time import random # 创建共享内存 money = Value("i", 5000) # 操作共享内存 def man(): for i in range(30): time.sleep(0.2) money.value += random.randint(1, 1000) def girl(): for i in range(30): time.sleep(0.15) money.value -= random.randint(100, 800) m = Process(target=man) g = Process(target=girl) m.start() g.start() m.join() g.join() print("一月余额", money.value)
Array
from multiprocessing import Process, Array # 创建共享内存 # 共享内存开辟5个整型列表空间 # shm = Array('i',5) # 共享内存初始化数据[1,2,3] # shm = Array('i', [1, 2, 3]) # 字节串 shm = Array('c', b'hello') def fun(): # 共享内存对象可迭代 for i in shm: print(i) # 修改共享内存 shm[0] = b'H' p = Process(target=fun) p.start() p.join() for i in shm: print(i, end=' ') # 通过value属性访问字节串 print(shm.value)
4×4 本地套接字
功能:用于本地两个程序之间进行数据的收发。
套接字文件:用于本地套接字之间通信时,进行数据传输的介质。
创建本地套接字流程:
【1】 创建本地套接字
>sockfd = socket(AF_UNIX,SOCK_STREAM)
【2】 绑定本地套接字文件
>sockfd.bind(file)
【3】 监听,接收客户端连接,消息收发
>listen()-->accept()-->recv(),send()
unix_send.py
from socket import * sock_file = "./sock" sockfd = socket(AF_UNIX,SOCK_STREAM) sockfd.connect(sock_file) while True: msg = input(">>") if not msg: break sockfd.send(msg.encode()) sockfd.close()
unix_recv.py
from socket import * import os # 确定本地套接字文件 sock_file = "./sock" # 判断文件是否存在,存在就删 if os.path.exists(sock_file): os.remove(sock_file) # 创建本地套接字 sockfd = socket(AF_UNIX, SOCK_STREAM) # 绑定本地套接字 sockfd.bind(sock_file) # 监听,连接 sockfd.listen(3) while True: c, addr = sockfd.accept() while True: data = c.recv(1024) if not data: break print(data.decode()) c.close() sockfd.close()
4×5 信号量(信号灯集)
4x5x1 通信原理
给定一个数量对多个进程可见。多个进程都可以操作该数量增减,并根据数量值决定自己的行为。
4x5x2 实现方法
from multiprocessing import Semaphore sem = Semaphore(num) # 功能 : 创建信号量对象 # 参数 : 信号量的初始值 # 返回值 : 信号量对象 sem.acquire() # 将信号量减1 当信号量为0时阻塞 sem.release() # 将信号量加1 sem.get_value() # 获取信号量数量
from multiprocessing import Semaphore, Process from time import sleep import os # 创建信号量 # 服务程序最多允许3个进程同时执行实践 sem = Semaphore(3) def handle(): print("%d 想执行实践" % os.getpid()) # 想执行必须获取信号量 sem.acquire() print("%d 开始执行操作" % os.getpid()) sleep(3) print("%d 完成操作" % os.getpid()) # 增加信号量 sem.release() jobs = [] for i in range(1000): p = Process(target=handle) jobs.append(p) p.start() for i in jobs: i.join()
5×0 线程编程(Thread)
5×1 线程基本概念
- 什么是线程
【1】 线程被称为轻量级的进程
【2】 线程也可以使用计算机多核资源,是多任务编程方式
【3】 线程是系统分配内核的最小单元
【4】 线程可以理解为进程的分支任务
- 线程特征
【1】 一个进程中可以包含多个线程
【2】 线程也是一个运行行为,消耗计算机资源
【3】 一个进程中的所有线程共享这个进程的资源
【4】 多个线程之间的运行互不影响各自运行
【5】 线程的创建和销毁消耗资源远小于进程
【6】 各个线程也有自己的ID等特征
5×2 threading模块创建线程
1.创建线程对象
from threading import Thread t = Thread() # 功能:创建线程对象 # 参数:target 绑定线程函数 # args 元组 给线程函数位置传参 # kwargs 字典 给线程函数键值传参
2.启动与回收线程
t.start() t.join([timeout])
5×3 线程对象属性
t.name | 线程名称 |
t.setName() | 设置线程名称 |
t.getName() | 获取线程名称 |
t.is_alive() | 查看线程是否在生周期 |
t.daemon | 设置主线程和分支线程的退出关系 |
t.setDaemon() | 设置daemon属性值 |
t.isDaemon() | 查看daemon属性值 |
Deamon为true时主线程退出分支线程也退出。要在start前设置,通常不和join一起使用
5×4 自定义线程类
- 创建步骤
【1】 继承Thread类
【2】 重写init方法添加自己的属性,使用super加载父类属性
【3】 重写run方法 - 使用方法
【1】 实例化对象
【2】 调用start自动执行run方法
【3】 调用join回收线程
from threading import Thread from time import sleep, ctime class MyThread(Thread): def __init__(self, target, args=(), kwargs={}, name='Tedu'): super().__init__() self.target = target self.ctime = args self.song = kwargs self.name = name def run(self): self.target(*self.ctime, **self.song) # 通过完成上面的MyThread类让整个程序可以正常执行,当调用start时player作为一个线程功能函数运行 # 注意:函数的名称和参数并不确定,player只是测试函数 def player(sec, song): for i in range(2): print("Playing %s:%s" % (song, ctime())) sleep(sec) t = MyThread(target=player, args=(3,), kwargs={"song": "凉凉"}, name="happy") t.start() t.join()
6×0 同步互斥
6×1 线程间通信方法
1.通信方法
线程间使用全局变量进行通信
2.共享资源争夺
- 共享资源:多个进程或者线程都可以操作的资源称为共享资源。对共享资源的操作代码段称为临界区。
- 影响 : 对共享资源的无序操作可能会带来数据的混乱,或者操作错误。此时往往需要同步互斥机制协调操作顺序。
3.同步互斥机制
同步:同步是一种写作关系,为完成操作,多进程或线程间形成一种协调,按照必要的步骤有序执行操作。
互斥:互斥是一种制约关系,当一个进程或者线程占有资源时会进程加锁处理,此时其他进程就无法操作该资源,直到解锁后才能操作。
6×2 线程同步互斥方法
线程Event
from threading import Event e = Event() # 创建线程event对象 e.wait([timeout]) # 阻塞等待e被set e.set() # 设置e,使wait结束阻塞 e.clear() # 使e回到未被设置状态 e.is_set() # 查看当前e是否被设置
右边的代码中e.wait() 表示阻塞等待子线程执行完毕。这里如果不阻塞等待的话就会出现2中结果。
1.父进程先结束,子进程在结束
1.子进程先结束,父进程在结束
from threading import Thread, Event from time import sleep s = None # 全局变量用于通信 e = Event() # 时间对象 def f1(): print("杨子荣前来拜山头") global s s = "天王盖地虎" e.set() # 共享资源操作完毕 t = Thread(target=f1) t.start() print("说对口令就是自己人") e.wait() # 阻塞等待 if s == "天王盖地虎": print("宝塔镇河妖") print("确认过眼神,你是对的人") else: print("打死他") t.join()
线程锁
from threading import Lock lock = Lock() # 创建锁对象 lock.acquire() # 上锁 如果lock已经上锁再调用会阻塞 lock.release() # 解锁 with lock: # 上锁 ... ... # with代码块结束自动解锁
with代码块结束自动解锁
from threading import Thread, Lock a = b = 0 lock = Lock() def value(): while True: lock.acquire() # 上锁 if a != b: print("a = %d,b = %d" % (a, b)) lock.release() # 解锁 t = Thread(target=value) t.start() while True: with lock: a += 1 b += 1 t.join()
6×3 死锁及其处理
1.定义
死锁是指两个或两个以上的线程在执行过程中,由于竞争资源或者由于彼此通信而造成的一种阻塞的现象,若无外力作用,它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁。
2.死锁产生的条件
- 互斥条件:指线程对所分配到的资源进行排它性使用,即在一段时间内某资源只由一个进程占用。如果此时还有其它进程请求资源,则请求者只能等待,直至占有资源的进程用毕释放。
- 请求和保持条件:指线程已经保持至少一个资源,但又提出了新的资源请求,而该资源已被其它进程占有,此时请求线程阻塞,但又对自己已获得的其它资源保持不放。互斥条件:指线程对所分配到的资源进行排它性使用,即在一段时间内某资源只由一个进程占用。如果此时还有其它进程请求资源,则请求者只能等待,直至占有资源的进程用毕释放。
- 不剥夺条件:指线程已获得的资源,在未使用完之前,不能被剥夺,只能在使用完时由自己释放,通常CPU内存资源是可以被系统强行调配剥夺的。
- 环路等待条件:指在发生死锁时,必然存在一个线程——资源的环形链,即进程集合{T0,T1,T2,···,Tn}中的T0正在等待一个T1占用的资源;T1正在等待T2占用的资源,……,Tn正在等待已被T0占用的资源。
死锁产生的原因:
- 当前线程拥有其他线程需要的资源
- 当前线程等待其他线程已拥有的资源
- 都不放弃自己拥有的资源
3.如何避免死锁
死锁是我们非常不愿意看到的一种现象,我们要尽可能避免死锁的情况发生。通过设置某些限制条件,去破坏产生死锁的四个必要条件中的一个或者几个,来预防发生死锁。预防死锁是一种较易实现的方法。但是由于所施加的限制条件往往太严格,可能会导致系统资源利用率。
使用定时锁
使用重入锁RLock(),用法同Lock。RLock内部维护着一个Lock和一个counter变量,counter记录了acquire的次数,从而使得资源可以被多次require。直到一个线程所有的acquire都被release,其他的线程才能获得资源。
7×0 python线程GIL
- python线程的GIL问题(全局解释器锁)
- 什么是GIL :由于python解释器设计中加入了解释器锁,导致python解释器同一时刻只能解释执行一个线程,大大降低了线程的执行效率。python线程的GIL问题(全局解释器锁)
- 导致后果: 因为遇到阻塞时线程会主动让出解释器,去解释其他线程。所以python多线程在执行多阻塞高延迟IO时可以提升程序效率,其他情况并不能对效率有所提升
- GIL问题建议(官方)
- 尽量使用进程完成无阻塞的并发行为
- 不使用c作为解释器(Java C#)
- 结论:在无阻塞状态下,多线程程序和单线程程序执行效率几乎差不多,甚至还不如单线程效率。但是多进程运行相同内容却可以有明显的效率提升
8×0 进程线程的区别联系
8×1 区别联系
- 两者都是多任务编程方式,都能使用计算机多核资源
- 进程的创建删除消耗的计算机资源比线程多
- 进程空间多里,数据互不干扰,有专门通信方法;线程使用去哪局变量通信
- 一个进程可以有多个分支线程,两者有包含关系
- 多个线程共享进程资源,在共享资源操作时往往需要同步互斥处理
- 进程线程在系统中都有自己的特有属性标志,如ID,代码段,命令集等。
8×2 使用场景
- 任务场景:如果是相对独立的任务模块,可能使用多进程,如果是多个分支共同形成一个整体任务可能用多线程
- 项目结构:多中编程语言实现不同任务模块,可能是多进程,或者前后端分离应该各自为一个进程。
- 难以程度:通信难度,数据处理的复杂度来判断用进程间通信还是同步互斥方法。
8×3 要求
- 对进程线程怎么理解/说说进程线程的差异
- 进程间通信知道哪些,有什么特点
- 什么是同步互斥,你什么情况下使用,怎么用
- 给一个情形,说说用进程还是线程,为什么
- 问一些概念,僵尸进程的处理,GIL问题,进程状态
9×0 并发网络通信模型
9×1 常见模型分类
- 循环服务器模型:循环接受客户端请求,处理请求。同一时刻只能处理一个请求,处理完毕后再处理下一个。
- 优点:实现简单,占用资源少
- 缺点:无法同时处理多个客户端请求
- 使用情况:处理的任务可以很快完成,客户端无需长期占用服务端程序。udp比tcp更适合循环。
- IO并发模型:使用IO多路复用,异步IO等技术,同时处理多个客户端IO请求
- 优点:资源消耗少,能同时高效处理多个IO行为
- 缺点:只能处理并发产生的IO事件,无法处理cpu计算
- 适用情况:HTTP请求,网络传输等都是IO行为。
- 多进程、多线程网络并发模型:每当一个客户端连接服务器,就创建一个新的进程、线程为该客户端服务,客户端退出时在销毁该进程、线程。
- 优点:能同时满足多个客户端长期占有服务端需求,可以处理各种请求。
- 缺点:消耗资源较大
- 使用情况:客户端同时连接量较少,需要处理行为较为复杂情况。
9×2 基于fork的多进程网络并发模型
实现步骤:
- 创建监听套接字
- 等待接收客户端请求
- 客户端连接创建新的进程处理客户端请求
- 原进程继续等待其他客户端连接
- 如果客户端退出,则销毁对应的进程
""" 基于fork的多进程网络并发 重点代码 """ from socket import * import sys, os import signal def handle(c): print("客户端:", c.getpeername()) while True: data = c.recv(1024) if not data: break print(data.decode()) c.send(b"OK") c.close() # 创建监听套接字 HOST = "0.0.0.0" PORT = 8888 ADDR = (HOST, PORT) s = socket() # tcp套接字 s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) # 设置端口的地址重用 s.bind(ADDR) s.listen(3) # 僵尸进程处理 signal.signal(signal.SIGCHLD, signal.SIG_IGN) print("Listen the port 8888....") # 循环等待客户端连接 while True: try: c, addr = s.accept() except KeyboardInterrupt: sys.exit("服务器退出") except Exception as e: print(e) continue # 创建子进程处理客户端请求 pid = os.fork() if pid == 0: s.close() # 子进程不需要s handle(c) # 具体处理客户端请求 os._exit(0) # 父进程实际只用来处理客户端链接 else: c.close() # 父进程不需要c
9×3 基于threading的多线程网络并发
实现步骤
- 创建监听套接字
- 循环接收客户端连接请求
- 当有新的客户端连接创建线程处理客户端请求
- 主线程继续等待其他客户端连接
- 当客户端退出,则对应分支线程退出
""" 多线程网络并发 重点代码 """ from socket import * from threading import Thread import sys # 客户端处理 def handle(c): print("Connect from:", c.getpeername()) while True: data = c.recv(1024) if not data: break print(data.decode()) c.send(b"OK") c.close() # 创建监听套接字 HOST = "0.0.0.0" PORT = 8888 ADDR = (HOST, PORT) s = socket() s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) s.bind(ADDR) s.listen(3) print("监听开始:8888") # 循环等待 while True: try: c, addr = s.accept() except KeyboardInterrupt: sys.exit("退出服务器") except Exception as e: print(e) continue # 创建新的线程处理客户端请求 t = Thread(target=handle, args=(c,)) t.setDaemon(True) # 主线程退出,分支线程也退出 t.start()
9×4 扩展:集成模块完成多进程、多线程网络并发
1.使用方法
import socketserver # 通过模块提供的不同的类的组合完成多进程或者多线程,tcp或者udp的网络并发模型
2.常用类说明
TCPServer 创建tcp服务端套接字
UDPServer 创建udp服务端套接字
StreamRequestHandler 处理tcp客户端请求
DatagramRequestHandler 处理udp客户端请求
ForkingMixIn 创建多进程并发
ForkingTCPServer ForkingMixIn + TCPServer
ForkingUDPServer ForkingMixIn + UDPServer
ThreadingMixIn 创建多线程并发
ThreadingTCPServer ThreadingMixIn + TCPServer
ThreadingUDPServer ThreadingMixIn + UDPServer
3.使用步骤
【1】 创建服务器类,通过选择继承的类,决定创建TCP或者UDP,多进程或者多线程的并发服务器模型。
【2】 创建请求处理类,根据服务类型选择stream处理类还是Datagram处理类。重写handle方法,做具体请求处理。
【3】 通过服务器类实例化对象,并绑定请求处理类。
【4】 通过服务器对象,调用serve_forever()启动服务
9×5 ftp文件服务器
功能:
【1】 分为服务端和客户端,要求可以有多个客户端同时操作。
【2】 客户端可以查看服务器文件库中有什么文件。
【3】 客户端可以从文件库中下载文件到本地。
【4】 客户端可以上传一个本地文件到文件库。
【5】 使用print在客户端打印命令输入提示,引导操作
10×0 IO并发
10×1 IO分类
阻塞IO、非阻塞IO、IO多路复用、异步IO等
- 阻塞IO
- 定义:在执行IO操作时如果执行条件不满足则阻塞。阻塞IO是IO的默认形态
- 效率:阻塞IO是效率很低的一种IO。但是由于逻辑简单所以是默认的IO行为。
- 阻塞情况:
- 因为某种执行条件没有满足造成的函数阻塞
- accept input recv
- 处理IO的时间较长产生的阻塞状态
- 网络传输,大文件读写
- 因为某种执行条件没有满足造成的函数阻塞
- 非阻塞IO
- 定义:通过修改IO属性行为,使原本阻塞的IO变为非阻塞的状态
- 设置套接字为非阻塞IO
sockfd.setblocking(bool)
- 功能:设置套接字为非阻塞IO
- 参数:默认为True,表示套接字IO阻塞;设置为False则套接字IO变为非阻塞
- 超时检测:设置一个最长阻塞时间,超过该时间后则不再阻塞等待。
sockfd.settimeout(sec)
- 功能:设置套接字的超时时间
- 参数:设置的时间
10×2 IO多路复用
1.定义
同时监控多个IO事件,当哪个IO事件准备就绪就执行哪个IO事件。以此形成可以同时处理多个IO的行为,避免一个IO阻塞造成其他IO均无法执行,提高了IO执行效率。
IO 多路复用,实现单进程同时处理多个 socket 请求。所以无法使用计算机多核资源!
2.具体方案
select方法 : windows linux unix
poll方法: linux unix
epoll方法: linux
select方法
rs, ws, xs=select(rlist, wlist, xlist[, timeout]) # 功能: 监控IO事件,阻塞等待IO发生 # 参数:rlist 列表 存放关注的等待发生的IO事件 # wlist 列表 存放关注的要主动处理的IO事件 # xlist 列表 存放关注的出现异常要处理的IO # timeout 超时时间 # 返回值: rs 列表 rlist中准备就绪的IO # ws 列表 wlist中准备就绪的IO # xs 列表 xlist中准备就绪的IO
select 实现tcp服务
【1】将关注的IO放入对应的监控类别列表 【2】通过select函数进行监控 【3】遍历select返回值列表,确定就绪IO事件 【4】处理发生的IO事件
注意
wlist中如果存在IO事件,则select立即返回给ws
处理IO过程中不要出现死循环占有服务端的情况
IO多路复用消耗资源较少,效率较高
""" IO多路复用select实现多客户端通信 重点代码 """ from socket import * from select import select # 设置套接字为关注IO s = socket() s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) s.bind(('0.0.0.0', 8888)) s.listen(5) # 设置关注的IO rlist = [s] wlist = [] xlist = [] while True: # 监控IO的发生 rs, ws, xs = select(rlist, wlist, xlist) # 遍历三个返回值列表,判断哪个IO发生 for r in rs: # 如果是套接字就绪则处理链接 if r is s: c, addr = r.accept() print("Connect from", addr) rlist.append(c) # 加入新的关注IO else: data = r.recv(1024) if not data: rlist.remove(r) r.close() continue print(data.decode()) # r.send(b'OK') # 希望我们主动处理这个IO wlist.append(r) for w in ws: w.send(b'Ok,Thanks') wlist.remove(w) for x in xs: pass
扩展:位运算
定义 : 将整数转换为二进制,按二进制位进行运算
运算符号:
& 按位与 | 按位或 ^ 按位异或 << 左移 >> 右移 e.g. 14 --> 01110 19 --> 10011 14 & 19 = 00010 = 2 一0则0 14 | 19 = 11111 = 31 一1则1 14 ^ 19 = 11101 = 29 相同为0不同为1 14 << 2 = 111000 = 56 向左移动低位补0 14 >> 2 = 11 = 3 向右移动去掉低位
poll方法
p = select.poll()
功能:创建poll对象
返回值:poll对象
p.register(fd,event) 功能: 注册关注的IO事件 参数:fd 要关注的IO event 要关注的IO事件类型 常用类型:POLLIN 读IO事件(rlist) POLLOUT 写IO事件 (wlist) POLLERR 异常IO (xlist) POLLHUP 断开连接 e.g. p.register(sockfd,POLLIN|POLLERR) p.unregister(fd) 功能:取消对IO的关注 参数:IO对象或者IO对象的fileno
events = p.poll() 功能: 阻塞等待监控的IO事件发生 返回值: 返回发生的IO events格式 [(fileno,event),()....] 每个元组为一个就绪IO,元组第一项是该IO的fileno,第二项为该IO就绪的事件类型
poll_server步骤
【1】 创建套接字
【2】 将套接字register
【3】 创建查找字典,并维护
【4】 循环监控IO发生
【5】 处理发生的IO
""" poll多路复用 次重点 """ from socket import * from select import * # 设置套接字为关注IO s = socket() s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) s.bind(('0.0.0.0',8888)) s.listen(5) # 创建poll p = poll() # 建立查找字典 {fileno: io_obj} fdmap = {s.fileno():s} # 设置关注IO p.register(s,POLLIN|POLLERR) # 循环监控io事件发生 while True: events = p.poll() # 阻塞等待IO发生 print(events) # 遍历列表处理IO for fd,event in events: if fd == s.fileno(): c,addr = fdmap[fd].accept() print("Connect from",addr) # 添加新的关注事件 p.register(c,POLLIN|POLLHUP) fdmap[c.fileno()] = c # elif event & POLLHUP: # 客户端断开 # print("客户端退出") # p.unregister(fd) # 取消关注 # fdmap[fd].close() # del fdmap[fd] # 从字典删除 elif event & POLLIN: # 客户端发消息 data = fdmap[fd].recv(1024) # 断开发生时data得到空此时POLLIN也会就绪 if not data: p.unregister(fd) # 取消关注 fdmap[fd].close() del fdmap[fd] continue print(data.decode()) fdmap[fd].send(b'OK')
epoll方法
- 使用方法 : 基本与poll相同
- 生成对象改为 epoll()
- 将所有事件类型改为EPOLL类型
- epoll特点
- epoll 效率比select poll要高
- epoll 监控IO数量比select poll要多
- epoll 的触发方式比poll要多 (EPOLLET边缘触发)
""" epoll多路复用 次重点 """ from socket import * from select import * # 设置套接字为关注IO s = socket() s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) s.bind(('0.0.0.0',8888)) s.listen(5) # 创建epoll ep = epoll() # 建立查找字典 {fileno: io_obj} fdmap = {s.fileno():s} # 设置关注IO ep.register(s,EPOLLIN|EPOLLERR) # 循环监控io事件发生 while True: events = ep.poll() # 阻塞等待IO发生 print(events) # 遍历列表处理IO for fd,event in events: if fd == s.fileno(): c,addr = fdmap[fd].accept() print("Connect from",addr) # 添加新的关注事件 ep.register(c,EPOLLIN|EPOLLHUP|EPOLLET) fdmap[c.fileno()] = c # elif event & EPOLLIN: # 客户端发消息 # data = fdmap[fd].recv(1024) # # 断开发生时data得到空此时EPOLLIN也会就绪 # if not data: # ep.unregister(fd) # 取消关注 # fdmap[fd].close() # del fdmap[fd] # continue # print(data.decode()) # fdmap[fd].send(b'OK')
10×3 协程技术
基础概念
- 定义:纤程,微线程。是为非抢占式多任务产生子程序的计算机组件。协程允许不同入口点在不同位置暂停或开始,简单来说,协程就是可以暂停执行的函数。
- 协程原理 : 记录一个函数的上下文栈帧,协程调度切换时会将记录的上下文保存,在切换回来时进行调取,恢复原有的执行内容,以便从上一次执行位置继续执行。
- 协程优缺点
优点
- 协程完成多任务占用计算资源很少
- 由于协程的多任务切换在应用层完成,因此切换开销少
- 协程为单线程程序,无需进行共享资源同步互斥处理
缺点
协程的本质是一个单线程,无法利用计算机多核资源
第三方协程模块:greenlet模块、gevent模块
11×0 HTTPServer v2.0
- 主要功能 :
【1】 接收客户端(浏览器)请求
【2】 解析客户端发送的请求
【3】 根据请求组织数据内容
【4】 将数据内容形参http响应格式返回给浏览器 - 升级点 :
【1】 采用IO并发,可以满足多个客户端同时发起请求情况
【2】 做基本的请求解析,根据具体请求返回具体内容,同时满足客户端简单的非网页请求情况
【3】 通过类接口形式进行功能封装
""" 技术点 1.使用tcp通信 2.select io多路复用 结构: 1.采用类封装 类的接口设计: 1.在用户使用角度进行工作流程设计 2.尽可能提供全面的功能,能为用户决定的在类中实现 3.不能替用户决定的变量可以通过实例化对象传入类中 4.不能替用户决定的复杂功能,可以通过重写让用户自己决定 """ # httpserver2.0 # IO并发处理 # 基本request解析 # 使用类封装 from socket import * from select import select # 将具体http server功能封装 class HTTPServer: def __init__(self, server_addr, static_dir): # 添加属性 self.server_addr = server_addr self.static_dir = static_dir self.rlist = self.wlist = self.xlist = [] self.create_socket() self.bind() self.serve_forever() def create_socket(self): """ 创建套接字 :return: """ self.sockfd = socket() self.sockfd.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) def bind(self): """ 绑定ip端口 :return: """ self.sockfd.bind(self.server_addr) self.ip = self.server_addr[0] self.port = self.server_addr[1] def serve_forever(self): """ 开启服务 :return: """ self.sockfd.listen(5) print("listen the port:%s" % self.port) self.rlist = [self.sockfd] while True: rs, ws, xs = select(self.rlist, self.wlist, self.xlist) for r in rs: if r is self.sockfd: c, addr = r.accept() print("connect from:", addr) self.rlist.append(c) else: # 处理浏览器请求 self.handle(r) def handle(self, conn): """ 处理浏览器请求 :param conn: obj类型,客户端连接的套接字 :return: """ # 接受http请求 request = conn.recv(4096) # 防止游览器断开 if not request: self.rlist.remove(conn) conn.close() return # 请求解析 request_line = request.splitlines()[0] info = request_line.decode().split(" ")[1] print(conn.getpeername, ":", info) # info 分为访问网页和其他 if info == "/" or info[-5:] == ".html": self.get_html(conn, info) else: self.get_data(conn, info) self.rlist.remove(conn) conn.close() def get_html(self, coon, info): """ 处理网页 :param coon: obj类型,客户端连接的套接字 :param info: str类型,客户端请求体 :return: """ if info == "/": filename = self.static_dir + "/index.html" else: filename = self.static_dir + info try: fr = open(filename) except Exception: # 没有网页返回404 responseHeader = "HTTP/1.1 404 Not Found\r\n" responseHeader += "\r\n" responseBody = "<h1>没有发现这个网页</h1>" else: responseHeader = "HTTP/1.1 200 OK\r\n" responseHeader += "\r\n" responseBody = fr.read() finally: response = responseHeader + responseBody coon.send(response.encode()) def get_data(self, coon, info): """ 表示访问的不是网页,是其他内容的情况 :param coon: obj类型,客户端连接的套接字 :param info: str类型,客户端请求体 :return: """ responseHeader = "HTTP/1.1 200 OK\r\n" responseHeader += "\r\n" responseBody = "<span>waiting HTTPServer 3.0<span>" response = responseHeader + responseBody coon.send(response.encode()) # 如何使用HTTPServer类 if __name__ == '__main__': # 用户自己决定:地址,内容 server_addr = ("0.0.0.0", 8888) # 服务器地址 static_dir = "./static" # 网页存放地址 httpd = HTTPServer(server_addr, static_dir) # 生成实例对象 httpd.serve_forever() # 启动服务