python多任务编程---进程

多任务编程

1. 意义:充分利用计算机CPU的多和资源,同时处理多个应用程序任务,一次提升程序的运行效率.
2. 实现方案:多进程,多线程

进程(process)

进程理论基础

定义:程序在计算机中的一次运行.

• 程序是一个可执行的文件,是静态的战友磁盘.
• 进程是一个动态的过程描述,占有计算机运行资源,有必定的生命周期

系统中如何产生一个进程

1. 用户空间经过调用程序接口或者命令发起请求
2. 操做系统接受用户请求,开始建立进程
3. 操做系统调配计算机资源,肯定进程状态等
4. 操做系统将建立的进程提供给用户使用

 

进程基本概念

• CPU时间片:若是一个进程占有CPU内核则称这个进程在cpu时间片上.
• PCB(进程控制块):在内存中开辟的一块空间,用于存放进程的基本信息,也用于系统查找识别进程
• 进程ID(PID):系统为每一个进程分配的一个大于0的整数,做为进程ID.每一个ID不重复

　　　　Linux查看进程ID:ps - aux

• 父子进程:系统中每个进程(除了系统初始化进程)都有惟一的父进程,能够有0个或者多个子进程.父子进程关系便于进程管理

　　　　查看进程树: pstree

• 进程状态
• 三态

1. 就绪态:进程具有执行条件,等待分配CPU资源
2. 运行态:进程占有CPU时间片正在运行
3. 等待态:进程暂时中止运行,让出CPU

 五态(在三态基础上增长新建和终止)

• 新建: 建立一个进程,获取资源过程
• 终止:进程结束,释放资源过程

状态查看命令: ps - aux --->STAT列

S---等待态

R---执行态

Z---僵尸

+---前台进程

l---有多线程

进程的运行特征

1. 多进程能够更充分使用计算机多核资源
2. 进程之间的运行互不影响,各自独立
3. 每一个进程拥有独立的空间,各自使用本身空间资源

面试要求:

　　什么是进程,进程和程序有什么区别?

　　进程有哪些状态,状态之间如何转化?

基于fork的多进程编程

fork使用

 pdi = os.fork() 

功能:建立新的进程

返回值:整数,若是建立进程失败返回一个负数,若是成功则在原有进程中返回新进程的PID,在新进程中返回0

注意:

• 子进程会复制父进程所有内存空间,从fork下一句开始执行
• 父子进程各自独立运行,运行顺序不必定
• 利用父子进程fork返回值的区别,配合if结构让父子进程执行不一样的内容几乎是固定搭配.
• 父子进程有各自特有特征好比PID,PCB命令集等.
• 父进程fork以前开辟的空间子进程一样拥有,父子进程对各自空间的操做不会相互影响

fork进程代码示例:

 1 import os
 2 from time import sleep
 3 
 4 # 建立子进程
 5 pid = os.fork()
 6 
 7 if pid < 0:
 8     print("Create process failed")
 9 elif pid == 0:
10     # 只有子进程执行
11     sleep(3)
12     print("The new process")
13 else:
14     # 只有父进程执行
15     sleep(4)
16     print("The old process")
17 
18 # 父子进程都执行
19 print("process test over")

View Code

 

fork进程代码示例细节:

 1 import os
 2 from time import sleep
 3 
 4 print("=========================")
 5 a = 1
 6 def fun():
 7     print("fun .... ")
 8 
 9 pid = os.fork()
10 
11 if pid < 0:
12     print("Create process failed")
13 elif pid == 0:
14     print("Child process")
15     print("a = ",a)  # 从父进程空间拷贝了变量
16     fun()
17     a = 10000  # 只是修改了本身空间的a
18 else:
19     sleep(1)
20     print("Parent process")
21     print("a:",a)
22 
23 print("All a ->",a)

View Code

 

进程相关函数

 os.getpid() 

功能:获取一个进程的PID值

返回值:返回当前进程的PID

 os.getppid() 

功能:获取父进程的PID值

返回值:返回父进程PID

 os._exit(status) 

功能:结束一个进程

参数:进程的终止状态

 sys.exit([status]) 

功能:退出进程

参数:整数　　表示退出状态

　　字符串　　表示退出时打印内容

获取进程pid号代码示例:

 1 import os
 2 from time import sleep
 3 
 4 pid = os.fork()
 5 
 6 if pid < 0:
 7     print("Error")
 8 elif pid == 0:
 9     sleep(1)
