python22-多进程

Posted 2020-11-01 liangjiongyao

一、多进程

为什么用多进程？

由于GIL的存在，python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。

 

多进程为什么消耗大？

之前说是因为切换，但这只是表面原因，根本原因是每开一个进程就要从父进程copy一份

multiprocessing.Process与threading.Thread

multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法相同，也有start(), run(), join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition类 (这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

二、Process类

构造方法：

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

　　group: 线程组，目前还没有实现，库引用中提示必须是None； 
　　target: 要执行的方法； 
　　name: 进程名； 
　　args/kwargs: 要传入方法的参数。

实例方法：

　　is_alive()：返回进程是否在运行。

　　join([timeout])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。

　　start()：进程准备就绪，等待CPU调度

　　run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。

　　terminate()：不管任务是否完成，立即停止工作进程

属性：

　　daemon：和线程的setDeamon功能一样

　　name：进程名字。

　　pid：进程号。

 

三、多进程调用

 一般调用

from multiprocessing import Process
import time

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

if __name__=="__main__":
    p_list=[]
    for i in range(3):
        p = Process(target=f,args=("ljy",))
        p_list.append(p)
        p.start()

    for p in p_list:
        p.join()

    print("end")

View Code

 

继承式调用

from multiprocessing import Process
import time

class MyProcess(Process):
    def run(self):
        time.sleep(1)
        print("hello",self.name,time.ctime())


if __name__=="__main__":
    p_list=[]

    for i in range(5):
        p = MyProcess()
        p.daemon=True # 打印end就结束了
        p.start()
        p_list.append(p)

    # for p in p_list:
    #     p.join()

    print("end")

View Code

 

待定

from multiprocessing import Process
import time,os

def info(title):
    print("title",title)
    print("parent process:",os.getppid()) #pycharm
    print("provess id:",os.getpid()) #父进程


if __name__=="__main__":
    info("main process line")

    time.sleep(1)
    print("------")

    p=Process(target=info,args=("ljy",))
    p.start() 
    p.join()

#最后的结果是pycharm进程的ID和这个程序的进程ID以及它的子进程ID

View Code

 

四、进程通信

Queue

线程队列用queue，进程队列用Queue，进程的Queue，是从一个进程复制一份数据给另一份进程，暂时了解到这一点就行了。

import multiprocessing,time

def foo(q,n):
    time.sleep(1)
    print("son process",id(q))
    q.put(n)


if __name__=="__main__":
    #p_list=[]
    q=multiprocessing.Queue()
    for i in range(3):
        p=multiprocessing.Process(target=foo,args=(q,i))
        p.start()
    print(q.get())
    print(q.get())
    print(q.get())

View Code

 

Pipe

生成两个对象，并且这两个对象是双向管道，有去有回。

import multiprocessing

def f(conn):
    conn.send([12,{"name":"yuan"},"hello"])
    response=conn.recv()
    print("response",response)
    conn.close()
    # print("q_ID2:",id(conn))

if __name__=="__main__":
    parent_conn,child_conn = multiprocessing.Pipe() #双向通道
    # print("q_ID1:",id(child_conn))

    p = multiprocessing.Process(target=f,args=(child_conn,))
    p.start()

    print(parent_conn.recv())
    parent_conn.send("hello son!")
    p.join()

View Code

这里的 recv和send看起来跟socket里面的recv、send很像，但其实它们是不一样，这一点要注意

 

到目前为止，Queue和Pipe完成的功能都是进程之间的通信(一发一收)，只是实现的方式不一样，它们都并没有实现数据的共享

 

Manager

实现多个进程操作同一份数据，可支持的数据类型有：

list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value ,Array

例子

from multiprocessing import Process, Manager

def f(d, l,n):

    d[n] = ‘1‘    #{0:"1"}
    d[‘2‘] = 2    #{0:"1","2":2}

    l.append(n)    #[0,1,2,3,4,   0,1,2,3,4,5,6,7,8,9]
    #print(l)


if __name__ == ‘__main__‘:

    with Manager() as manager:

        d = manager.dict()#{},得到一个manager封装过的字典

        l = manager.list(range(5))#[0,1,2,3,4]，得到一个manager封装过的列表


        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)

View Code

 

五、进程同步

当时讲线程的同步锁的时候，就是希望当某个线程在处理某个数据的时候，其他线程在外面等待，直到该线程操作完成，其他线程才能进行操作。

那么在多进程的世界里面有没有共用一种资源的情形呢？肯定有，必须有。比如进程1此刻想在屏幕打印一句，进程2也想在此刻打印一句，结果就黏在一行了。虽然进程与进程之间的内存是独立的，但并不意味着它们没有共同的资源。

from multiprocessing import Process

import time

def f(i):
    time.sleep(1)
    print("hello world %s" %i)

if __name__=="__main__":
    for num in range(10):
        Process(target=f,args=(num,)).start()

View Code

有可能看到两行黏在一起打印，多试几次，运气差的有可能看不到。

 

解决办法，对该数据进行加锁，副作用是，又变成串行

from multiprocessing import Process,Lock

import time

def f(l,i):
    l.acquire()
    time.sleep(1)
    print("hello world %s" %i)
    l.release()

if __name__=="__main__":
    lock=Lock()
    for num in range(10):
        Process(target=f,args=(num,lock)).start()

View Code