multiprocessing 测试代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19


def pooltest(key):

    print('%s :start test!'%(key))

    time.sleep(3)

    print('%s :end test'%(key))



if __name__ == '__main__':

    if sys.argv[1]=='test_pool':

        from multiprocessing import Pool

        print('start test pool')

        results = []

        pool = Pool(processes=3)  # 设定并发进程的数量

        for key in range(10):

            results.append(pool.apply_async(pooltest, (key,)))

        pool.close()

        pool.join()

        print('end')

        for res in results:

            if type(res.get())!=type(None):

                print(res.get())