VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > temp > 简明python教程 >
  • 【2020Python修炼记】python并发编程(三)多进程-应用部分

一、 multiprocessing模块介绍

# python中的多线程无法利用多核优势,如果想要充分地使用多核CPU的资源(os.cpu_count()查看),在python中大部分情况需要使用多进程

# Python提供了multiprocessing 模块——

# 作用:multiprocessing模块用来开启子进程,并在子进程中执行我们定制的任务(比如函数),该模块与多线程模块threading的编程接口类似

# 功能:multiprocessing模块的功能众多:支持子进程、通信和共享数据、执行不同形式的同步,提供了Process、Queue、Pipe、Lock等组件。

PS:需要再次强调的一点是:与线程不同,进程没有任何共享状态,进程修改的数据,改动仅限于该进程内。

 

 

 

二、 process类的介绍

 1、创建进程的 类 process

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

由该类实例化得到的对象,表示一个子进程中的任务(尚未启动)

强调:

1、需要使用关键字的方式来指定参数  eg: Process(target=task, args=('jason',))

2、group 参数未使用,值始终为None

3、target 表示调用对象,即子进程要执行的任务

4、args 表示调用对象target函数的位置参数元组,args=(1,2,'egon',)  ,(元组形式,用逗号隔开元素,末尾一定必须有逗号)

5、kwargs 表示调用对象的字典,kwargs={'name':'egon','age':18}

6、name 为子进程的名称

 

 

 

2、方法介绍

1、p.start()

启动进程,并调用该子进程中的p.run()
2、p.run()

进程启动时运行的方法,正是它去调用target指定的函数,我们自定义类的类中一定要实现该方法

3、p.terminate()

强制终止进程p,不会进行任何清理操作,如果p创建了子进程,该子进程就成了僵尸进程,使用该方法需要特别小心这种情况。

如果p还保存了一个锁那么也将不会被释放,进而导致死锁

4、p.is_alive()   

如果p仍然运行,返回True

5、p.join([timeout])

主线程等待p终止(强调:是主线程处于等的状态,而p是处于运行的状态)。

timeout是可选的超时时间,需要强调的是,p.join只能join住start开启的进程,而不能join住run开启的进程

 

 

 

3、属性介绍

1、p.daemon

默认值为False,如果设为True,代表p为后台运行的守护进程,当p的父进程终止时,p也随之终止,

并且设定为True后,p不能创建自己的新进程,必须在p.start()之前设置

2、p.name     进程的名称

3、p.pid      当前进程的pid

4、p.ppid   当前进程的父进程的pid

5、p.exitcode

 进程在运行时为None、如果为–N,表示被信号N结束(了解即可)

6、p.authkey

进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。

这个键的用途是为涉及网络连接的底层进程间通信提供安全性,这类连接只有在具有相同的身份验证键时才能成功(了解即可)

 

 

 

三、 process类的使用

# windows操作系统下 创建进程一定要在main内创建,

因为windows下创建进程类似于模块导入的方式,会从上往下依次执行代码

# linux中则是直接将代码完整地拷贝一份

 

 

注意:在windows中,Process() 必须放到 # if __name__ == '__main__':

 

 

1、创建进程的两种方式

 (1)、类实例化产生对象

 

复制代码
# 第一种
from multiprocessing import Process
import time

def task(name):
    print('%s is running'%name)
    time.sleep(3)
    print('%s is over'%name)


if __name__ == '__main__':
    # 1 创建一个对象
    p = Process(target=task, args=('jason',))
    # 容器类型哪怕里面只有1个元素 建议要用逗号隔开
    # 2 开启进程
    p.start()  # 告诉操作系统帮你创建一个进程  异步
    print('')
复制代码

 

(2)、类的继承  run方法

 

复制代码
# 第二种方式 类的继承
from multiprocessing import Process
import time


class MyProcess(Process):
    def run(self):
        print('hello bf girl')
        time.sleep(1)
        print('get out!')


if __name__ == '__main__':
    p = MyProcess()
    p.start()
    print('')
复制代码

 

2、进程之间的内存空间是隔离的

 

复制代码
from multiprocessing import Process
n=100 #在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就可以了
def work():
    global n
    n=0
    print('子进程内: ',n)


if __name__ == '__main__':
    p=Process(target=work)
    p.start()
    print('主进程内: ',n)
复制代码

 

3、Process对象的join方法——join:是主进程在等,等待子进程结束

 

 例子一
复制代码
from multiprocessing import Process
import time,os

def task():
    print('%s is running' %os.getpid())
    time.sleep(3)
    
if __name__ == '__main__':
    p=Process(target=task)
    p.start()
    p.join()  # 等待进程p结束后,join函数内部会发送系统调用wait,去告诉操作系统回收掉进程p的id号

    print(p.pid) # ???此时能否看到子进程p的id号
    print('')


#  ————————————————————————————

