python网络-多进程(21)

1、什么是进程

进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操做系统结构的基础。在早期面向进程设计的计算机结构中，进程是程序的基本执行实体；在当代面向线程设计的计算机结构中，进程是线程的容器。程序是指令、数据及其组织形式的描述，进程是程序的实体。---来自百度百科

狭义定义：进程是正在运行的程序的实例（an instance of a computer program that is being executed）。

广义定义：进程是一个具备必定独立功能的程序关于某个数据集合的一次运行活动。它是操做系统动态执行的基本单元，在传统的操做系统中，进程既是基本的分配单元，也是基本的执行单元。

 

进程的概念主要有两点：第一，进程是一个实体。每个进程都有它本身的地址空间，通常状况下，包括文本区域（text region）、数据区域（data region）和堆栈（stack region）。文本区域存储处理器执行的代码；数据区域存储变量和进程执行期间使用的动态分配的内存；堆栈区域存储着活动过程调用的指令和本地变量。第二，进程是一个“执行中的程序”。程序是一个没有生命的实体，只有处理器赋予程序生命时（操做系统执行之），它才能成为一个活动的实体，咱们称其为进程。

 

2、程序和进程的关系

编写完毕的代码，在没有运⾏的时候，称之为程序

正在运⾏着的代码，就成为进程

进程除了包含代码之外还有须要运⾏的环境等因此和程序是有区别的

 

3、fork()

fork（）函数只能够在Linux和Mac系统中，在windows中不能够用，因此它使用的也比较少

#-*- coding:utf-8 -*-
import os pid = os.fork() if pid == 0: print("子进程") else: print("主进程")

运行结果为：

主进程
子进程

 getpid()、getppid()

import os pid = os.fork() if pid == 0: print("我是子进程(%d),个人父进程(%d)"%(os.getpid(),os.getppid())) else: print("我是父进程(%d),个人子进程(%d)"%(os.getpid,pid)) print("父子进程均可以执行的代码")

运行结果为：