10     print("Child PID:",os.getpid()) # 本身pid
11     print("Get parent PID:",os.getppid()) # 父pid
12 else:
13     print("Parent PID:", os.getpid())  # 本身pid
14     print("Get child PID:",pid)

View Code

 

进程退出代码示例:

 1 import os,sys
 2 
 3 pid = os.fork()
 4 
 5 # 父子进程退出不会影响对方继续执行
 6 if pid < 0:
 7     print("Error")
 8 elif pid == 0:
 9     # os._exit(0) # 子进程退出
10     print("Child process")
11 else:
12     sys.exit("退出父进程")
13     print("Parent process")

View Code

 

孤儿和僵尸

• 孤儿进程:父进程先于子进程退出,此时子进程成为孤儿进程

　　特色:孤儿进程会被系统进程收养,此时系统进程就会成为孤儿进程新的父进程,孤儿进程退出该进程会自动处理

• 僵尸进程:子进程先与父进程退出,父进程又没有处理子进程的退出状态,此时子进程就会称为僵尸进程

　　特色:僵尸进程虽然结束,可是会存留部分PCD在内存中,大量的僵尸进程会浪费系统的内存资源

• 如何避免僵尸进程产生
• 使用wait函数处理子进程退出

模拟僵尸进程产生以及处理示例:

 1 import os,sys
 2 import signal
 3 
 4 # 忽略子进程的退出行为，子进程退出自动由系统处理
 5 signal.signal(signal.SIGCHLD,signal.SIG_IGN)
 6 
 7 pid = os.fork()
 8 if pid < 0:
 9     print("Error")
10 elif pid == 0:
11     print("Child PID:",os.getpid())
12     sys.exit(2)
13 else:
14     """
15     os.wait() 处理僵尸
16     """
17     # pid,status = os.wait()
18     # print("pid:",pid)
19     # print('status:',os.WEXITSTATUS(status))
20     while True: # 让父进程不退出
21         pass

View Code

 

• 建立二级子进程处理僵尸

1. 父进程建立子进程,等待回收子进程
2. 子进程建立二级子进程而后退出
3. 二级子进程称为孤儿进程,和原来父进程一同执行事件

代码示例:

 1 import os
 2 from time import sleep
 3 
 4 def f1():
 5     for i in range(3):
 6         sleep(2)
 7         print("写代码")
 8 
 9 def f2():
10     for i in range(2):
11         sleep(4)
12         print("测代码")
13 
14 pid = os.fork()
15 if pid == 0:
16     p = os.fork()  # 建立二级子进程
17     if p == 0:
18         f1()
19     else:
20         os._exit(0)  # 一级子进程退出
21 else:
22     os.wait()  # 等待回收一级子进程
23     f2()

View Code

 

经过信号处理子进程退出

原理:子进程退出时会发送信号给父进程,若是父进程忽略子进程信号,则系统就会自动处理子进程退出

方法:使用signal模块在父进程建立子进程前写以下语句:

 import signal

signal.signal(signal.SIGCHLD,signal.SIG_IGN) 

特色:非阻塞,不会影响父进程运行,能够处理全部子进程退出

multiprocessing模块建立进程

进程建立方法

• 流程特色

1. 将须要子进程执行的事件封装为函数
2. 经过模块的Process类建立进程对象,关联函数
3. 能够经过进程对象设置进程信息及属性
4. 经过进程对象调用start启动进程
5. 经过进程对象调用join回收进程

• 基本接口使用

Process()
功能 ： 建立进程对象
参数 ： target 绑定要执行的目标函数 
    args 元组，用于给target函数位置传参
    kwargs 字典，给target函数键值传参

p.start()
功能 ： 启动进程

注意:启动进程此时target绑定函数开始执行,该函数做为子进程执行内容,此时进程真正被建立

p.join([timeout])
功能：阻塞等待回收进程
参数：超时时间

注意:

• 使用multiprocessing建立进程一样是子进程复制父进程空间代码段，父子进程运行互不影响。

• 子进程只运行target绑定的函数部分，其他内容均是父进程执行内容。

• multiprocessing中父进程每每只用来建立子进程回收子进程，具体事件由子进程完成。

• multiprocessing建立的子进程中没法使用标准输入

进程对象

p.name　　       进程名称

p.pid　　　　   对应子进程的PID号

p.is_alive()　　查看子进程是否在生命周期

p.daemon　　  设置父子进程的退出关系

