Python程序中的进程池和线程池

目录

一、什么是进程池或线程池

池:
池的是为了限制进程数和线程数
什么是时候该用进程池/线程池:
当python程序是 计算密集型 且并发的任务量远大于计算机所能承受的范围,
无法一次性开启过多的任务数量就应该考虑使用进程池和线程池

二、理解同步、异步、

同步: 提交了一个任务,必须等任务执行完了,才能执行下一个.无论该任务是否存在阻塞,同步调用都会原地等待,相当于串行
异步: 提交了一个任务,可以不用等任务的执行结果,直接执行下一个,并发的去执行

三、multiprocess.Pool模块实现进程池

Pool([numprocess [,initializer [, initargs]]])：创建进程池

  1. numprocess：要创建的进程数，如果省略，将默认使用cpu_count()的值
  2. initializer：是每个工作进程启动时要执行的可调用对象，默认为None
  3. initargs：是要传给initializer的参数组

3.1.1 Pool使用方法

p.apply(func [, args [, kwargs]])：在一个池工作进程中执行func(*args,**kwargs),然后返回结果。需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()

p.apply_async(func [, args [, kwargs]])：在一个池工作进程中执行func(*args,**kwargs),然后返回结果。此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。

p.close()：关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成

P.join()：等待所有工作进程退出。此方法只能在close()或teminate()之后调用

obj.get()：返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。

obj.ready()：如果调用完成，返回True

obj.successful()：如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常

obj.wait([timeout])：等待结果变为可用。

obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数

3.1.1 代码实例——multiprocess.Pool

同步运行方式(串行等待)：

def task(i):
    print(f'进程 {current_process().name} 在执行任务 {i}')
    time.sleep(1)
    return i**2,current_process().name

### Pool方式
p = Pool(3) # 创建进程池对象,可容纳3个进程

### 同步运行(串行等待)
for i in range(20):
    res = p.apply(task, args=(i,))
    print(res)

异步运行方式：

def task(i):
    print(f'进程 {current_process().name} 在执行任务 {i}')
    time.sleep(1)
    return i**2,current_process().name

for i in range(20):
    res = p.apply_async(task, args=(i,), callback=lambda res: print(res))    # 异步处理并设置了
    
    p.close()   # 先关闭进程池
    p.join()    # 在等待进程结果

四、Python标准模块——concurrent.futures实现进程池和线程池

4.1 介绍

concurrent.futures模块提供了高度封装的异步调用接口

ThreadPoolExecutor：线程池，提供异步调用

ProcessPoolExecutor：进程池，提供异步调用

4.2 基本方法

submit(fn, *args, **kwargs)：异步提交任务

map(func, *iterables, timeout=None, chunksize=1)：取代for循环submit的操作

shutdown(wait=True)：相当于进程池的pool.close()+pool.join()操作

  • wait=True，等待池内所有任务执行完毕回收完资源后才继续
  • wait=False，立即返回，并不会等待池内的任务执行完毕
  • 但不管wait参数为何值，整个程序都会等到所有任务执行完毕
  • submit和map必须在shutdown之前

result(timeout=None)：取得结果

add_done_callback(fn)：回调函数

done()：判断某一个线程是否完成

cancle()：取消某个任务

4.3 代码实例——ProcessPoolExecutor

方式1:

异步的执行任务,在把每次返回的对象添加到列表中,最后统一得到结果

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process,Pool
import time

def task(i):
    print(f'进程 {current_process().name} 在执行任务 {i}')
    time.sleep(1)
    return i**2,current_process().name

def parse(future):
    # 处理拿到的结果
    print(future.result())  # 在这里拿到结果
    
if __name__ == '__main__':
    ### ProcessPoolExecutor方式(异步调用接口)
    pool = ProcessPoolExecutor(3)   # maxsize 填最大数
    
    ### ProcessPoolExecutor方式1:
    futures = []
    for i in range(20):
        future = pool.submit(task, i)   # 提交异步任务
        futures.append(future)
    pool.shutdown() # 这句话相当于 进程池关闭和进程的join方法
    for future in futures:
        print(future.result())

