python -- 多进程

Posted 2020-09-22

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

Python提供了非常好用的多进程包multiprocessing，只需要定义一个函数，Python会完成其他所有事情。

1.Process

创建进程的类：Process([group [, target [, name [, args [, kwargs]]]]])，target表示调用对象，args表示调用对象的位置参数元组。kwargs表示调用对象的字典。name为别名。group实质上不使用。
方法：is_alive()、join([timeout])、run()、start()、terminate()。其中，Process以start()启动某个进程。

属性：authkey、daemon（要通过start()设置）、exitcode(进程在运行时为None、如果为–N，表示被信号N结束）、name、pid。其中daemon是父进程终止后自动终止，且自己不能产生新进程，必须在start()之前设置。

 

　　(1)创建多进程

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 
 4 from multiprocessing import Process
 5 import time
 6 
 7 
 8 def f(name):
 9     time.sleep(1)
10     print("hello ", name)
11 
12 if __name__ == ‘__main__‘:
13     p1 = Process(target=f, args=[‘Bob1‘, ])     #target是创建的进程要执行的函数，args是传递给该函数的参数
14     p2 = Process(target=f, args=[‘Bob2‘, ])   #这里进程p*的父进程是当前脚本，脚本的父进程是执行环境(eg：pycharm)
15     p1.start()
16     p2.start()
17     p1.join()
18     p2.join()
19     time.sleep(1)
20     print("----main process over----")

　　执行结果

hello  Bob2
hello  Bob1
----main process over----

 

　　(2)进程同步

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 
 4 ‘‘‘
 5 进程同步是说的进程一起执行
 6 ‘‘‘
 7 from multiprocessing import Process,Lock
 8 
 9 def f(lock,i):
10     lock.acquire()
11     try:
12         print(‘hello world‘,i)
13     finally:
14         lock.release()
15 
16 if __name__ == ‘__main__‘:
17     lock = Lock()
18 
19     for num in range(10):
20         Process(target=f,args=(lock,num)).start()

　　执行结果

hello world 0
hello world 3
hello world 2
hello world 6
hello world 5
hello world 1
hello world 4
hello world 7
hello world 9
hello world 8

 

　　(3)同一父进程下的子进程间通信

#!/usr/bin/env python
# -*- coding: utf-8 -*-

‘‘‘
进程之间是不能直接通信的，需要第三方作为通信媒介
‘‘‘

‘‘‘以下是Queue实现数据传递代码‘‘‘
from multiprocessing import  Process,Queue

def f1(q):
    q.put([22,‘root‘,‘hello‘])

def f2(q):
    q.put([23,‘admin‘,‘hello‘])


‘‘‘以下是Pipes实现数据传递代码‘‘‘
from multiprocessing import  Process,Pipe

def f3(conn):
    conn.send([23,‘admin1‘,‘hello‘])

def f4(conn):
    conn.send([24, ‘root‘, ‘hello‘])


‘‘‘以下是Manager实现数据共享代码‘‘‘
from multiprocessing import Process,Manager

def f(d,l,n):
    d[n] = n
    d[‘name‘] = ‘root‘
    l.append(n)
    print(l)

#以下是主程序
if __name__ == ‘__main__‘:
    print(
        ‘‘‘\033[42;1m 利用Queue实现进程间的通信 \033[0m ‘‘‘
    )
    ‘‘‘
    这里的进程间通信是在同一父进程下不同子进程间的通信，因为队列q是在父进程实例化并
    传递到每个子进程中。所以Queue是不能用于独立的进程间通信的
    ‘‘‘
    q = Queue()    #生成一个第三方队列，用来存放子进程put的值
                # 这里的Queue是在queue.Queue的基础上，封装了能多进程访问
    pq1 = Process(target=f1,args=(q,))
    pq2 = Process(target=f2,args=(q,))  #要把队列对象q当参数传递给各个子进程
    pq1.start()
    pq2.start()
    print(‘from parent 1:‘,q.get())
    print(‘from parent 2:‘,q.get())

    print(
        ‘‘‘\n\033[43;1m 利用Pipes实现进程间的通信 \033[0m ‘‘‘
    )
    parent_conn,child_conn = Pipe()
    pp1 = Process(target=f3,args=(child_conn,))
    pp2 = Process(target=f4,args=(child_conn,))
    pp1.start()
    pp2.start()
    print(‘parent data from pp1:‘, parent_conn.recv())
    print(‘parent data from pp2:‘, parent_conn.recv())
    parent_conn.close() #这个在子进程或父进程关闭都可以

    print(
        ‘‘‘\n\033[43;1m 利用Manager实现进程间的数据共享 \033[0m ‘‘‘
    )
    ‘‘‘
    这里Manager()支持：list、dict、Namespace、Lock、Rlock、Semaphore、
     BoundedSemaphore、Condition、Event、Barrier、Queue、Value、Array
     其中常用的就list、dict。
     注意：其中的list、dict都要通过Manager() 实现
    ‘‘‘
    #manager = Manager()  #注意这样后面需要关闭，用with...as...语句更方便
    with Manager() as manager:
        d = manager.dict()
        # l = manager.list()  #这样生成空列表
        l = manager.list(range(5))

        process_list = []
        for i in range(10):
            p = Process(target=f,args=(d,l,i))
            p.start()
            process_list.append(p)

        for res in process_list:
            res.join()

        print("执行结束l：",l)
        print("执行结束d：",d)

 

　　(4)进程池

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 
 4 ‘‘‘
 5 进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有
 6 可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。
 7 
 8 进程池方法：
 9     apply: 同步
10     apply_async: 异步
11 ‘‘‘
12 from multiprocessing import Process,Pool,freeze_support
13 import time
14 
15 def Foo(i):
16     time.sleep(1)
17     return i,i+100
18 
19 def Bar(arg):
20     # print(‘arg:‘,arg)
21     data.append(arg)
22     print(‘--> exec done [%s]: %s‘ % (arg[0],arg[1]))
23 
24 if __name__ == ‘__main__‘:
25     freeze_support()   #要导入，在win下必须有它，不然会报错。linux下不用
26 
27     pool = Pool(3)  #生成一个有5个进程序列的进程池实例
28     data = []
29     for i in range(10):
30         # func为进程每次执行的函数，args是参数，callback是执行完func后执行的函数，
31         # 并且callback会自动接收func的返回值作为函数参数。Foo是子进程执行的函数，callback是在
32         # 父进程执行的函数，所以这里说明是子进程把执行结果返回给父进程里面的函数
33         pool.apply_async(func=Foo,args=(i,),callback=Bar)
34         #pool.apply(func=Foo,args=(i,))   #进程同步，没啥用
35 
36     print("end")
37     pool.close()
38     pool.join()  #这里表示进程池中进程执行完毕后再关闭，若注释，那么程序直接关闭
39         #注意：这里是先close，再join
40     print(‘全局变量data：‘,data)

　　执行结果

end
--> exec done [0]: 100
--> exec done [1]: 101
--> exec done [2]: 102
--> exec done [3]: 103
--> exec done [4]: 104
--> exec done [5]: 105
--> exec done [7]: 107
--> exec done [6]: 106
--> exec done [8]: 108
--> exec done [9]: 109
全局变量data： [(0, 100), (1, 101), (2, 102), (3, 103), (4, 104), (5, 105), (7, 107), (6, 106), (8, 108), (9, 109)]