欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

python基础(32):进程(二)

程序员文章站 2022-05-09 15:03:28
1. multiprocess模块 仔细说来,multiprocess不是一个模块而是python中一个操作、管理进程的包。 之所以叫multi是取自multiple的多功能的意思,在这个包中几乎包含了和进程有关的所有子模块。由于提供的子模块非常多,为了方便大家归类记忆,我将这部分大致分为四个部分: ......

1. multiprocess模块

仔细说来,multiprocess不是一个模块而是python中一个操作、管理进程的包。 之所以叫multi是取自multiple的多功能的意思,在这个包中几乎包含了和进程有关的所有子模块。由于提供的子模块非常多,为了方便大家归类记忆,我将这部分大致分为四个部分:创建进程部分,进程同步部分,进程池部分,进程之间数据共享。

1.1 multiprocess.process模块

1.1.1 process模块介绍

process模块是一个创建进程的模块,借助这个模块,就可以完成进程的创建。

process([group [, target [, name [, args [, kwargs]]]]]),由该类实例化得到的对象,表示一个子进程中的任务(尚未启动)

强调:
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数,是一个元组形式,必须有逗号

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

方法介绍:

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

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

p.terminate():强制终止进程p,不会进行任何清理操作,如果p创建了子进程,该子进程就成了僵尸进程,使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放,进而导致死锁

p.is_alive():如果p仍然运行,返回true

p.join([timeout]):主线程等待p终止(强调:是主线程处于等的状态,而p是处于运行的状态)。timeout是可选的超时时间,需要强调的是,p.join只能join住start开启的进程,而不能join住run开启的进程

属性介绍:

p.daemon:默认值为false,如果设为true,代表p为后台运行的守护进程,当p的父进程终止时,p也随之终止,并且设定为true后,p不能创建自己的新进程,必须在p.start()之前设置

p.name:进程的名称

p.pid:进程的pid

p.exitcode:进程在运行时为none、如果为–n,表示被信号n结束(了解即可)

p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性,这类连接只有在具有相同的身份验证键时才能成功(了解即可)

在windows操作系统中由于没有fork(linux操作系统中创建进程的机制),在创建子进程的时候会自动 import 启动它的这个文件,而在 import 的时候又执行了整个文件。因此如果将process()直接写在文件中就会无限递归创建子进程报错。所以必须把创建子进程的部分使用if __name__ ==‘__main__’ 判断保护起来,import 的时候 ,就不会递归运行了。

1.1.2 使用process模块创建进程

在一个python进程中开启子进程,start方法和并发效果。

import time
from multiprocessing import process

def f(name):
    print('hello', name)
    print('我是子进程')

if __name__ == '__main__':
    p = process(target=f, args=('bob',))
    p.start()
    time.sleep(1)
    print('执行主进程的内容了')

join方法:

import time
from multiprocessing import process

def f(name):
    print('hello', name)
    time.sleep(1)
    print('我是子进程')


if __name__ == '__main__':
    p = process(target=f, args=('bob',))
    p.start()
    #p.join()
    print('我是父进程')

查看主进程和子进程的进程号:

import os
from multiprocessing import process

def f(x):
    print('子进程id :',os.getpid(),'父进程id :',os.getppid())
    return x*x

if __name__ == '__main__':
    print('主进程id :', os.getpid())
    p_lst = []
    for i in range(5):
        p = process(target=f, args=(i,))
        p.start()

进阶,多个进程同时运行(注意,子进程的执行顺序不是根据启动顺序决定的)

多个进程同时运行:

import time
from multiprocessing import process


def f(name):
    print('hello', name)
    time.sleep(1)


if __name__ == '__main__':
    p_lst = []
    for i in range(5):
        p = process(target=f, args=('bob',))
        p.start()
        p_lst.append(p)

多个进程同时运行,再谈join方法(1):

import time
from multiprocessing import process


def f(name):
    print('hello', name)
    time.sleep(1)


