Python 多进程 -- 2019-08-16 17:44:03

Posted 2022-07-18 bbb001

原文: http://blog.gqylpy.com/gqy/228

"

一、模块介绍

1. multiprocess模快
   仔细说来，multiprocess不是一个模块，而是python中的一个操作、管理进程的包，之所以叫multi是取自multiple的多功能的意思，这个包中几乎包含了和进程有关的所有子模块。

2. multiprocess.Process模块
   Process能够帮助我们创建子进程，以及对子进程的一些控制.

• 参数：def __init__(self, group=None, target=None, name=None, args=(), kwargs=):

group=None：该参数未使用，值始终为None

target：指定要调用的函数，即子进程要执行的任务

args：指定被调用对象的位置参数，以元组的形式传入，必须有逗号，如:args=(‘a‘,）；此参数在源码中定义的只接收元组，但由于tuple数据类型底层原理的原因，此参数是可以传入一切可迭代对象，如：列表、字典、字符串、生成器表达式等

kwargs：指定被调用对象关键字参数，以字典的形式传入，关键字作为键，参数作为值

name：指定子进程的名称，默认名称为：Process-1、Process-2、...

• 方法

obj.start()：启动子进程，并调用run()方法

obj.run()：启动子进程时自动被调用，正是它去调用target参数指定的函数，自定义类中必须实现此方法

obj.terminate()：强制终止子进程，不会进行任何清理操作，事实上是让操作系统去终止子进程；而操作系统终止进程是有一个过程的，所以子进程不会立即被终止；使用此方法需谨慎以下两种情况：1. 如果子进程中创建了子子进程，强制终止子进程后，子子进程将变成僵尸进程. 2. 如果子进程保存了一个锁，强制终止后，内存将不会被释放，进而导致死锁

obj.is_alive()：判断子进程是否存活，若存活返回True，否则False

obj.join(timeout=None)：阻塞父进程，等待子进程终止后再继续执行父进程，timeout指定等待超时时间；此方法只能用于使用start方法启动的子进程，对run方法启动的子进程无效

• 属性

obj.daemon：默认值为False，如果设为True，子进程将成为守护进程，此属性必须写在start()的前面；守护进程无法创建子进程，且会随着父进程的终止而终止

obj.name：返回子进程的名称

obj.pid：返回子进程的pid

obj.exitcode：子进程在运行时为None，如果为-N，则表示被信号N结束了

obj.authkey：子进程的身份验证键，默认由os.urandom()随机成生成的32bytes；是为涉及网络连接的底层进程间通讯提供安全性，这类连接只有在具有相同身份验证键时才能成功

---

二、使用Process创建进程

• windows系统使用Process模块需注意
  由于windows操作系统中没有fork（Linux操作系统中创建进程的机制），因此启动子进程时采用的是导入启动子进程的文件的方式来完成进程间的数据共享。而导入文件即等于执行文件，因此如果将process()直接写在文件中将会无限递归子进程而报错，所以必须把创建子进程的部分使用if __name__ == ‘__main__‘:判断保护起来，以防止递归.

1. 基本操作

# 创建单个子进程及对子进程简单的控制
 
from multiprocessing import Process
from time import sleep
import os
 
 
def func(par):
    print("%s, PID: %s" %(par, os.getpid()))
    print("子进程终止后才继续执行父进程")
    sleep(5)
 
 
if __name__ == ‘__main__‘:
 
    p = Process(target=func, args=("子进程",))     # 实例化一个子进程对象p
    # p.run()   # run方法启动的子进程免疫join方法
    p.start()   # 子进程进入就绪状态，等待操作系统调度
 
    print(p.is_alive())     # True: 子进程存活
    p.join(1)    # 阻塞父进程，等待子进程终止，等待超时时间1s
    print(p.is_alive())     # False: 子进程终止
 
    print("父进程, PID: %s" % os.getpid())

# 创建多个子进程
 
from multiprocessing import Process
from time import sleep
 
 
def func(n):
    print("子进程", n)
    sleep(0.1)
 
 
if __name__ == ‘__main__‘:
    p_lst = []
    for i in range(5):  # 通过for循环启动多个子进程
        p = Process(target=func, args=(i,))
        p.start()
        p_lst.append(p)
    [p.join() for p in p_lst]   # 阻塞父进程，等待所有子进程终止
 
    print("父进程")

2. 自定义多进程类

# 自定义多进程类
 
from multiprocessing import Process
from time import sleep
 
 
class MyProcess(Process):
 
    def __init__(self, target, name, sex):
        self.sex = sex
        super(MyProcess, self).__init__(target=target, name=name)
        # self.name = name
        # 父类Process初始化时会自动定义name属性，所以定义本类的name属性要写到super语句后面，
        # 否则会被父类初始化时覆盖掉(MyProcess-子进程序号),或者给父类传参__init__(name=name)
 
    def run(self):  # 必写方法，用于调用函数
        print("子进程:", self.name)
        super(MyProcess, self).run()
 
 
func = lambda :print("执行了函数")
 
# 如果是Windows系统，此处要写一条__main__判断语句
p = MyProcess(func, ‘zyk‘, ‘boy‘)
p.start()
# p.run()   # start方法会自动调用此方法

3.进程之间的数据共享与隔离

# 进程之间的数据共享与隔离
 
from multiprocessing import Process, Value
 
 
def func():
    global n    # ！
    n = 2  # 数据隔离：不会影响父进程中的变量n
    print("子进程n: ", n)  # 2
 
    s.value = 4     # 数据共享：会影响父进程中的s
    print("子进程s: ", s.value)
 
 
if __name__ == ‘__main__‘:
    n = 1
    s = Value(‘i‘, 3)
 
    p = Process(target=func)
    p.start()
    p.join()
 
    print("父进程n: ", n)  # 1
    print("父进程s: ", s.value)    # 4

4.守护进程

# 守护进程
 
from multiprocessing import Process
from os import getpid
 
 
class MyProcess(Process):
 
    def __init__(self, name):
        super(MyProcess, self).__init__()
        self.name = name
 
    def run(self):
        print(self.name, getpid())
        # 父进程是先进入就绪状态的，所以父进程一般会先终止（除非父进程还有很长的逻辑要走）
        # 子进程可能还未打印信息便随父进程终止而终止了，
 
p = MyProcess(‘zyk‘)
p.daemon = True     # 确认为守护进程，随父进程终止而终止
p.start()
# p.join()    # 阻塞父进程，方可正常执行完子进程，而显示打印信息
 
print("父进程")

5.进阶：socket消息加并发实例

# Server
 
from multiprocessing import Process
from socket import *
 
 
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(1472)
            if not msg:break
            conn.send(msg.upper())
        except Exception:
            break
 
 
if __name__ == ‘__main__‘:
    while 1:
        conn, client_addr = server.accept()
        p = Process(target=talk, args=(conn, client_addr))
        p.start()
        print(p.name)

# Client
 
import socket
 
 
client = socket.socket()
client.connect_ex((‘127.0.0.1‘, 8080))
 
 
while 1:
    msg = input(‘>>>‘).strip()
    if not msg:continue
 
    client.send(msg.encode(‘utf-8‘))
    msg = client.recv(1472)
    print(msg.decode(‘utf-8‘))

完结

"

原文: http://blog.gqylpy.com/gqy/228