#答案:可以
#分析:
p.join()是向操作系统发送请求,告知操作系统p的id号不需要再占用了,回收就可以,
此时在父进程内还可以看到p.pid,但此时的p.pid是一个无意义的id号,因为操作系统已经将该编号回收

打个比方:
我党相当于操作系统,控制着整个中国的硬件,每个人相当于一个进程,每个人都需要跟我党申请一个身份证号
该号码就相当于进程的pid,人死后应该到我党那里注销身份证号,p.join()就相当于要求我党回收身份证号,但p的家人(相当于主进程)
仍然持有p的身份证,但此刻的身份证已经没有意义
复制代码

 

4、Process对象的其他方法或属性(了解)

 

复制代码
from multiprocessing import Process
import time
import random

class Piao(Process):
    def __init__(self,name):
        self.name=name
        super().__init__()

    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randrange(1,5))
        print('%s is piao end' %self.name)


p1=Piao('egon1')
p1.start()

p1.terminate()#关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活
print(p1.is_alive()) #结果为True

print('开始')
print(p1.is_alive()) #结果为False
复制代码
复制代码
from multiprocessing import Process
import time
import random
class Piao(Process):
    def __init__(self,name):
        # self.name=name
        # super().__init__() #Process的__init__方法会执行self.name=Piao-1,
        #                    #所以加到这里,会覆盖我们的self.name=name

        #为我们开启的进程设置名字的做法
        super().__init__()
        self.name=name

    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randrange(1,3))
        print('%s is piao end' %self.name)

p=Piao('egon')
p.start()
print('开始')
print(p.pid) #查看pid
复制代码

 

四、僵尸进程和孤儿进程

1、僵尸进程——“死了没死透”

僵尸进程(有害)

一个进程使用fork创建子进程,如果子进程退出,而父进程并没有调用wait或waitpid获取子进程的状态信息,那么子进程的进程描述符仍然保存在系统中。这种进程称之为僵死进程。

详解:我们知道在unix/linux中,正常情况下子进程是通过父进程创建的,子进程在创建新的进程。子进程的结束和父进程的运行是一个异步过程,即父进程永远无法预测子进程到底什么时候结束,如果子进程一结束就立刻回收其全部资源,那么在父进程内将无法获取子进程的状态信息。

 

 
复制代码
#coding:utf-8
from multiprocessing import Process
import time,os

def run():
    print('',os.getpid())

if __name__ == '__main__':
    p=Process(target=run)
    p.start()
    
    print('',os.getpid())
    time.sleep(1000)
复制代码

2、孤儿进程——“没爹没娘”

 孤儿进程(无害)

一个父进程退出,而它的一个或多个子进程还在运行,那么那些子进程将成为孤儿进程。孤儿进程将被init进程(进程号为1)所收养,并由init进程对它们完成状态收集工作。    

孤儿进程是没有父进程的进程,孤儿进程这个重任就落到了init进程身上,init进程就好像是一个民政局,专门负责处理孤儿进程的善后工作。每当出现一个孤儿进程的时候,内核就把孤 儿进程的父进程设置为init,而init进程会循环地wait()它的已经退出的子进程。这样,当一个孤儿进程凄凉地结束了其生命周期的时候,init进程就会代表党和政府出面处理它的一切善后工作。因此孤儿进程并不会有什么危害。

 

 
孤儿进程-测试

 

五 、守护进程

 主进程创建守护进程——“皇帝(主进程)驾崩,好多陪葬(守护进程)”

  其一:守护进程会在主进程代码执行结束后就终止

  其二:守护进程内无法再开启子进程,

     否则抛出异常:AssertionError: daemonic processes are not allowed to have children

注意:进程之间是互相独立的,主进程代码运行结束,守护进程随即终止

 

 
复制代码
from multiprocessing import Process
import time
import random

class Piao(Process):
    def __init__(self,name):
        self.name=name
        super().__init__()
    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randrange(1,3))
        print('%s is piao end' %self.name)


p=Piao('egon')
p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
p.start()
print('')
复制代码
复制代码
#主进程代码运行完毕,守护进程就会结束
from multiprocessing import Process
from threading import Thread
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")


p1=Process(target=foo)
p2=Process(target=bar)

p1.daemon=True
p1.start()
p2.start()
print("main-------") #打印该行则主进程代码结束,则守护进程p1应该被终止,可能会有p1任务执行的打印信息123,因为主进程打印main----时,p1也执行了,但是随即被终止
复制代码

 

六 、进程同步(互斥锁)

进程之间数据不共享,但是共享同一套文件系统,所以访问 同一个文件,或 同一个打印终端,是没有问题的,

而共享带来的是竞争,竞争带来的结果就是错乱,如何控制——加锁处理——牺牲效率,保证有序和数据安全

 

 

1、多个进程共享同一打印终端

 