if __name__ == '__main__':
    p_lst = []
    for i in range(5):
        p = process(target=f, args=('bob',))
        p.start()
        p_lst.append(p)
        p.join()
    # [p.join() for p in p_lst]
    print('父进程在执行')

多个进程同时运行,再谈join方法(2):

import time
from multiprocessing import process

def f(name):
    print('hello', name)
    time.sleep(1)

if __name__ == '__main__':
    p_lst = []
    for i in range(5):
        p = process(target=f, args=('bob',))
        p.start()
        p_lst.append(p)
    # [p.join() for p in p_lst]
    print('父进程在执行')

除了上面这些开启进程的方法,还有一种以继承process类的形式开启进程的方式。

import os
from multiprocessing import process

class myprocess(process):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print(os.getpid())
        print('%s 正在和女主播聊天' %self.name)

p1=myprocess('wupeiqi')
p2=myprocess('yuanhao')
p3=myprocess('nezha')

p1.start() #start会自动调用run
p2.start()
# p2.run()
p3.start()

p1.join()
p2.join()
p3.join()

print('主线程')

进程之间的数据隔离问题:

from multiprocessing import process

def work():
    global n
    n=0
    print('子进程内: ',n)


if __name__ == '__main__':
    n = 100
    p=process(target=work)
    p.start()
    print('主进程内: ',n)

1.1.3 守护进程

会随着主进程的结束而结束。

主进程创建守护进程:

1.守护进程会在主进程代码执行结束后就终止

2.守护进程内无法再开启子进程,否则抛出异常:assertionerror: daemonic processes are not allowed to have children

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

守护进程的启动:

import os
import time
from multiprocessing import process

class myprocess(process):
    def __init__(self,person):
        super().__init__()
        self.person = person
    def run(self):
        print(os.getpid(),self.name)
        print('%s正在和女主播聊天' %self.person)

p=myprocess('哪吒')
p.daemon=true #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
p.start()
time.sleep(10) # 在sleep时查看进程id对应的进程ps -ef|grep id
print('主')

主进程代码执行结束守护进程立即结束:

from multiprocessing import process

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()
time.sleep(0.1)
print("main-------")#打印该行则主进程代码结束,则守护进程p1应该被终止.#可能会有p1任务执行的打印信息123,因为主进程打印main----时,p1也执行了,但是随即被终止.

1.1.4 socket聊天并发实例

server:

from socket import *
from multiprocessing import process

server=socket(af_inet,sock_stream)
server.setsockopt(sol_socket,so_reuseaddr,1)
server.bind(('127.0.0.1',8080))
server.listen(5)

def talk(conn,client_addr):
    while true:
        try:
            msg=conn.recv(1024)
            if not msg:break
            conn.send(msg.upper())
        except exception:
            break

if __name__ == '__main__': #windows下start进程一定要写到这下面
    while true:
        conn,client_addr=server.accept()
        p=process(target=talk,args=(conn,client_addr))
        p.start()

client:

from socket import *

client=socket(af_inet,sock_stream)
client.connect(('127.0.0.1',8080))


while true:
    msg=input('>>: ').strip()
    if not msg:continue

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

1.1.4 多进程中的其他方法

进程对象的其他方法:terminate,is_alive:

from multiprocessing import process
import time
import random

class myprocess(process):
    def __init__(self,person):
        self.name=person
        super().__init__()

    def run(self):
        print('%s正在和网红脸聊天' %self.name)
        time.sleep(random.randrange(1,5))
        print('%s还在和网红脸聊天' %self.name)

p1=myprocess('哪吒')
p1.start()

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

print('开始')
print(p1.is_alive()) #结果为false

进程对象的其他属性:pid和name:

class myprocess(process):
    def __init__(self,person):
        self.name=person   # name属性是process中的属性,标示进程的名字
        super().__init__() # 执行父类的初始化方法会覆盖name属性
        #self.name = person # 在这里设置就可以修改进程名字了
        #self.person = person #如果不想覆盖进程名,就修改属性名称就可以了
    def run(self):
        print('%s正在和网红脸聊天' %self.name)
        # print('%s正在和网红脸聊天' %self.person)
        time.sleep(random.randrange(1,5))
        print('%s正在和网红脸聊天' %self.name)
        # print('%s正在和网红脸聊天' %self.person)

