Python 进程、进程池

Python 进程

1、创建进程

# 创建进程

from multiprocessing import Process

def foo(i):
    print('say hi', i)

if __name__ == '__main__':
    for i in range(3):
        p = Process(target=foo, args=(i, ))
        p.start()
    print('main proccess stop')
    
-----------输出结果-----------
    main proccess stop
    say hi 0
    say hi 2
    say hi 1

 注意：由于进程之间的数据需要各自持有一份，所以创建进程需要的非常大的开销。

2、进程数据共享

进程各自持有一份数据，默认无法共享数据

# 进程数据共享（默认不共享）

from multiprocessing import Process, Array
from multiprocessing import Manager
import time
li = []

def foo(i):
    li.append(i)
    print(li)
    print('say hi', i)

if __name__ == '__main__':
    for i in range(3):
        p = Process(target=foo, args=(i, ))
        p.start()
    print('main proccess stop')
    
----------输出结果-----------
    main proccess stop
    [1]
    say hi 1
    [0]
    say hi 0
    [2]
    say hi 2

1）、通过Array 实现进程间数据共享

# 通过Array实现进程数据共享

from multiprocessing import Process, Array
import time
temp = Array('i', [11, 22, 33, 44, 55])

def foo(i):
    temp[i] = 100 + i
    for item in temp:
        print('第%s个进程' % i, '--------->', item)


if __name__ == '__main__':
    for i in range(1, 4, 1):
        p = Process(target=foo, args=(i, ))
        p.start()
    print('main proccess stop')

输出结果

----------输出结果------------
    main proccess stop
    第1个进程 ---------> 11
    第1个进程 ---------> 101
    第1个进程 ---------> 33
    第1个进程 ---------> 44
    第1个进程 ---------> 55
    第2个进程 ---------> 11
    第2个进程 ---------> 22
    第2个进程 ---------> 102
    第2个进程 ---------> 44
    第2个进程 ---------> 55
    第3个进程 ---------> 11
    第3个进程 ---------> 22
    第3个进程 ---------> 33
    第3个进程 ---------> 103
    第3个进程 ---------> 55

2）、通过manage.dict()实现进程间数据共享

from multiprocessing import Process, Manager
import time

mangers = Manager()
dic = mangers.dict()

def foo(i):
    time.sleep(1)
    dic[i] = 100 + 1
    print(dic.values())

for i in range(10):
    p = Process(target=foo, args=(i, ))
    p.start()
    p.join()

print('main proccess stop')

-------------输出结果-----------------
[101]
[101, 101]
[101, 101, 101]
[101, 101, 101, 101]
[101, 101, 101, 101, 101]
[101, 101, 101, 101, 101, 101]
[101, 101, 101, 101, 101, 101, 101]
[101, 101, 101, 101, 101, 101, 101, 101]
[101, 101, 101, 101, 101, 101, 101, 101, 101]
[101, 101, 101, 101, 101, 101, 101, 101, 101, 101]
main proccess stop

  'c': ctypes.c_char,  'u': ctypes.c_wchar,
    'b': ctypes.c_byte,  'B': ctypes.c_ubyte,
    'h': ctypes.c_short, 'H': ctypes.c_ushort,
    'i': ctypes.c_int,   'I': ctypes.c_uint,
    'l': ctypes.c_long,  'L': ctypes.c_ulong,
    'f': ctypes.c_float, 'd': ctypes.c_double

 3）、通过queues.Queue()实现进程间数据共享

# Queues 数据共享

from multiprocessing import Process, queues
import multiprocessing

def f(i, q):
    print(i, q.get())

if __name__ == '__main__':
    q = queues.Queue(ctx=multiprocessing)

    q.put('h1')
    q.put('h2')
    q.put('h3')

    for i in range(3):
        p = Process(target=f, args=(i, q, ))
        p.start()

---------输出结果----------
    2 h1
    1 h2
    0 h3

当创建进程时（非使用时），共享数据会被拿到子进程中，当进程中执行完毕后，再赋值给原值。

from multiprocessing import Process, Array, RLock

def Foo(lock,temp,i):
    """
    将第0个数加100
    """
    lock.acquire()
    temp[0] = 100 + i
    for item in temp:
        print(i,'----->',item)
    lock.release()

lock = RLock()
temp = Array('i', [11, 22, 33, 44])
if __name__ == '__main__':
    for i in range(10):
        p = Process(target=Foo,args=(lock,temp,i,))
        p.start()

1 -----> 101
1 -----> 22
1 -----> 33
1 -----> 44
0 -----> 100
0 -----> 22
0 -----> 33
0 -----> 44
2 -----> 102
2 -----> 22
2 -----> 33
2 -----> 44
3 -----> 103
3 -----> 22
3 -----> 33
3 -----> 44
4 -----> 104
4 -----> 22
4 -----> 33
4 -----> 44
5 -----> 105
5 -----> 22
5 -----> 33
5 -----> 44
9 -----> 109
9 -----> 22
9 -----> 33
9 -----> 44
7 -----> 107
7 -----> 22
7 -----> 33
7 -----> 44
8 -----> 108
8 -----> 22
8 -----> 33
8 -----> 44
6 -----> 106
6 -----> 22
6 -----> 33
6 -----> 44

 3、进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。

# 进程池

from multiprocessing import Pool
from time import sleep

def f(x):
    for i in range(3):
        print('%s---------->%s' % (i, x))
        sleep(1)

def main():
    # 创建进程池，进程池最多有三个可用进程
    pool = Pool(processes=3)
    for i in range(11, 20):
        result = pool.apply_async(f, (i,))
    pool.close()
    pool.join()
    if result.successful():
        print('successful')

if __name__ == '__main__':
    main()

'''
先创建容量为3的进程池，
然后将f(i)依次传递给它，运行脚本后利用ps aux | grep pool.py查看进程情况，会发现最多只会有三个进程执行。
pool.apply_async()  用来向进程池提交目标请求，
pool.join()         是用来等待进程池中的worker进程执行完毕，防止主进程在worker进程结束前结束。但必pool.join()必须使用在pool.close()或者pool.terminate()之后。
close()跟terminate()的区别
    close()会等待池中的worker进程执行结束再关闭pool,
    terminate()则是直接关闭。
result.successful()表示整个调用执行的状态，如果还有worker没有执行完，则会抛出AssertionError异常。

'''

函数解释：
•apply_async(func[, args[, kwds[, callback]]]) 它是非阻塞，apply(func[, args[, kwds]])是阻塞的
•close()    关闭pool，使其不在接受新的任务。
•terminate()    结束工作进程，不在处理未完成的任务。
•join()    主进程阻塞，等待子进程的退出， join方法要在close或terminate之后使用。

# 阻塞进程池

from multiprocessing import Pool
import time
def func(msg):
    for i in range(3):
        print('msg:', msg)
        time.sleep(1)
        print('end')

def main():
    pool = Pool(processes=3)
    for i in range(4):
        msg = 'hello %s' % i
        # 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
        result = pool.apply(func, (msg,))
    pool.close()
    pool.join()
    # 调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束

    print('successful')

if __name__ == '__main__':
    main()