我是父进程(4488),个人子进程(4491） 父子进程均可以执行的代码 我是子进程(4491),个人父进程(4488) 父子进城均可以执行的代码

说明：

• 程序执⾏到os.fork()时，操做系统会建立⼀个新的进程（⼦进程），而后复制⽗进程的全部信息到⼦进程中
• 而后⽗进程和⼦进程都会从fork()函数中获得⼀个返回值，在⼦进程中这个值⼀定是0，⽽⽗进程中是⼦进程的id号
• 普通的函数调⽤，调⽤⼀次，返回⼀次，可是fork()调⽤⼀次，返回两次，由于操做系统⾃动把当前进程（称为⽗进程）复制了⼀份（称为⼦进程），而后，分别在⽗进程和⼦进程内返回
• ⼀个⽗进程能够fork出不少⼦进程，因此，⽗进程要记下每一个⼦进程的ID，⽽⼦进程只须要调⽤getppid()就能够拿到⽗进程的ID

多个fork()

#-*- coding:utf-8 -*-
import os pid1 = os.fork() if pid1 == 0:#子进程1
    print("1:我是第一个子进程%d,个人父进程是%d"%(os.getpid(),os.getppid())) else:#父进程
    print("2:我是父进程%d"%os.getpid())  pid2 = os.fork() if pid2==0: print("3:我是谁%d,个人父进程是%d"%(os.getpid(),os.getppid())) else: print("4:我是谁%d,个人父进程是%d"%(os.getpid(),os.getppid()))

运行结果为：

2:我是父进程3189 1:我是第一个子进程3190,个人父进程是3189 4:我是谁3190,个人父进程是3189 3:我是谁3191,个人父进程是3189 3:我是谁3192,个人父进程是3190 4:我是谁3189,个人父进程是991

说明：

• pid2开辟的进程将会被子进程1和父进程同时调用
• 当父线程调用pid2
  • if pid2 == 0:会在建立一个子进程2，父进程是主进程 　　
  • else:及父线程自己，不会再建立进程
• 当子进程1调用pid2
  • if pid2 ==0:会建立一个子子进程，父进程是子进程1
  • else：即子线程1自己，不会再建立进程

其实上面的代码就至关于：

#-*- coding:utf-8 -*-
import os pid1 = os.fork() if pid1 == 0:#子进程1
    print("1:我是第一个子进程%d,个人父进程是%d"%(os.getpid(),os.getppid())) else:#父进程
    print("2:我是父进程%d"%os.getpid()) pid2 = os.fork() if pid1 == 0:#子进程1 if pid2==0:#子子进程 print("3:我是谁%d,个人父进程是%d"%(os.getpid(),os.getppid())) else:#子进程1 print("4:我是谁%d,个人父进程是%d"%(os.getpid(),os.getppid())) else:#父进程 if pid2==0:#子进程2 print("3:我是谁%d,个人父进程是%d"%(os.getpid(),os.getppid())) else:#父进程 print("4:我是谁%d,个人父进程是%d"%(os.getpid(),os.getppid()))

 

4、多进程使用全局变量

import os import time g_num = 100 ret = os.fork() if ret == 0: print("----process-1----") g_num += 1
    print("---process-1 g_num=%d---"%g_num) else: time.sleep(3) print("----process-2----") print("---process-2 g_num=%d---"%g_num)

运行结果为：

----process-1----
---process-1 g_num=101---
----process-2----
---process-2 g_num=100---

说明：多进程间全局变量是不共享的，每一个进程里面全局变量都是独自一份的

 

5、multiprocessing

因为Python是跨平台的，天然也应该提供一个跨平台的多进程支持。multiprocessing模块就是跨平台版本的多进程模块。

multiprocessing模块提供了一个Process类来表明一个进程对象

#coding=utf-8
from multiprocessing import Process import os #子进程要执行的代码
def sub_process(name): print("这是在子进程中，name=%s,pid=%d"%(name,os.getpid())) if __name__ == "__main__": print("父进程：%d"%os.getpid()) p=Process(target=sub_process,args=("test",)) print("----子进程将要开启----") p.start()#开启子进程
 p.join()#用于等待子进程执行完毕再继续往下执行 print("----子进程已经结束----")

运行结果为：

父进程：8344
----子进程将要开启---- 这是在子进程中，name=test,pid=9064
----子进程已经结束----

说明

• 建立子进程时，只须要传入一个执行函数和函数的参数，建立一个Process实例，用start()方法启动，这样建立进程比fork()还要简单。
• join()方法能够等待子进程结束后再继续往下运行，一般用于进程间的同步。

Process语法结构以下：

Process([group [, target [, name [, args [, kwargs]]]]])

 

1 group：参数未使用，值始终为None 2 target：表示调用对象，即子进程要执行的任务 3 args：表示调用对象的位置参数元组，args=(1,2,'a',) 4 kwargs：表示调用对象的字典,kwargs={'name':'a','age':18} 5 name：为子进程的名称

 

Process类经常使用方法：

 

1 start()：启动进程，并调用该子进程中的p.run() 
2 run():进程启动时运行的方法，正是它去调用target指定的函数，咱们自定义类的类中必定要实现该方法  
3 terminate():强制终止进程p，不会进行任何清理操做，若是p建立了子进程，该子进程就成了僵尸进程，使用该方法须要特别当心这种状况。若是p还保存了一个锁那么也将不会被释放，进而致使死锁
4 is_alive():若是p仍然运行，返回True
5 join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，须要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程 

 

Process类经常使用属性：

1 daemon：默认值为False，若是设为True，表明p为后台运行的守护进程，当p的父进程终止时，p也随之终止，而且设定为True后，p不能建立本身的新进程，必须在p.start()以前设置
2 name:进程的名称
3 pid：进程的pid
4 exitcode:进程在运行时为None、若是为–N，表示被信号N结束(了解便可)
5 authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络链接的底层进程间通讯提供安全性，这类链接只有在具备相同的身份验证键时才能成功（了解便可）

#coding=utf-8
from multiprocessing import Process import time import os #两个子进程将会调用的两个方法
print("1:%d"%os.getpid()) def worker_1(interval): print("worker_1：父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid())) t_start = time.time() time.sleep(interval) #程序将会被挂起interval秒
    t_end = time.time() print("worker_1,执行时间为'%0.2f'秒"%(t_end - t_start)) print("2:%d"%os.getpid()) def worker_2(interval): print("worker_2,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid())) t_start = time.time() time.sleep(interval) t_end = time.time() print("worker_2,执行时间为'%0.2f'秒"%(t_end - t_start)) #输出当前程序的ID
print("3:%d"%os.getpid())
if __name__=='__main__': print("4:%d"%os.getpid()) p1=Process(target=worker_1,args=(2,)) p2=Process(target=worker_2,name="Se7eN_HOU",args=(1,)) print("5:%d"%os.getpid()) p1.start() p2.start() #同时父进程仍然往下执行，若是p2进程还在执行，将会返回True
    print("p2.is_alive=%s"%p2.is_alive()) #输出p1和p2进程的别名和pid
    print("p1.name=%s"%p1.name) print("p1.pid=%s"%p1.pid) print("p2.name=%s"%p2.name) print("p2.pid=%s"%p2.pid) print("6:%d"%os.getpid()) #join括号中不携带参数，表示父进程在这个位置要等待p1进程执行完成后，再继续执行下面的语句，通常用于进程间的数据同步
 p1.join() print("p1.is_alive=%s"%p1.is_alive()) p2.join() print("7:%d"%os.getpid())

 

运行结果为：

1:10452
2:10452
3:10452
4:10452
5:10452 p2.is_alive=True p1.name=Process-1 p1.pid=10688 p2.name=Se7eN_HOU p2.pid=2192
6:10452 
1:2192
2:2192
3:2192
7:2192 worker_2,父进程(10452),当前进程(2192) worker_2,执行时间为'1.00'秒 1:10688
2:10688
3:10688
7:10688 worker_1：父进程(10452),当前进程(10688) worker_1,执行时间为'2.00'秒 p1.is_alive=False 7:10452

 

6、建立Process子类建立多进程

建立新的进程还可以使用类的方式，能够自定义一个类，继承Process类，每次实例化这个类的时候，就等同于实例化一个进程对象

from multiprocessing import Process import time import os #建立一个类，继承Process类
class My_Process(Process): def __init__(self,interval): #由于Process类自己也有__init__方法，这个子类至关于重写了这个方法，
        #但这样就会带来一个问题，咱们并无彻底的初始化一个Process类，因此就不能使用从这个类继承的一些方法和属性，
        #最好的方法就是将继承类自己传递给Process.__init__方法，完成这些初始化操做
        Process.__init__(self) self.interval = interval #重写了Process类的run()方法
    def run(self): print("子进程：%d,开始执行，父进程：%d"%(os.getpid(),os.getppid())) t_start = time.time() time.sleep(self.interval) t_stop = time.time() print("子进程：%d,执行结束，耗时%0.2f秒"%(os.getpid(),t_stop-t_start)) if __name__ == '__main__': t_start = time.time() print("当前进程是%d"%os.getpid()) p1 = My_Process(3) p1.start() p1.join() t_stop = time.time() print("当前进程%d执行结束，耗时：%0.2f"%(os.getpid(),t_stop-t_start))

运行结果为：

当前进程是9980 子进程：7084,开始执行，父进程：9980 子进程：7084执行结束，耗时3.00秒 当前进程9980执行结束，耗时：3.23

 

7、进程池Pool

当须要建立的子进程数量很少时，能够直接利用multiprocessing中的Process动态成生多个进程，但若是是上百甚至上千个目标，手动的去建立进程的工做量巨大，此时就能够用到multiprocessing模块提供的Pool方法。

初始化Pool时，能够指定一个最大进程数，当有新的请求提交到Pool中时，若是池尚未满，那么就会建立一个新的进程用来执行该请求；但若是池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会建立新的进程来执行

from multiprocessing import Pool import os import time import random def worker(msg): t_start = time.time() print("%d进程开始执行%d"%(os.getpid(),msg)) #random.random()随机生成0~1之间的浮点数
    time.sleep(random.random()*2) t_stop = time.time() print(msg,"执行完毕，耗时%0.2f"%(t_stop-t_start)) if __name__ == '__main__': po=Pool(3) #定义一个进程池，最大进程数3
    for i in range(0,10): #Pool.apply_async(要调用的目标,(传递给目标的参数元祖,))
        #每次循环将会用空闲出来的子进程去调用目标
 po.apply_async(worker,(i,)) print("----start----") po.close() #关闭进程池，关闭后po再也不接收新的请求
    po.join() #等待po中全部子进程执行完成，必须放在close语句以后
    print("-----end-----")

运行结果为：

----start---- 4353进程开始执行0 4354进程开始执行1 4355进程开始执行2 2,执行完毕，耗时0.20 4355进程开始执行3 1,执行完毕，耗时1.19 4354进程开始执行4 4,执行完毕，耗时0.37 4354进程开始执行5 0,执行完毕，耗时1.57 4353进程开始执行6 5,执行完毕，耗时0.19 4354进程开始执行7 3,执行完毕，耗时1.63 4355进程开始执行8 6,执行完毕，耗时0.49 4353进程开始执行9 8,执行完毕，耗时0.75
7,执行完毕，耗时0.90
9,执行完毕，耗时0.63
-----end-----

multiprocessing.Pool经常使用函数解析：

• apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
• apply(func[, args[, kwds]])：使用阻塞方式调用func
• close()：关闭Pool，使其再也不接受新的任务；
• terminate()：无论任务是否完成，当即终止；
• join()：主进程阻塞，等待子进程的退出， 必须在close或terminate以后使用；

apply堵塞式

from multiprocessing import Pool import os import time import random def worker(msg): t_start = time.time() print("%d进程开始执行%d"%(os.getpid(),msg)) #random.random()随机生成0~1之间的浮点数
    time.sleep(random.random()*2) t_stop = time.time() print(msg,"执行完毕，耗时%0.2f"%(t_stop-t_start)) if __name__ == '__main__': po=Pool(3) #定义一个进程池，最大进程数3
    for i in range(0,10): #Pool.apply_async(要调用的目标,(传递给目标的参数元祖,))
        #每次循环将会用空闲出来的子进程去调用目标
 po.apply(worker,(i,)) print("----start----") po.close() #关闭进程池，关闭后po再也不接收新的请求
    po.join() #等待po中全部子进程执行完成，必须放在close语句以后
    print("-----end-----")

运行结果为：

4400进程开始执行0 0,执行完毕，耗时1.89 4401进程开始执行1 1,执行完毕，耗时1.91 4402进程开始执行2 2,执行完毕，耗时1.64 4400进程开始执行3 3,执行完毕，耗时1.16 4401进程开始执行4 4,执行完毕，耗时1.85 4402进程开始执行5 5,执行完毕，耗时0.29 4400进程开始执行6 6,执行完毕，耗时0.19 4401进程开始执行7 7,执行完毕，耗时1.19 4402进程开始执行8 8,执行完毕，耗时0.61 4400进程开始执行9 9,执行完毕，耗时1.08
----start----
-----end-----

说明：经过运行结果能够看出来，阻塞式会等进程池中的进程都执行完毕了才会运行主进程的start和end的打印

 

8、进程间的通讯-Queue

1. Queue的使用

可使用multiprocessing模块的Queue实现多进程之间的数据传递，Queue自己是一个消息列队程序，首先用一个小实例来演示一下Queue的工做原理：

#-*- coding:utf-8 -*-
from multiprocessing import Queue #建立一个Queue对象，最多可接受三条put消息
q = Queue(3) q.put("消息1") q.put("消息2") print(q.full()) q.put("消息3") print(q.full()) try: q.put("消息4",True,2) except : print("消息队列已满，现有消息数量：%s"%q.qsize()) try: q.put_nowait("消息5") except : print("消息队列已满，现有消息数量：%s"%q.qsize()) #推荐方式，先判断消息队列是否已满，在写入
if not q.full(): q.put_nowait("消息6") #读取消息时，先判断消息队列是否为空，在读取
if not q.empty(): for i in range(q.qsize()): print(q.get_nowait())

运行结果为：

False True 消息队列已满，现有消息数量：3 消息队列已满，现有消息数量：3 消息1 消息2 消息3

说明

初始化Queue()对象时（例如：q=Queue()），若括号中没有指定最大可接收的消息数量，或数量为负值，那么就表明可接受的消息数量没有上限（直到内存的尽头）；

• Queue.qsize()：返回当前队列包含的消息数量；
• Queue.empty()：若是队列为空，返回True，反之False ；
• Queue.full()：若是队列满了，返回True,反之False；
• Queue.get([block[, timeout]])：获取队列中的一条消息，而后将其从列队中移除，block默认值为True；

1）若是block使用默认值，且没有设置timeout（单位秒），消息列队若是为空，此时程序将被阻塞（停在读取状态），直到从消息列队读到消息为止，若是设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出"Queue.Empty"异常；

2）若是block值为False，消息列队若是为空，则会马上抛出"Queue.Empty"异常；

• Queue.get_nowait()：至关Queue.get(False)；

• Queue.put(item,[block[, timeout]])：将item消息写入队列，block默认值为True；

1）若是block使用默认值，且没有设置timeout（单位秒），消息列队若是已经没有空间可写入，此时程序将被阻塞（停在写入状态），直到从消息列队腾出空间为止，若是设置了timeout，则会等待timeout秒，若还没空间，则抛出"Queue.Full"异常；

2）若是block值为False，消息列队若是没有空间可写入，则会马上抛出"Queue.Full"异常；

• Queue.put_nowait(item)：至关Queue.put(item, False)；

2. Queue实例

咱们以Queue为例，在父进程中建立两个子进程，一个往Queue里写数据，一个从Queue里读数据：

from multiprocessing import Process from multiprocessing import Queue import os import time import random #写数据进程执行的代码
def write(q): for value in ["A","B","C"]: print("Put %s to Queue "%value) q.put(value) time.sleep(random.random()) #读取数据进程的代码 
def read(q): while True: if not q.empty(): value = q.get(True) print("Get %s from Queue "%value) time.sleep(random.random()) else: break

if __name__ == '__main__': #父进程建立Queue，并传递给各个子进程
    q = Queue() pw = Process(target = write,args=(q,)) pr = Process(target = read,args=(q,)) #启动子进程pw，写入
 pw.start() #等待pw结束
 pw.join() #启动子进程pr，读取
 pr.start() pr.join() print("全部数据都写入而且读完")

运行结果为：

Put A to Queue Put B to Queue Put C to Queue Get A from Queue Get B from Queue Get C from Queue 全部数据都写入而且读完

3. 进程池中的Queue

若是要使用Pool建立进程，就须要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，不然会获得一条以下的错误信息：

RuntimeError: Queue objects should only be shared between processes through inheritance.

#coding=utf-8
from multiprocessing import Manager from multiprocessing import Pool import os import time import random def reader(q): print("reader启动(%d),父进程为(%d)"%(os.getpid(),os.getppid())) for i in range(q.qsize()): print("reader从Queue获取到的消息时：%s"%q.get(True)) def writer(q): print("writer启动(%d),父进程为(%d)"%(os.getpid(),os.getppid())) for i in "Se7eN_HOU": q.put(i) if __name__ == '__main__': print("-------(%d) Start-------"%os.getpid()) #使用Manager中的Queue来初始化
    q = Manager().Queue() po = Pool() #使用阻塞模式建立进程，这样就不须要在reader中使用死循环了，可让writer彻底执行完成后，再用reader去读取
 po.apply(writer,(q,)) po.apply(reader,(q,)) po.close() po.join() print("-------(%d) End-------"%os.getpid())

运行结果为：

-------(880) Start------- writer启动(7744),父进程为(880) reader启动(7936),父进程为(880) reader从Queue获取到的消息时：S reader从Queue获取到的消息时：e reader从Queue获取到的消息时：7 reader从Queue获取到的消息时：e reader从Queue获取到的消息时：N reader从Queue获取到的消息时：_ reader从Queue获取到的消息时：H reader从Queue获取到的消息时：O reader从Queue获取到的消息时：U -------(880) End-------