p1=myprocess('哪吒')
p1.start()
print(p1.pid)    #可以查看子进程的进程id

1.2 进程同步(multiprocess.lock)

1.2.1 锁--multiprocess.lock

通过刚刚的学习,我们千方百计实现了程序的异步,让多个任务可以同时在几个进程中并发处理,他们之间的运行没有顺序,一旦开启也不受我们控制。尽管并发编程让我们能更加充分的利用io资源,但是也给我们带来了新的问题。

当多个进程使用同一份数据资源的时候,就会引发数据安全或顺序混乱问题。

多进程抢占输出资源:

import os
import time
import random
from multiprocessing import process

def work(n):
    print('%s: %s is running' %(n,os.getpid()))
    time.sleep(random.random())
    print('%s:%s is done' %(n,os.getpid()))

if __name__ == '__main__':
    for i in range(3):
        p=process(target=work,args=(i,))
        p.start()

使用锁维护执行顺序:

# 由并发变成了串行,牺牲了运行效率,但避免了竞争
import os
import time
import random
from multiprocessing import process,lock

def work(lock,n):
    lock.acquire()
    print('%s: %s is running' % (n, os.getpid()))
    time.sleep(random.random())
    print('%s: %s is done' % (n, os.getpid()))
    lock.release()
if __name__ == '__main__':
    lock=lock()
    for i in range(3):
        p=process(target=work,args=(lock,i))
        p.start()

上面这种情况虽然使用加锁的形式实现了顺序的执行,但是程序又重新变成串行了,这样确实会浪费了时间,却保证了数据的安全。

接下来,我们以模拟抢票为例,来看看数据安全的重要性。

多进程同时抢购余票:

#文件db的内容为:{"count":1}
#注意一定要用双引号,不然json无法识别
#并发运行,效率高,但竞争写同一文件,数据写入错乱
from multiprocessing import process,lock
import time,json,random
def search():
    dic=json.load(open('db'))
    print('\033[43m剩余票数%s\033[0m' %dic['count'])

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

def task():
    search()
    get()

if __name__ == '__main__':
    for i in range(100): #模拟并发100个客户端抢票
        p=process(target=task)
        p.start()

使用锁来保证数据安全:

