当前位置：首页 > news >正文

python进程池的使用

news 2025/7/27 4:20:02

进程池的创建

apply()

apply()方法用于向进程池提交一个任务，并等待任务完成并返回结果。

apply_async()

apply_async()方法用于向进程池提交一个异步任务（即无需等待任务完成），将任务加入到进程池的队列里，并立即返回一个AsyncResult对象。

# Pool 表示一个工作进程池
# 如果要启动大量的子进程，可以用进程池的方式批量创建子进程
from multiprocessing.pool import Pool
from time import sleep, time
import random,osdef run(name):print(f'{name},子进程开始，进程ID:{os.getpid()}')start = time()sleep(random.choice([1,2,3,4]))end = time()print(f'{name}子进程结束，进程ID:{os.getpid()},耗时{end-start}')if __name__ == '__main__':print("父进程开始")# 创建多个进程，表示可以同时执行的进程数量，默认大小时cpu的数量p = Pool(4)for i in range(10):# 创建进程，放入进程池统一管理p.apply_async(run,args=(i,))# 如果我们用的时进程池，再调用join()之前必须先调用closs()# 并且closs()之后，不能再继续往进程池添加新的进程了p.close()# 进程池对象调用join()，会等待进程池中所有进程都结束完毕再去结束父进程p.join()print("父进程结束")p.terminate()# close() ：如果我们用的是进程池，再调用join()之前必须要先close()
# 并且closs()之后，不能再继续往进程池添加新的进程了
# join():进程池对象调用join()，会等待进程池中所有进程都结束完毕再去结束父进程
# terminate()：一旦运行到此步，不管任务是否完成，立即终止

timeout的使用

from multiprocessing import Pool
import timedef f(x):return x*xif __name__ == '__main__':with Pool(processes = 4) as pool:                   # 进程池包含4个进程result = pool.apply_async(f,args=(10,))         # 执行1个子进程print(result.get(timeout=1))                    # 显示执行结果result = pool.apply_async(time.sleep,args=(10,))print(result.get(timeout=1))                    # raise multiprocessing.TimeoutError

map的使用

map()方法用于向进程池提交多个任务，并等待所有任务完成并返回结果。

import multiprocessing as mpdef task(num):return num * numif __name__ == "__main__":pool = mp.Pool()res = pool.map(task, [1, 2, 3, 4, 5])print(res)

from multiprocessing import Pool
import timedef f(x):return x * xif __name__ == '__main__':with Pool(processes=4) as pool:print(pool.map(f,range(10)))   # 输出"[0,1,4.....81]"it = pool.imap(f,range(10))  # map输出列表,imap输出迭代器print(it)print(next(it))             # "0"print(next(it))             # "1"print(it.next(timeout = 1)) # "4"

map_async()

map_async()方法用于向进程池提交多个异步任务（即无需等待任务完成），并立即返回一个AsyncResult对象。

import multiprocessing as mpdef task(num):return num * numif __name__ == "__main__":pool = mp.Pool()res = pool.map_async(task, [1, 2, 3, 4, 5])print(res.get())

close()和join()

close()方法用于向进程池发送一个停止信号，表示不再接受新任务。
join()方法则用于等待进程池中的所有任务完成。

import multiprocessing as mpdef task(num):return num * numif __name__ == "__main__":pool = mp.Pool()res = pool.map_async(task, [1, 2, 3, 4, 5])pool.close()pool.join()print(res.get())