• 　　若是设置为True则子进程会随父进程的退出而结束
• 　　要求必须在start()前设置
• 　　若是daemon设置成True一般就不会使用join()

进程对象代码示例:

 1 from multiprocessing import Process
 2 import time
 3 
 4 def tm():
 5     for i in range(3):
 6         print(time.ctime())
 7         time.sleep(2)
 8 
 9 p = Process(target = tm,name = 'Tarena')
10 
11 # 父进程退出，其全部子进程也退出
12 p.daemon = True
13 
14 p.start()  # 进程真正产生
15 
16 print("Name:",p.name)  # 进程名
17 print("PID：",p.pid) # pid号
18 print("is alive:",p.is_alive()) # 是否在生命周期

View Code

 

自定义进程类

• 建立步骤

1. 继承Process类
2. 重写__init__方法添加本身的属性,使用super()加载父类属性
3. 重写run()方法

• 使用方法

1. 实例化对象
2. 调用start自动执行run方法
3. 调用join()回收进程

自定义进程类代码示例:

 1 from multiprocessing import Process
 2 from time import *
 3 
 4 # 自定义进程类
 5 class MyProcess(Process):
 6     def __init__(self,value):
 7         self.value = value
 8         super().__init__() # 加载父类init
 9 
10     def f1(self):
11         print("步骤1")
12     def f2(self):
13         print("步骤2")
14 
15     # 做为流程启动函数
16     def run(self):
17         for i in range(self.value):
18             self.f1()
19             self.f2()
20 
21 if __name__ == '__main__':
22     p = MyProcess(2)
23     p.start()
24     p.join()

View Code

 

进程池实现

• 必要性

1. 进程的建立和销毁过程消耗的资源较多
2. 当任务量众多,每一个任务在很短期内完成时,须要频繁的建立和销毁进程.此时对计算机压力较大
3. 进程池技术很好的解决了以上问题

• 原理

　　建立必定数量的进程来处理事件,事件处理完进程不退出而是继续处理其余时间,直到全部事件全都处理完毕统一销毁.增长进程的重复利用,下降资源消耗.

• 进程池实现

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()
功能： 回收进程池中进程

进程池代码示例:

 1 from multiprocessing import Pool
 2 from time import sleep,ctime
 3 
 4 # 进程池事件
 5 def worker(msg):
 6     sleep(2)
 7     print(ctime(),'--',msg)
 8 
 9 # 建立进程池
10 pool = Pool(4)
11 
12 # 向进程池队列中添加事件
13 for i in range(10):
14     msg = "Tedu %d"%i
15     pool.apply_async(func=worker,args=(msg,))
16 
17 # 关闭进程池
18 pool.close()
19 
20 # 回收进程池
21 pool.join()

View Code

 

管道通讯

• 通讯原理

在内存中开辟管道空间,生成管道操做对象,多个进程使用同一管道对象,进行读写便可实现通讯

• 实现方法

from  multiprocessing import Pipe

fd1,fd2 = Pipe(duplex = True) 功能: 建立管道 参数：默认表示双向管道 若是为False 表示单向管道 返回值：表示管道两端的读写对象 若是是双向管道都可读写 若是是单向管道fd1只读 fd2只写 fd.recv() 功能 ： 从管道获取内容 返回值：获取到的数据 fd.send(data) 能够是任何内容,如:整数,浮点数,字符串等 功能： 向管道写入内容 参数： 要写入的数据

管道操做

注意:

1. 　　multiprocessing中提供的通讯只用于亲缘关系进程间通讯
2. 　　管道在父进程中建立,子进程从父进程中获取管道对象

管道操做代码示例:

 1 from multiprocessing import Process,Pipe
 2 
 3 # 建立管道对象
 4 # 参数False 表示fd1 只能 recv , fd2 只能 send
 5 fd1,fd2 = Pipe()
 6 
 7 # APP1可使用app2提供的信息登陆
 8 def app1():
 9     print("启动app1，请登陆")
10     print("请求app2受权")
11     # 写管道
12     fd1.send("app1 能够用你的帐号登陆吗？")
13     data = fd1.recv()
14     if data:
15         print("登陆成功：",data)
16 
17 def app2():
18     request = fd2.recv() # 阻塞等待读取管道
19     print(request)
20     fd2.send(('Joy','123')) # 发送python数据类型
21 
22 p1 = Process(target=app1)
23 p2 = Process(target=app2)
24 p1.start()
25 p2.start()
26 p1.join()
27 p2.join()
28 
29 代码示例

View Code

 

