多进程

多进程

进程

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

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

进程是一个实体

进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。也是线程的容器

每一个进程都有它自己的地址空间，一般情况下，包括文本区域（text region）、数据区域（data region）和堆栈（stack region）。文本区域存储处理器执行的代码；数据区域存储变量和进程执行期间使用的动态分配的内存；堆栈区域存储着活动过程调用的指令和本地变量。

进程和程序的区别

程序：指令的一个有序集合，其本身没有任何意义，是一个静态的概念

进程：进程是程序在处理机上执行的一次执行过程，是一个动态概念

注意：如果一个一个程序运行两次，就是两个进程（前提程序没有做判断）

进程的三种状态

想要多个进程交替运行，需要遵循一些进程调度算法

1.先来先服务调度算法

2.短作业优先调度算法

3.时间片轮转法

4.多级反馈队列

僵尸进程和孤儿进程

僵尸进程（有害）

当一个进程结束了，他的父进程没有wait它（释放某个进程所占用的资源），那么这个进程就会变成一个僵尸进程。但是如果父进程先结束了，他就不会变成僵尸，他会被一个init进程接管，最后由init进程来wait他

但是如果主线程一直开辟子线程，然后不回收结束的子线程，会导致进程号等资源一直被占用（进程号是有限的），最后会导致内存不够，无法开启新的进程等问题。

孤儿进程（无害）

父进程结束，但是他的子进程还在运行，那么这些子进程就被称为孤儿进程。最终孤儿进程也会交由init进程管理，最后孤儿进程结束生命周期的时候，init也会wait()他，所以孤儿线程是无害的。

由于python的GIL锁，多进程才能够实现并行

开辟多进程

父进程启动一个新的Python解释器进程。子进程只会继承那些运行进程对象的 run() 方法所需的资源。特别是父进程中非必须的文件描述符和句柄不会被继承。相对于使用 fork 或者 forkserver，使用这个方法启动进程相当慢

windows开辟进程，子进程会把当前执行文件的代码重新运行一遍，所以要加上main，防止一直开辟新的进程，让程序崩掉

from multiprocessing import Process

def haha():
    print(__name__)

if __name__ == '__main__':
    for i in range(3):
        p=Process(target=haha)
        p.start()
        print(p.pid)

21456
9912
16392
__mp_main__
__mp_main__
__mp_main__

多进程的PID

cmd 查看指定pid tasklist /v /fi "PID eq 9420"

from multiprocessing import Process
import os
import time

class zx(Process):
    def run(self):
        time.sleep(1)
        print(f"{self.name}的PID为{os.getpid()},他的fu进程PID为{os.getppid()}")

if __name__ == '__main__':
    j1=zx()
    j1.start()
    j2=zx()
    j2.start()
    j3=zx()
    j3.start()

zx-3的PID为21236,他的fu进程PID为13116
zx-2的PID为20108,他的fu进程PID为13116
zx-1的PID为4312,他的fu进程PID为13116

他们的父进程都为主进程，因为是主进程创建的

进程join

子进程执行完毕才会执行主函数，但是子进程执行完毕就会wait了

from multiprocessing import Process

def haha():
    print(__name__)

if __name__ == '__main__':
    p=Process(target=haha)
    p.start()
    p.join()
    print("end")

__mp_main__
end

terminate

from multiprocessing import Process

def haha():
    print(__name__)

if __name__ == '__main__':
    p=Process(target=haha)
    p.start()
    print(p.is_alive())
    p.terminate()#向操作系统发一个结束进程的指令
    p.join()
    print(p.is_alive())
    print("end")

守护进程

守护进程会在主进程执行结束后终止

守护进程内无法在开启子进程，否则会抛出异常

主进程的代码执行完毕守护进程直接结束。但是此时主进程可能没有结束

from multiprocessing import Process
import time

def haha():
    while True:
        print(__name__)

if __name__ == '__main__':
    p=Process(target=haha)
    p.daemon=True
    p.start()
    print("end")
    time.sleep(3)

end
__mp_main__
__mp_main__
__mp_main__
__mp_main__

from multiprocessing import Process
import os
import time

class zx(Process):
    def run(self):
        time.sleep(1)
        print(f"{self.name}的PID为{os.getpid()},他的fu进程PID为{os.getppid()}")
        