复制代码
#并发运行,效率高,但竞争同一打印终端,带来了打印错乱
from multiprocessing import Process
import os,time
def work():
    print('%s is running' %os.getpid())
    time.sleep(2)
    print('%s is done' %os.getpid())

if __name__ == '__main__':
    for i in range(3):
        p=Process(target=work)
        p.start()
复制代码
复制代码
#由并发变成了串行,牺牲了运行效率,但避免了竞争
from multiprocessing import Process,Lock
import os,time
def work(lock):
    lock.acquire()
    print('%s is running' %os.getpid())
    time.sleep(2)
    print('%s is done' %os.getpid())
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    for i in range(3):
        p=Process(target=work,args=(lock,))
        p.start()
复制代码

 

2、多个进程共享同一文件-- 文件当数据库,模拟抢票

 

复制代码
#文件db的内容为:{"count":1}
#注意一定要用双引号,不然json无法识别
from multiprocessing import Process,Lock
import time,json,random
def search():
    dic=json.load(open('db.txt'))
    print('\033[43m剩余票数%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db.txt'))
    time.sleep(0.1) #模拟读数据的网络延迟
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(0.2) #模拟写数据的网络延迟
        json.dump(dic,open('db.txt','w'))
        print('\033[43m购票成功\033[0m')

def task(lock):
    search()
    get()
if __name__ == '__main__':
    lock=Lock()
    for i in range(100): #模拟并发100个客户端抢票
        p=Process(target=task,args=(lock,))
        p.start()
复制代码
复制代码
#文件db的内容为:{"count":1}
#注意一定要用双引号,不然json无法识别
from multiprocessing import Process,Lock
import time,json,random
def search():
    dic=json.load(open('db.txt'))
    print('\033[43m剩余票数%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db.txt'))
    time.sleep(0.1) #模拟读数据的网络延迟
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(0.2) #模拟写数据的网络延迟
        json.dump(dic,open('db.txt','w'))
        print('\033[43m购票成功\033[0m')

def task(lock):
    search()
    lock.acquire()
    get()
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    for i in range(100): #模拟并发100个客户端抢票
        p=Process(target=task,args=(lock,))
        p.start()
复制代码

 

3、总结——抛‘互斥锁’砖,引出‘管道队列’玉(请看第七部分--队列)

#加锁可以保证多个进程修改同一块数据时,同一时间只能有一个任务可以进行修改,即串行的修改,没错,速度是慢了,但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信,但问题是:
1.效率低(共享数据基于文件,而文件是硬盘上的数据)
2.需要自己加锁处理

 

#因此我们最好找寻一种解决方案能够兼顾——基于消息的IPC通信机制:队列和管道

1、效率高(多个进程共享一块内存的数据)

2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制:队列和管道。


队列和管道都是将数据存放于内存中,
队列又是基于(管道+锁)实现的,可以让我们从复杂的锁问题中解脱出来,
我们应该尽量避免使用共享数据,尽可能使用消息传递和队列,避免处理复杂的同步和锁问题,而且在进程数目增多时,往往可以获得更好的可获展性。

 

 

七 、队列--生产者消费者模型

 进程彼此之间互相隔离,要实现进程间通信(IPC :Inter-Process Communication)

multiprocessing模块支持两种形式:队列和管道,这两种方式都是使用消息传递的

 

 

 1、类 Queue 的介绍与基本使用

(1)创建队列的类(底层就是以管道和锁定的方式实现)

 

Queue([maxsize]):创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递。 

    参数介绍:

maxsize是队列中允许最大项数,省略则无大小限制。    

(2)主要方法介绍:

q.put()

 

  •  

用以插入数据到队列中,put方法还有两个可选参数:blocked和timeout。

如果blocked为True(默认值),并且timeout为正值,该方法会阻塞timeout指定的时间,直到该队列有剩余的空间。如果超时,会抛出Queue.Full异常。

如果blocked为False,但该Queue已满,会立即抛出Queue.Full异常。

q.get()

 

  •  

可以从队列读取并且删除一个元素。同样,get方法有两个可选参数:blocked和timeout。

如果blocked为True(默认值),并且timeout为正值,那么在等待时间内没有取到任何元素,会抛出Queue.Empty异常。

如果blocked为False,有两种情况存在,如果Queue有一个值可用,则立即返回该值,否则,如果队列为空,则立即抛出Queue.Empty异常.
 

q.get_nowait()  同q.get(False)

 

  •  

q.put_nowait()  同q.put(False)

 

  •  

q.empty()

 

  •  

调用此方法时q为空则返回True,该结果不可靠,比如在返回True的过程中,如果队列中又加入了项目。

q.full()

 

  •  

调用此方法时q已满则返回True,该结果不可靠,比如在返回True的过程中,如果队列中的项目被取走。

q.qsize()

 

  •  

返回队列中目前项目的正确数量,结果也不可靠,理由同q.empty()和q.full()一样

 


相关教程