1 进程管理
在传统操作系统中, 程序不能独立运行, 进程才是作为资源分配和独立运行的基本单位
1.1 进程的基本概念
如果不设置操作系统, 那程序的执行方式是顺序执行
在多道程序环境下, 则允许多个程序并发执行
程序的顺序执行的特征
1) 顺序性
2) 封闭性, 程序在封闭的环境下执行, 程序运行时独占全机资源
3) 可再现性, 只要初始环境和初始条件相同, 程序重复执行的结果是相同的
前趋图
顺序执行可以用前趋图来表示
程序并发执行
程序兵法执行的前趋图
程序并发执行的特征
程序的兵法执行提高了系统吞吐量
1) 间断性, 程序并发执行, 由于共享系统资源, 图中C2需要C1和I2都完成才能执行
2) 失去封闭性, 由于共享各种资源, 资源的状态由多个程序来改变, 因而程序失去封闭性
3) 不可再现性, 失去封闭性的同时也失去了再现性
由于程序的并发执行的结果是不可再现的
所以引入进程的概念
进程实体 = 程序段 + 相关的数据段 + PCB
在早期Unix中, 这个进程实体也叫"进程映像"
PCB(Process Control Block)
进程的特征
1) 结构特征, 也就是 进程实体的组成
2) 动态性, 进程的实质是 进程实体的一次执行过程
3) 并发性, 进程能够兵法执行
4) 独立性, 进程实体是一个独立运行, 独立分配资源, 独立接受调度的基本单位
5) 异步性, 进程实体按照异步的方式运行
进程的三种状态
就绪(Ready); 执行; 阻塞
进程的五种状态的基本转换
PCB
PCB中保留了操作系统所需的, 用于描述进程的当前情况以及控制进程运行的全部信息
PCB使得 程序+相关的数据 能够成为独立运行的基本单位
操作系统是通过PCB来对兵法执行的进程进行控制和管理的, 具体的是优先级,恢复现场环境, 进程间的同步和通信等
PCB随着进程的创建创建, 随着进程的消亡而消亡
PCB中具体包含的内容
1) 进程标识符, 主要是内部标识符和外部标识符
2) 处理机状态, 主要有通用寄存器; 指令计数器; 程序状态字PSW; 用户栈指针
3) 进程调度信息, 主要有进程状态; 进程优先级; 进程调度所需要的其他纤细包括调度算法,已等待的时间等; 用户栈指针
4) 进程控制信息, 包括程序和数据的地址; 进程同步和通信机制(消息队列指针, 信号量); 资源清单; 链接指针
PCB组织方式
1) 链接方式, 统一状态的PCB, 形成一个个队列, 然后记录不同类别的队列起始指针
2) 索引方式, 根绝进程的状态建立几张索引表
1.2 进程控制
进程控制使用的是原语(Primitive), 也就是"原子操作", 要么全部做要么不做, 原语执行在管态中
进程图
进程图用于描述一个进程的家族关系的有向树
引起创建进程的事件
1) 用户登录
2) 作业调度
3) 提供服务
4) 应用请求, 前三种是系统为自己创建新锦成, 而第四类是基于应用进程的请求
进程的创建
1) 申请空白的PCB
2) 为新进程分配资源
3) 初始化PCB
4) 将新进程插入到就绪队列
进程的终止
进程的阻塞与唤醒
进程的刮起和激活
1.3 进程同步
引入进程之后, 提高了资源的利用率和系统的吞吐量, 但是由于进程的异步性, 程序在使用临界资源的时候混乱不堪, 结果无法复现
进程的同步就是协调进程执行的次序, 使其有效共享资源和相互合作
进程同步的两支制约关系
1) 间接相互制约, 例如进程A请求打印并获取, 此时进程B请求打印, 就需要阻塞等待A完成打印
2) 直接相互制约, 例如进程A的输出给进程B, 当进程A没有输出了, 进程B就需要阻塞等待
临界资源
需要胡吃饭我跟的共享资源
临界区
访问临界资源的代码
信号量机制
主要是使用两个原子操作wait(S)和signal(S), 也就是PV操作
例如信号量S初始值为1, P操作自后S变为0, 当S大于0时执行V操作, V操作之后S增加1, 此时PV就可以控制操作了
管程机制
管程 = 代表共享资源的数据结构 + 对共享数据实施操作的一组过程
组成
1)管程的名字
2) 数据结构的说明
3) 对数据结构的操作过程
4) 管程内部的共享数据
经典同步问题
1) 生产者消费者问题
2) 哲学家就餐问题
3) 读者写着问题
进程间的通信
进程之间的通信的数据量有多有少, 交换的信息量少的叫做低级通信(例如进程的互斥和同步)
进程同步使用的信号量机制在进程通信商表现的不足为
1) 效率低
2) 通信对用户不透明
通信的类型
1) 共享存储器系统
2) 消息传递系统
3) 管道系统
2 用户态和内核态
用户态和内核态是操作系统的两种运行级别
内核态可以执行任何指令, 可以访问系统中任何存储位置
用户模式的进程不允许执行特权命令(停止CPU, 改变模式为, 发起一个IO操作), 也不允许方位内核区的代码和数据
在intel的CPU中, 有四种运行界别, Ring0~Ring3, 其中Ring0是内核态的权限
在32位的系统, 前8位指定给Ring0(大约寻址1G), 后24位分配给Ring1~Ring3(大约寻址3G)
也就是对应虚拟空间
具体情况如下所示
3 常见概念
串行, 顺序执行
并行, 同时运行
并发, 时间片切换运行
同步, 一个进程在执行某个请求的时候, 如果没有返回则会一直等待
异步, 不管其他进程状态, 继续执行自己之后的内容
4 实现多进程
4.1 在Uniux中使用os模块的fork()生成一个进程
其中fork创建的子进程会复制父进程的数据, 如果创建失败会返回OSError
如果fork返回为0, 则转而运行子进程
如果fork返回非0, 则继续执行父进程, 返回的值就是子进程PID
具体示例如下
import os
print('父进程 (%s) 启动' % os.getpid())
pid = os.fork()
if pid == 0:
print('子进程 (%s) , 父进程 %s' % (os.getpid(), os.getppid()))
else:
print('父进程 (%s) , 创建了子进程 (%s)' % (os.getpid(), pid))
4.2 使用Process创建进程
需要传入参数target是指定子进程运行的目标(一般是函数), 指定args为传入的参数, 用元组格式, 可以指定name来设置进程的名字
创建进程之后使用start()运行进程
使用join()使得进程执行完毕之后才能执行主进程, 可以设置一个参数表示过了这个时候之后主程序会执行
具体代码如下
from multiprocessing import Process
import os
# 子进程要执行的代码
def run_proc(name):
print('子进程 %s (%s)...' % (name, os.getpid()))
if __name__=='__main__':
print('父进程 %s.' % os.getpid())
p = Process(target=run_proc, args=('sub_process',))
print('启动子进程.')
p.start()
p.join()
print('子进程结束.')
4.3 用进程池创建进程
具体代码如下
from multiprocessing import Pool
import os
import time
import random
def pool_run(name):
print("启动子进程{}, pid:{}".format(name, os.getpid()))
start_time = time.time()
# 睡0~3秒
time.sleep(random.random()*3)
print("进程{}运行{:.2f}秒".format(name, time.time() - start_time))
if __name__ == '__main__':
print("父进程{}".format(os.getpid()))
my_pool = Pool(4)
for i in range(5):
my_pool.apply_async(pool_run, args=(i,))
print("等待子进程结束")
# 执行了close()就无法继续添加新的Process
my_pool.close()
my_pool.join()
print("程序结束")
4.4 使用subprocess
suibprocess可以启动一个子进程, 并且控制其的输入和输出
可以使用call方法来执行一条命令, 传入数据是一个列表, 分别是命令和参数
具体代码如下
import subprocess
r = subprocess.call(['tracert', 'www.baidu.com'])
print('Exit code:', r)
使用communicate来传入参数
具体如下
import subprocess
print('$ nslookup')
p = subprocess.Popen(['nslookup'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, err = p.communicate(b'set q=mx
python.org
exit
')
print(output.decode('utf-8'))
print('Exit code:', p.returncode)
4.5 进程之间的通信
进程之间的通信可以使用Queue来传递, 需要生成一个Queue对象, 用set()方法设置消息, 通过get()方法获得消息
具体实例如下
from multiprocessing import Process, Queue
import os, time, random
# 写数据进程执行的代码:
def write(q):
print('写进程%s' % os.getpid())
for value in ['A', 'B', 'C']:
print('发送消息: %s' % value)
q.put(value)
# 如果不睡的话, 一下子就把消息写到队列中了
# time.sleep(random.random())
# 读数据进程执行的代码:
def read(q):
print('读进程: %s' % os.getpid())
while True:
value = q.get(True)
print('得到消息: %s' % value)
if __name__=='__main__':
# 父进程创建Queue，并传给各个子进程：
q = Queue()
# 生成两个对象, 一个发送消息, 一个接受消息
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
# 启动子进程pw，写入:
pw.start()
# 启动子进程pr，读取:
pr.start()
# 等待pw结束:
pw.join()
# pr进程里是死循环，无法等待其结束，只能强行终止:
pr.terminate()
5 实现多线程
5.1 使用threading创建线程
threading的使用和进程创建是类似的
使用Thread类直接创建对象
具体如下
import time
import random
import threading
def sub_thread():
print("子线程{}启动".format(threading.current_thread().name))
n = 0
while n < 5:
n += 1
print("{}输出消息{}".format(threading.current_thread().name, n))
time.sleep(random.random()*2)
print("子线程结束")
# 主线程的名字是MainThread
print("主线程{}启动".format(threading.current_thread().name))
t = threading.Thread(target=sub_thread, name="my_sub_thread")
t.start()
t.join()
print("主线程{}结束".format(threading.current_thread().name))
继承Thread来创建线程
具体如下
import threading
import time
class MyThread(threading.Thread):
def __init__(self,num):
threading.Thread.__init__(self)
self.num=num
def run(self):
print("running on number:%s" %self.num)
time.sleep(3)
t1=MyThread(56)
t2=MyThread(78)
t1.start()
t2.start()
print("ending")
5.2 常用方法
threading.activeCount() 查看存活的线程的数量
threading.enumerate() 查看当前存活线程
线程.setDaemon(True/False)
为True的时候, 程序为守护线程, 主程序结束了, 该线程也死亡
为False的时候, 程序为前台线程, 自己运行自己的, 主程序影响不了
具体实例
当只有t1设为守护进程, 由于t2执行时间长与t1, 所以t1实际上在第3秒死亡, t2和主程序在第5秒死亡
当只有t2设为守护进程, 此时t1在第3秒死亡, 同时主进程死亡, 因而此时t2也死亡
5.3 GIL
python的多线程虽然是真的多线程
但是由于全局锁(GIL)的关系, 每执行100条字节码, 就会自动释放GIL, 让别的线程可以执行
这是历史遗留问题
GIL使得Python在cPython解释器下, 同一时间点只能运行一个先后才能
Python使用多线程的时候要考虑使用多线程处理的内容的类型
如果是IO密集型, 那么使用多线程是一个很合适的能够提高处理效果的解决办法
但是如果是计算密集型, 那么使用多线程在绝大多数情况下还不如使用顺序执行来得更有效率
5.4 线程同步问题
如果在线程中, 处理临界资源不恰当, 会引起线程同步问题
解决办法是使用同步锁来解决
使用threading.Lock()来生成一个锁, 使用acquire()来实现锁定, 使用release()来解锁
具体代码如下
import time, threading
# 假定这是你的银行存款:
balance = 0
lock = threading.Lock()
def change_it(n):
# 先存后取，结果应该为0:
global balance
lock.acquire()
balance = balance + n
lock.release()
def run_thread(n):
for i in range(100000):
change_it(n)
t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)
5.5 死锁与递归锁
死锁的出现是因为滥用锁
如果一个程序需要拿到了一把锁需要另一把锁, 而另一个程序拿到了第一个程序需要的锁却等着第一个程序拿到的那个锁
那么这两个程序就会相互等待对方完成之后释放锁, 这就是死锁现象
具体实例如下
import threading
import time
mutexA = threading.Lock()
mutexB = threading.Lock()
class MyThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
self.fun1()
self.fun2()
def fun1(self):
mutexA.acquire()
print("{} func1 拿到mutexA".format(threading.current_thread().name))
mutexB.acquire()
print("{} func1 拿到mutexB".format(threading.current_thread().name))
mutexB.release()
mutexA.release()
def fun2(self):
mutexB.acquire()
print("{} func2 拿到mutexA".format(threading.current_thread().name))
# 如果不睡眠, 那么就不一定出现死锁
time.sleep(0.2)
mutexA.acquire()
print("{} func2 拿到mutexB".format(threading.current_thread().name))
mutexA.release()
mutexB.release()
if __name__ == "__main__":
for i in range(0, 10):
my_thread = MyThread()
my_thread.start()
其中要把握一个关键点, 如果没有sleep(), 由于python中的GUL的关系, 在这个例子中的进程没有IO操作, CPU不容易发生切换, 因而执行完func1之后执行func2的速度要远大于创建一个新线程并拿到func1的第一把锁
另外Python还提供递归锁
使用threading.RLock()创建一个锁, 这个锁的特点就是可以被多次锁定, 在其内部中, 有一个计数器counter来记录当前被锁的次数
其余用法不变
5.6 Event和信号量
event对象类似于PV操作
isSet() 返回event的状态值
wait() 如果isSet为False则线程阻塞
set() 设置event的状态值为True
clear() 设置evet状态值为False
使用代码如下
import threading
import time
import logging
logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)
# def worker(event):
# logging.debug('等待redis启动')
# event.wait()
# logging.debug('redis启动完毕, 正在使用 [%s]', time.ctime())
# time.sleep(1)
def worker(event):
while not event.is_set():
logging.debug('等待redis启动')
event.wait(1)
logging.debug('redis启动完毕, 正在使用 [%s]', time.ctime())
time.sleep(1)
def main():
readis_ready = threading.Event()
t1 = threading.Thread(target=worker, args=(readis_ready,), name='t1')
t1.start()
t2 = threading.Thread(target=worker, args=(readis_ready,), name='t2')
t2.start()
logging.debug('主程序在3秒钟后启动redis')
time.sleep(3)
readis_ready.set()
if __name__=="__main__":
main()
信号量是使用threading.Semaphore来创建的, 创建的时候可以设置数量, 这个数量也就是最大的使用锁的数量
可以通过创建的信号量对象调用acquire()和release()来进入锁和释放锁
具体使用代码如下
import threading
import time
semaphore = threading.Semaphore(5)
def func():
if semaphore.acquire():
print (threading.currentThread().getName() + '得到了一个信号量')
time.sleep(2)
semaphore.release()
for i in range(20):
t1 = threading.Thread(target=func)
t1.start()
5.6 生产者消费者
所需的知识是队列
生产者生产出内容之后, 将内容放在队列里, 消费者可以获得队列的产品从而消费
具体代码如下
import time,random
import queue,threading
q = queue.Queue()
def Producer():
count = 0
while count <3:
print("生产者正在生产")
time.sleep(random.randrange(1))
q.put(count)
print('生产者 生产的第 %s 个产品' % count)
count +=1
def Consumer():
count = 0
while count <5:
time.sleep(random.randrange(2))
if not q.empty():
data = q.get()
print(' 33[32;1m 消费者消费产品 %s 33[0m' % data )
else:
print(" 33[32;1m 现在没有产品可以消费 33[0m")
count +=1
p1 = threading.Thread(target=Producer)
c1 = threading.Thread(target=Consumer)
p1.start()
c1.start()
6 协程
6.1 协程描述生产者与消费者
先生成消费者的生成器, 完成第一次next()之后, 像携程函数send()生产好的产品
消费者函数消费产品并返回状态
具体代码如下
import time
def consumer():
r = ''
while True:
n = yield r
if not n:
return
print('[消费者] ←← 消费 %s' % n)
time.sleep(1)
r = '200 OK'
def produce(c):
next(c)
n = 0
while n < 5:
n = n + 1
print('[生产者] →→ 生产 %s' % n)
cr = c.send(n)
print('[生产者] 消费者给我回复的信息: %s' % cr)
c.close()
if __name__=='__main__':
c = consumer()
produce(c)
6.2 使用greenlet
使用greenlet可以在程序中自己书写代码来完成函数的切换
切换时使用函数 switch()
使用greenlet需要安装第三方包, 可以在pycharm上安装
具体代码如下
from greenlet import greenlet
def test1():
print(12)
gr2.switch()
print(34)
gr2.switch()
def test2():
print(56)
gr1.switch()
print(78)
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()
得到的结果是12, 56, 34, 78
6.3 使用gevent
基于greenlet之上, 有提供的gevent来实现协程的切换
其中gevent也是需要安装
同样的道理, 在遇到IO操作的时候, gevent会自动切换执行其他协程
使用gevent.joinall()来创建需要执行的程序
具体代码如下
import gevent
import time
def foo():
print("执行foo")
gevent.sleep(2)
print("切换回foo")
def bar():
print("切换到bar")
gevent.sleep(5)
print("又切换到bar")
start=time.time()
gevent.joinall(
[gevent.spawn(foo),
gevent.spawn(bar)]
)
print(time.time()-start)
7 IO模型
IO墨香有五种, 阻塞IO, 非阻塞IO, IO多路复用, 信号驱动IO, 异步IO
其中信号驱动IO不常用
7.1 阻塞IO
在套接字模型中, 阻塞IO就是正常的模式
在链接完毕之后, 服务器等待客户端发送数据是一个阻塞状态
数据从内核态数据区域复制到用户态数据区域供程序使用又是一个阻塞状态
具体模型如下
7.2 非阻塞IO
在网络编程模型中, 就是设置setblocking(False), 这样程序不管是否接受到数据都会立马返回
其中如果没有数据会带着异常返回
具体模型如下
具体代码如下
import socket
import time
server = socket.socket()
server.bind(("127.0.0.1", 8080))
server.listen(5)
server.setblocking(False)
while True:
try:
print('等待客户端连接')
connection, address = server.accept() # 进程主动轮询
client_messge = connection.recv(1024)
print(str(client_messge, 'utf8'))
connection.close()
except Exception as e:
print(e)
time.sleep(4)
########################
import time
import socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
while True:
client.connect(('127.0.0.1', 6667))
print("hello")
client.sendall(bytes("hello", "utf8"))
time.sleep(2)
break
7.3 多路复用IO
重点
多路复用IO也是事件驱动IO
使用多路复用IO可以实现在单个程序里相应多个客户端程序
在接受客户端信息的时候, 使用select来完成数据的接受
使用select的时候, 会发起系统调用, 内核会监视select中哪些发生了变化, 发证变化的就会立即返回
模型如下
具体代码如下
import socket
import select
server = socket.socket()
server.bind(("127.0.0.1", 8080))
server.listen(5)
conn_lists = [server, ]
while True:
r, w, e = select.select(conn_lists, [], [], 5)
for obj in r:
if obj == server:
conn, add = obj.accept()
conn_lists.append(conn)
else:
data_byte = obj.recv(1024)
print(str(data_byte, 'utf8'))
inp = input('回答%s号客户>>>' % conn_lists.index(obj))
obj.sendall(bytes(inp, 'utf8'))
#######################################
import socket
client = socket.socket()
client.connect(('127.0.0.1', 8080))
while True:
inp = input(">>")
client.sendall(bytes(inp, "utf8"))
data = client.recv(1024)
print(str(data, 'utf8'))
其中select.select()可以传入多个套接字对象, 返回的第一个值也是一个列表, 里面有接受到内容的套接字
如果返回的是原来的server, 说明有新的连接进入
将连接accept()可以获得连接的套接字, 将其放入select()中可以监控它发送消息, 执行的时候针对它处理就可以回复处理该链接了
7.4 异步IO
异步IO全程无阻塞
实现的模型如下
7.5 IO模型比较
在IO模型中, 同步IO是指在整个过程中有阻塞, 只有没有阻塞的才是异步IO
各个模型比较如下
