python网络-多进程(21)

2023-01-31 00:31

短信预约 -IT技能 免费直播动态提醒

一、什么是进程

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

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

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

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

二、程序和进程的关系

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

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

进程除了包含代码以外还有需要运⾏的环境等所以和程序是有区别的

三、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()))

四、多进程使用全局变量

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---

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

五、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

六、创建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

七、进程池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的打印

八、进程间的通信-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-------

免责声明：

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

进程网络 python

阅读原文内容投诉