消息队列

• 通讯原理

　　在内存中创建队列模型,进程经过队列将消息存入,或者从队列取出完成进程间通讯.

• 实现方法

from multiprocessing import Queue

q = Queue(maxsize=0)
功能: 建立队列对象
参数：最多存放消息个数
返回值：队列对象

q.put(data,[block,timeout])
功能：向队列存入消息
参数：data  要存入的内容
block  设置是否阻塞 False为非阻塞
timeout  超时检测

q.get([block,timeout])
功能：从队列取出消息
参数：block  设置是否阻塞 False为非阻塞
timeout  超时检测
返回值： 返回获取到的内容

q.full()   判断队列是否为满
q.empty()  判断队列是否为空
q.qsize()  获取队列中消息个数
q.close()  关闭队列

消息队列代码示例:

 1 from multiprocessing import Queue,Process
 2 from time import sleep
 3 from random import randint
 4 
 5 # 建立队列
 6 q = Queue(5) # 最大存储5个消息
 7 
 8 def request():
 9     for i in range(10):
10         x = randint(1,100)
11         y = randint(1,100)
12         q.put((x,y))  # 写消息队列
13         print("=============")
14 
15 def handle():
16     while not q.empty():
17         data = q.get() # 读消息队列
18         print("x + y = ",data[0]+data[1])
19         sleep(2)
20 
21 p1 = Process(target=request)
22 p2 = Process(target=handle)
23 p1.start()
24 p2.start()
25 p1.join()
26 p2.join()
27 
28 代码示例

View Code

 

 共享内存

• 通讯原理:在内存中开辟一块空间,进程能够写入内容和读取内容完成通讯,可是每次写入内容会覆盖以前内容
• 实现方法

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直接打印共享内存中的字节串

共享内存代码示例:

注意:共享内存中只能有一个值

 1 from multiprocessing import Process,Value
 2 import time
 3 import random
 4 
 5 # 建立共享内存
 6 money = Value('i',5000)
 7 
 8 def man():
 9     for i in range(30):
10         time.sleep(0.2)
11         # 修改共享内存
12         money.value += random.randint(1,1000)
13 
14 def girl():
15     for i in range(30):
16         time.sleep(0.15)
17         money.value -= random.randint(100,800)
18 
19 p1 = Process(target=man)
20 p2 = Process(target=girl)
21 p1.start()
22 p2.start()
23 p1.join()
24 p2.join()
25 print("一个月余额：",money.value) #读取共享内存

View Code

 

 共享内存存放列表,字节串:

 1 from multiprocessing import Process,Array
 2 
 3 # 建立共享内存，初始数据 [1,2,3,4]
 4 # shm = Array('i',[1,2,3,4])
 5 # shm = Array('i',4) # 开辟4个整形的列表空间
 6 shm = Array('c',b'hello')
 7 
 8 def fun():
 9     # 共享内存对象能够迭代
10     for i in shm:
11         print(i)
12     shm[0] = b'H' # 修改共享内存
13 
14 p = Process(target=fun)
15 p.start()
16 p.join()
17 for i in shm:
18     print(i)
19 print(shm.value) # 总体打印字节串

View Code

 

信号量(信号灯集)

• 通讯原理

　　给定一个数量对多个进程可见.多个进程均可以操做改数量增减,并根据数量值决定本身的行为

• 实现方法

from multiprocessing import Semaphore

sem = Semaphore(num)
功能 ： 建立信号量对象
参数 ： 信号量的初始值
返回值 ： 信号量对象

sem.acquire()  将信号量减1 当信号量为0时阻塞
sem.release()  将信号量加1
sem.get_value() 获取信号量数量

信号量代码示例:

注意: 信号量能够当作是一种资源，执行任务须要消耗信号量资源，
这样能够控制进程执行行为

 1 from multiprocessing import Process,Semaphore
 2 from time import sleep
 3 import os
 4 
 5 # 建立信号量资源
 6 sem = Semaphore(3)
 7 
 8 # 任务函数 (系统中最多可以同时运行三个该任务)
 9 def handle():
10     sem.acquire() # 消耗一个信号量
11     print("%s执行任务"%os.getpid())
12     sleep(2)
13     print("%s 拯救了宇宙"%os.getpid())
14     sem.release() # 增长一个信号量
15 
16 jobs = []
17 for i in range(10):
18     p = Process(target = handle)
19     jobs.append(p)
20     p.start()
21 
22 for i in jobs:
23     i.join()

View Code