Python利用多进程提升代码执行效率
from multiprocessing import Pool, Process, cpu_count
import threading
from tqdm import tqdm
import os
import time
# 分别对很大的若干个数字计算范围内的素数个数
# POWER = 4
POWER = 5
UPPER_LIST = [7 ** POWER, 8 ** POWER, 9 ** POWER, 10 ** POWER, 6 ** POWER, 7 ** POWER, 8 ** POWER, 9 ** POWER, 10 ** POWER]
TASK_NUM = 6
UPPER_LIST = UPPER_LIST[: TASK_NUM]
def long_time_task(task_info, upper):
"""
用n^2复杂度的素数筛选去模拟长运行时间的任务
"""
def count_prime(upper_bound):
cnt = 0
for e in tqdm(list(range(upper_bound+1))):
if e <= 1:
continue
elif e == 2:
cnt += 1
continue
else:
flag = True
for i in list(range(2, e)):
if e % i == 0:
flag = False
break
if flag:
cnt += 1
return cnt
print(f'pid: {os.getpid()}, task: {task_info}, upper: {upper}, prime_cnt: {count_prime(upper)}')
if __name__=='__main__':
print("=================================================================================================")
print(f'test1: father pid: {os.getpid()}')
start = time.time()
for i in range(TASK_NUM):
long_time_task(i, UPPER_LIST[i])
end = time.time()
print(f"test2: use {round(end-start, 2)} seconds")
print("=================================================================================================")
print(f'test2: father pid: {os.getpid()}')
start = time.time()
p_list = [Process(target=long_time_task, args=(i, UPPER_LIST[i])) for i in range(TASK_NUM)]
for p in p_list:
p.start()
for p in p_list:
p.join()
end = time.time()
print(f"test2: use {round(end-start, 2)} seconds")
print("=================================================================================================")
print(f'test3: father pid: {os.getpid()}')
print(f"CPU cores num:{cpu_count()}")
start = time.time()
p = Pool(cpu_count())
for i in range(TASK_NUM):
p.apply_async(long_time_task, args=(i, UPPER_LIST[i]))
p.close()
p.join()
end = time.time()
print(f"test3: use {round(end-start, 2)} seconds")
# print("=================================================================================================")
# print(f'test4: main thread: {threading.current_thread().name}')
# start = time.time()
# thread_list = [threading.Thread(target=long_time_task, args=(i, upper_list[i])) for i in range(9)]
# for t in thread_list:
# t.start()
# for t in thread_list:
# t.join()
# end = time.time()
# print(f"test4: use {round(end-start, 2)} seconds")
执行结果
test1:
Reference
转载请注明来源, from goldandrabbit.github.io