方式2:

每执行一个进程都拿到结果.等于把异步任务变成同步任务,原地等待结果

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process,Pool
import time

def task(i):
    print(f'进程 {current_process().name} 在执行任务 {i}')
    time.sleep(1)
    return i**2,current_process().name

def parse(future):
    # 处理拿到的结果
    print(future.result())  # 在这里拿到结果
    
if __name__ == '__main__':
    ### ProcessPoolExecutor方式(异步调用接口)
    pool = ProcessPoolExecutor(3)   # maxsize 填最大数
    
### ProcessPoolExecutor方式2:  
    for i in range(20):
        future = pool.submit(task, i)   # 提交异步任务
        print(future.result())      # 如果加了这一句就变成 同步 串行了,获取每一次任务的执行结果
    pool.shutdown() # 这句话相当于 进程池关闭和进程的join方法

方式3

通过回调函数来异步的执行任务,并得到每一次的结果

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process,Pool
import time

def task(i):
    print(f'进程 {current_process().name} 在执行任务 {i}')
    time.sleep(1)
    return i**2,current_process().name

def parse(future):
    # 处理拿到的结果
    print(future.result())  # 在这里拿到结果
    
if __name__ == '__main__':
    ### ProcessPoolExecutor方式(异步调用接口)
    pool = ProcessPoolExecutor(3)   # maxsize 填最大数
    
##### ProcessPoolExecutor方式3:  
    for i in range(20):
        future = pool.submit(task, i)  # 提交异步任务
        future.add_done_callback(parse)   # 因此不能直接写result,要通过回调函数
    pool.shutdown() # 这句话相当于 进程池关闭和进程的join方法

4.4 代码实例——ThreadPoolExecutor

方式1：

异步的执行任务,在把每次返回的对象添加到列表中,最后统一得到结果

from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
import time

def task(i):
    print(f'{currentThread().name} 在执行任务 {i}')
    time.sleep(1)
    return i**2

def parse(future):
    # 处理拿到的结果
    print(future.result())
    
if __name__ == '__main__':
    ### ThreadPoolExecutor(异步调用接口)
    pool = ThreadPoolExecutor(3)  # 池子里只有3个线程

    ### ProcessPoolExecutor方式1:  
    futures = []
    for i in range(20):
        future = pool.submit(task, i)   # 提交异步任务
        futures.append(future)
    pool.shutdown() # 这句话相当于 线程池关闭和线程的join方法
    for future in futures:
        print(future.result())

方式2：

每执行一个线程都拿到结果.等于把异步任务变成同步任务,原地等待结果了

from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
import time

def task(i):
    print(f'{currentThread().name} 在执行任务 {i}')
    time.sleep(1)
    return i**2

def parse(future):
    # 处理拿到的结果
    print(future.result())
    
if __name__ == '__main__':
    ### ThreadPoolExecutor(异步调用接口)
    pool = ThreadPoolExecutor(3)  # 池子里只有3个线程
    
    ### ProcessPoolExecutor方式2:  
    for i in range(20):
        future = pool.submit(task, i)   # 提交异步任务
        print(future.result())      # 如果加了这一句就变成 同步 串行了,获取每一次任务的执行结果
    pool.shutdown() # 这句话相当于 线程池关闭和线程的join方法

方式3：

from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
import time

def task(i):
    print(f'{currentThread().name} 在执行任务 {i}')
    time.sleep(1)
    return i**2

def parse(future):
    # 处理拿到的结果
    print(future.result())
    
if __name__ == '__main__':
    ### ThreadPoolExecutor(异步调用接口)
    pool = ThreadPoolExecutor(3)  # 池子里只有3个线程
    
    ##### ProcessPoolExecutor方式3:  通过回调函数来异步的执行任务,并得到每一次的结果
    for i in range(20):
        future = pool.submit(task, i)  # 提交异步任务
        future.add_done_callback(parse)   # 因此不能直接写result,要通过回调函数
    pool.shutdown() # 这句话相当于 线程池关闭和线程的join方法
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/XuChengNotes/p/11553068.html