js=[]
if __name__ == '__main__':
    j1=zx()
    js.append(j1)
    j2=zx()
    js.append(j2)
    j3=zx()
    js.append(j3)
    for i in js:
        i.daemon=True
        i.start()
    print("end")

end

进程同步

虽然进程的资源是相互独立的，但是多进程是并行，可能在同一时间去抢占系统的资源，比如屏幕资源

注意：这个锁是进程模块的锁

from multiprocessing import Process,Lock
import time
def haha(lock):
    lock.acquire()
    for i in range(11):
        time.sleep(0.5)
        print(i)
    lock.release()

if __name__ == '__main__':
    lock=Lock()
    p=Process(target=haha,args=(lock,))
    p1=Process(target=haha,args=(lock,))
    p2=Process(target=haha,args=(lock,))
    p.start()
    p1.start()
    p2.start()

#重复打印三次，但不会打乱，因为进程锁
0
1
2
3
4
5
6
7
8
9
10
...

进程间通讯（IPC）

Queue和pipe只是实现了数据交互，并没实现数据共享，即一个进程去更改另一个进程的数据。

进程队列Queue

队列满了put，或者空了取值会阻塞

import multiprocessing
import time

def zx(q):
    while 1:
        time.sleep(0.5)
        q.put("来啊客官")
        print("来啊客官")

if __name__ == '__main__':
    q = multiprocessing.Queue(3)
    j=multiprocessing.Process(target=zx,args=(q,))
    j.start()

来啊客官
来啊客官
来啊客官

队列属性

当队列存满put，或者为空get直接抛出异常，不进行阻塞（这个时候设置timeout属性是无效的）

q=Queue(3)
q.get(block=False)
q.put(block=False)

当队列存满put，或者为空get,进行阻塞，如果4秒内队列还是满或者为空抛出异常（block默认为Ture，时间为无限）

q=Queue(3)
q.get(block=True,timeout=4)
q.put(block=True,timeout=4)

JoinableQueue

get一次相当于+1，task_done一次相当于-1，只有为0才可以join通过

from multiprocessing import Process,Queue,JoinableQueue
import time

def zx_put(q):
    for i in range(5):
        time.sleep(1)
        print("放入西瓜")
        q.put(f"西瓜{i}")

def zx_get(q):
    while 1:
        time.sleep(1)
        print("取出"+q.get())
        #计数-1
        q.task_done()

if __name__ == '__main__':
    q=JoinableQueue()
    p1=Process(target=zx_put,args=(q,))
    p2=Process(target=zx_get,args=(q,))
    p1.start()
    time.sleep(5)
    p2.start()

    #堵塞一直等到西瓜没有
    q.join()
    print("西瓜没了")
    p2.terminate()

放入西瓜
放入西瓜
放入西瓜
放入西瓜
放入西瓜
取出西瓜0
取出西瓜1
取出西瓜2
取出西瓜3
取出西瓜4
西瓜没了

管道

from multiprocessing import Process, Pipe

def f(conn):
    conn.send("爸爸你好！")
    response=conn.recv()
    print(response)
    conn.close()
    
if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, 'hello']"
    parent_conn.send("儿子你好!")
    p.join()

爸爸你好！
儿子你好!

Managers

from multiprocessing import Process, Manager

def f(d, l,n):
    d[n] = '1'
    d['2'] = 2
    d[0.25] = None
    l.append(n)

if __name__ == '__main__':

    with Manager() as manager:
        d = manager.dict()
        l = manager.list(range(5))
        p_list = []

        for i in range(10):
            p = Process(target=f, args=(d,l,i))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()

        print(d)
        print(l)

{6: '1', '2': 2, 0.25: None, 2: '1', 7: '1', 1: '1', 4: '1', 0: '1', 3: '1', 8: '1', 5: '1', 9: '1'}
[0, 1, 2, 3, 4, 6, 2, 7, 1, 4, 0, 3, 8, 5, 9]

进程池

• apply 同步 相当于串行

• apply_async 异步

  可以加回调函数，注意，回调函数是运行在主进程的

  可以返回子进程的run函数return

import multiprocessing
import time

def zx(i):
    time.sleep(0.5)
    print(i)
if __name__ == '__main__':
    #默认进程池的数量为你电脑的cpu数量
    pool=multiprocessing.Pool()

    for i in range(40):
        pool.apply_async(zx,args=(i,))
    pool.close()
    pool.join()
    print("搬砖结束")

0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
搬砖结束