#文件db的内容为:{"count":5}
#注意一定要用双引号,不然json无法识别
#并发运行,效率高,但竞争写同一文件,数据写入错乱
from multiprocessing import process,lock
import time,json,random
def search():
    dic=json.load(open('db'))
    print('\033[43m剩余票数%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db'))
    time.sleep(random.random()) #模拟读数据的网络延迟
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(random.random()) #模拟写数据的网络延迟
        json.dump(dic,open('db','w'))
        print('\033[32m购票成功\033[0m')
    else:
        print('\033[31m购票失败\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()

#加锁可以保证多个进程修改同一块数据时,同一时间只能有一个任务可以进行修改,即串行的修改,没错,速度是慢了,但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信,但问题是:

1.效率低(共享数据基于文件,而文件是硬盘上的数据)
2.需要自己加锁处理

因此我们最好找寻一种解决方案能够兼顾:

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

2.帮我们处理好锁问题

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

1.3 队列(multiprocess.queue)

1.3.1 队列

(1) 概念介绍

创建共享的进程队列,queue是多进程安全的队列,可以使用queue实现多进程之间的数据传递。 

queue([maxsize])

创建共享的进程队列。

参数 :maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。

底层队列使用管道和锁定实现。

方法介绍:

queue([maxsize])

创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数,则无大小限制。底层队列使用管道和锁定实现。另外,还需要运行支持线程以便队列中的数据传输到底层管道中。 queue的实例q具有以下方法:

q.get( [ block [ ,timeout ] ] )

返回q中的一个项目。如果q为空,此方法将阻塞,直到队列中有项目可用为止。block用于控制阻塞行为,默认为true. 如果设置为false,将引发queue.empty异常(定义在queue模块中)。timeout是可选超时时间,用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用,将引发queue.empty异常。

q.get_nowait( )

同q.get(false)方法。

q.put(item [, block [,timeout ] ] )

将item放入队列。如果队列已满,此方法将阻塞至有空间可用为止。block控制阻塞行为,默认为true。如果设置为false,将引发queue.empty异常(定义在queue库模块中)。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发queue.full异常。

q.qsize()

返回队列中目前项目的正确数量。此函数的结果并不可靠,因为在返回结果和在稍后程序中使用结果之间,队列中可能添加或删除了项目。在某些系统上,此方法可能引发notimplementederror异常。

q.empty()

如果调用此方法时 q为空,返回true。如果其他进程或线程正在往队列中添加项目,结果是不可靠的。也就是说,在返回和使用结果之间,队列中可能已经加入新的项目。

q.full()

如果q已满,返回为true. 由于线程的存在,结果也可能是不可靠的(参考q.empty()方法)。

其他方法(了解):

q.close()

关闭队列,防止队列中加入更多数据。调用此方法时,后台线程将继续写入那些已入队列但尚未写入的数据,但将在此方法完成时马上关闭。如果q被垃圾收集,将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如,如果某个使用者正被阻塞在get()操作上,关闭生产者中的队列不会导致get()方法返回错误。

q.cancel_join_thread()

不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。

q.join_thread()

连接队列的后台线程。此方法用于在调用q.close()方法后,等待所有队列项被消耗。默认情况下,此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。

(2) 代码实例

'''
multiprocessing模块支持进程间通信的两种主要形式:管道和队列
都是基于消息传递实现的,但是队列接口
'''

from multiprocessing import queue
q=queue(3)

#put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(3)
q.put(3)
# q.put(3)   # 如果队列已经满了,程序就会停在这里,等待数据被别人取走,再将数据放入队列。
           # 如果队列中的数据一直不被取走,程序就会永远停在这里。
try:
    q.put_nowait(3) # 可以使用put_nowait,如果队列满了不会阻塞,但是会因为队列满了而报错。
except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去,但是会丢掉这个消息。
    print('队列已经满了')

# 因此,我们再放入数据之前,可以先看一下队列的状态,如果已经满了,就不继续put了。
print(q.full()) #满了

print(q.get())
print(q.get())
print(q.get())
# print(q.get()) # 同put方法一样,如果队列已经空了,那么继续取就会出现阻塞。
try:
    q.get_nowait(3) # 可以使用get_nowait,如果队列满了不会阻塞,但是会因为没取到值而报错。
except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。
    print('队列已经空了')

print(q.empty()) #空了

上面这个例子还没有加入进程通信,只是先来看看队列为我们提供的方法,以及这些方法的使用和现象。

子进程发送数据给父进程:

import time
from multiprocessing import process, queue

def f(q):
    q.put([time.asctime(), 'from eva', 'hello'])  #调用主函数中p进程传递过来的进程参数 put函数为向队列中添加一条数据。

if __name__ == '__main__':
    q = queue() #创建一个queue对象
    p = process(target=f, args=(q,)) #创建一个进程
    p.start()
    print(q.get())
    p.join()

上面是一个queue的简单应用,使用队列q对象调用get函数来取得队列中最先进入的数据。 接下来看一个稍微复杂一些的例子:

批量生产数据放入队列再批量获取结果:

import os
import time
import multiprocessing

# 向queue中输入数据的函数
def inputq(queue):
    info = str(os.getpid()) + '(put):' + str(time.asctime())
    queue.put(info)

# 向queue中输出数据的函数
def outputq(queue):
    info = queue.get()
    print ('%s%s\033[32m%s\033[0m'%(str(os.getpid()), '(get):',info))

# main
if __name__ == '__main__':
    multiprocessing.freeze_support()
    record1 = []   # store input processes
    record2 = []   # store output processes
    queue = multiprocessing.queue(3)

    # 输入进程
    for i in range(10):
        process = multiprocessing.process(target=inputq,args=(queue,))
        process.start()
        record1.append(process)

    # 输出进程
    for i in range(10):
        process = multiprocessing.process(target=outputq,args=(queue,))
        process.start()
        record2.append(process)

    for p in record1:
        p.join()

    for p in record2:
        p.join()

1.4 进程之间的数据共享

展望未来,基于消息传递的并发编程是大势所趋

即便是使用线程,推荐做法也是将程序设计为大量独立的线程集合,通过消息队列交换数据。

这样极大地减少了对使用锁定和其他同步手段的需求,还可以扩展到分布式系统中。

但进程间应该尽量避免通信,即便需要通信,也应该选择进程安全的工具来避免加锁带来的问题。

以后我们会尝试使用数据库来解决现在进程之间的数据共享问题。

manager模块介绍:

进程间数据是独立的,可以借助于队列或管道实现通信,二者都是基于消息传递的
虽然进程间数据独立,但可以通过manager实现数据共享,事实上manager的功能远不止于此

a manager object returned by manager() controls a server process which holds python objects and allows other processes to manipulate them using proxies.

a manager returned by manager() will support types list, dict, namespace, lock, rlock, semaphore, boundedsemaphore, condition, event, barrier, queue, value and array.

manager例子:

from multiprocessing import manager,process,lock
def work(d,lock):
    with lock: #不加锁而操作共享的数据,肯定会出现数据错乱
        d['count']-=1

if __name__ == '__main__':
    lock=lock()
    with manager() as m:
        dic=m.dict({'count':100})
        p_l=[]
        for i in range(100):
            p=process(target=work,args=(dic,lock))
            p_l.append(p)
            p.start()
        for p in p_l:
            p.join()
        print(dic)

1.5 进程池和multiprocess.pool模块

1.5.1 进程池

为什么要有进程池,进程池的概念?

在程序实际处理问题过程中,忙时会有成千上万的任务需要被执行,闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候,我们就需要去创建成千上万个进程么?首先,创建进程需要消耗时间,销毁进程也需要消耗时间。第二即便开启了成千上万的进程,操作系统也不能让他们同时执行,这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。那么我们要怎么做呢?

在这里,要给大家介绍一个进程池的概念,定义一个池子,在里面放上固定数量的进程,有需求来了,就拿一个池中的进程来处理任务,等到处理完毕,进程并不关闭,而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行,池中的进程数量不够,任务就要等待之前的进程执行任务完毕归来,拿到空闲进程才能继续执行。也就是说,池中进程的数量是固定的,那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度,还节省了开闭进程的时间,也一定程度上能够实现并发效果。

1.5.2 multiprocess.pool模块

(1) 概念介绍

pool([numprocess  [,initializer [, initargs]]]):创建进程池

参数介绍:

numprocess:要创建的进程数,如果省略,将默认使用cpu_count()的值
initializer:是每个工作进程启动时要执行的可调用对象,默认为none
initargs:是要传给initializer的参数组

主要方法:

p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
'''需要强调的是:此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数,必须从不同线程调用p.apply()函数或者使用p.apply_async()'''

p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
'''此方法的结果是asyncresult类的实例,callback是可调用对象,接收输入参数。当func的结果变为可用时,将理解传递给callback。callback禁止执行任何阻塞操作,否则将接收其他异步操作中的结果。'''
   
p.close():关闭进程池,防止进一步操作。如果所有操作持续挂起,它们将在工作进程终止前完成

p.jion():等待所有工作进程退出。此方法只能在close()或teminate()之后调用

其他方法(了解):

方法apply_async()和map_async()的返回值是asyncresul的实例obj。实例具有以下方法
obj.get():返回结果,如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达,将引发一场。如果远程操作中引发了异常,它将在调用此方法时再次被引发。
obj.ready():如果调用完成,返回true
obj.successful():如果调用完成且没有引发异常,返回true,如果在结果就绪之前调用此方法,引发异常
obj.wait([timeout]):等待结果变为可用。
obj.terminate():立即终止所有工作进程,同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收,将自动调用此函数