使用multiprocessing库,涉及到进程间共享数据

Manager支持以下类型:
list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Queue, Value and Array.

1.当使用进程池Pool时,得使用Manager来进行进程池中通信管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
def test(arg1,arg2,flag):
    while 1:
        if flag.value>0:
            return
        try:
            if(arg1<3):
                raise Exception("xxx")
        except Exception as a:
            print(f"{os.getpid()}errod:{a}")
            sleep(1)
        else:
            flag.value=arg1+arg2
            print(f"{os.getpid()}else,",flag.value)
if __name__ == '__main__':
    pool=multiprocessing.Pool()
    manager=multiprocessing.Manager()
    flag=manager.Value("i",0)
    for i in range(1,4):
        pool.apply_async(test,(i,i+1,flag))
    pool.close()
    pool.join()
    print(f"flag:{flag.value}")

2不使用进程池时可以不用Manager类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def test(arg1,arg2,flag):
    while 1:
        if flag.value>0:
            return
        try:
            if(arg1<3):
                raise Exception("xxx")
        except Exception as a:
            print(f"{os.getpid()}errod:{a}")
            sleep(1)
        else:
            flag.value=arg1+arg2
            print(f"{os.getpid()}else,",flag.value)
if __name__ == '__main__':
    flag=multiprocessing.Value("i",0)
    ps = [multiprocessing.Process(target=test, args=(x, x+1, flag)) for x in range(4)]
    for p in ps:
        p.start()
    for p in ps:
        p.join()
    print(f"flag:{flag.value}")