线程与进程代码示例

import multiprocessing
import concurrent.futures

# 分割成list
def data_split(batch_data, batch_num):
    split_data = [[] for _ in range(batch_num)]
    for i, data in enumerate(batch_data):
        split_data[i%batch_num].append(data)
    
    return split_data

# 分割成dict
'''
def data_split(self, batch_data, batch_num):
	split_data = [{} for _ in range(batch_num)]
	for i, data in enumerate(batch_data.items()):
	    split_data[i%batch_num][data[0]] = data[1]
	return split_data
'''

num_threads = multiprocessing.cpu_count()  # 获取CPU核心数
batch_data = data_split(export_data, num_threads)

executor = concurrent.futures.ThreadPoolExecutor(max_workers=num_threads)
    
for i in range(num_threads):
     executor.submit(err_vis.draw_all_info, batch_data[i])

# 关闭线程池
executor.shutdown(wait=True)

多进程

import multiprocessing

def data_split(batch_data, batch_num):
    split_data = [[] for _ in range(batch_num)]
    for i, data in enumerate(batch_data):
        split_data[i % batch_num].append(data)
    return split_data

num_processes = multiprocessing.cpu_count()  # 获取CPU核心数
batch_data = data_split(export_data, num_processes)

processes = []
for i in range(num_processes):
    p = multiprocessing.Process(target=err_vis.draw_all_info, args=(batch_data[i],))
    p.start()
    processes.append(p)

for p in processes:
    p.join()

import multiprocessing

def process_data(data):
    err_vis.draw_all_info(data)

num_processes = multiprocessing.cpu_count()  # 获取CPU核心数
batch_data = data_split(export_data, num_processes)

with multiprocessing.Pool(processes=num_processes) as pool:
    pool.map(process_data, batch_data)

同步调用

from multiprocessing import Pool
import time,os,random
def test(n):
    print(f"子进程:{os.getpid()}")
    time.sleep(2)
    return n*random.randint(2,9)

if __name__ == '__main__':
    n = os.cpu_count()     # 本机CPU个数,我的是4,进程池容量个数自定义,默认CPU核数
    p = Pool(processes=n)  # 设置进程池进程个数,从无到有,并且以后一直只有这四个进程在执行任务
    li = []
    start_time = time.time()
    for i in range(10):
        res = p.apply(test,args=(2,))  # 创建十个个任务, 使用同步调用的方式
        li.append(res)
    p.close()  # 先关闭进程池, 不会再有新的进程加入到pool中, 防止进一步的操作(同步调用可以不加此方法)
    p.join()   # 必须在close调用之后执行, 否则报错, 执行后等待所有子进程结束(同步调用可以不加此方法)
    print(li)  # 同步调用, 得到的就是最终结果,(异步调用得到的是对象, 需要使用get方法取值)
    print(f'使用时间:{time.time()-start_time}')
    
'''输出
子进程:7768
子进程:16276
子进程:17544
子进程:15680
子进程:7768
子进程:16276
子进程:17544
子进程:15680
子进程:7768
子进程:16276
[4, 18, 14, 14, 12, 14, 16, 14, 6, 10]
使用时间:20.226498126983643
'''

异步调用
from multiprocessing import Pool
import time,os,random

def test(n):
    print(f"子进程:{os.getpid()}")
    time.sleep(2)
    return n*n*random.randint(2,9)

if __name__ == '__main__':
    n = os.cpu_count()     # 本机CPU个数,我的是4,进程池容量个数自定义,默认CPU核数
    p = Pool(processes=n)  # 设置进程池大小, 从无到有, 并之后只有这四个进程执行任务
    li = []
    start_time = time.time()
    for i in range(10):
        res = p.apply_async(test,args=(2,))  # 开启十个任务, 使用异步调用的方式
        li.append(res)
    p.close()  # 关闭进程池, 不会再有新的进程加入到pool中, 防止进一步的操作
    p.join()   # join必须在close函数之后进行, 否则报错, 执行后等待所有子进程结束
    print(li)  # 返回的是AsyncResul的对象[<multiprocessing.pool.ApplyResult object at 0x000002318511B408>,....]
    print([i.get() for i in li])  # 使用get方法来获取异步调用的值(同步调用没有该方法),并放入列表中打印
    print(f"使用时间:{time.time()-start_time}")

'''输出
子进程:8636
子进程:10828
子进程:7432
子进程:13976
子进程:8636
子进程:10828
子进程:7432
子进程:13976
子进程:8636
子进程:10828
[<multiprocessing.pool.ApplyResult object at 0x000001623059B308>,...省略]
[16, 24, 24, 24, 16, 28, 36, 28, 8, 32]
使用时间:6.301024436950684
'''

同步调用和异步调用
http://t.csdn.cn/wNoEA