Python程序中的线程操作(线程池)-concurrent模块

Python程序中的线程操作(线程池)-concurrent模块

目录

• Python程序中的线程操作(线程池)-concurrent模块
• 一、Python标准模块——concurrent.futures
• 二、介绍
• 三、基本方法
• 四、ProcessPoolExecutor
• 五、ThreadPoolExecutor
• 六、map的用法
• 七、回调函数

一、Python标准模块——concurrent.futures

官方文档：https://docs.python.org/dev/library/concurrent.futures.html

二、介绍

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

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

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

Both implement the same interface, which is defined by the abstract Executor class.

三、基本方法

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()：取消某个任务

四、ProcessPoolExecutor

介绍

The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. ProcessPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but also means that only picklable objects can be executed and returned.

class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None)

An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workers is None or not given, it will default to the number of processors on the machine. If max_workers is lower or equal to 0, then a ValueError will be raised.

#用法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os, time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ProcessPoolExecutor(max_workers=3)

    futures=[]
    for i in range(11):
        future=executor.submit(task,i)
        futures.append(future)
    executor.shutdown(True)
    print('+++>')
    for future in futures:
        print(future.result())

五、ThreadPoolExecutor

介绍

ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously.

class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='')

An Executor subclass that uses a pool of at most max_workers threads to execute calls asynchronously.

Changed in version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for ProcessPoolExecutor.

New in version 3.6: The thread_name_prefix argument was added to allow users to control the threading.Thread names for worker threads created by the pool for easier debugging.

#用法
与ProcessPoolExecutor相同
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from threading import currentThread
from multiprocessing import current_process
import time

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

if __name__ == '__main__':
    # 池子里只有四个线程
    # pool = ThreadPoolExecutor(4) # 池子里面有4个线程

    # 池子里有四个进程
    pool = ProcessPoolExecutor(4)

    fu_list = []

    for i in range(20):
        # task任务要做20次， 4个进程负责做这个事
        future = pool.submit(task, i)  # task任务要做20次，4个进程负责做这个事情
        fu_list.append(future)

    # 关闭池的入口， 会等待所有的任务执行完，结束阻塞
    pool.shutdown()
    for fu in fu_list:
        print(fu.result())

六、map的用法

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

import os, time, random

def task(n):
    print('%s is runing' % os.getpid())
    time.sleep(random.randint(1, 3))
    return n ** 2

if __name__ == '__main__':
    executor = ThreadPoolExecutor(max_workers=3)

    # for i in range(11):
    #     future=executor.submit(task,i)

    res = executor.map(task, range(1, 12))  # map取代了for+submit
    executor.shutdown()
    for r in res:
        print(r)

54480 is runing

54480 is runing

54480 is runing

54480 is runing

54480 is runing

54480 is runing

54480 is runing

54480 is runing

54480 is runing

54480 is runing

54480 is runing

1

4

9

16

25

36

49

64

81

100

121

七、回调函数

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os

def get_page(url):
    print('<进程%s> get %s' %(os.getpid(),url))
    respone=requests.get(url)
    if respone.status_code == 200:
        return {'url':url,'text':respone.text}

def parse_page(res):
    res=res.result()
    print('<进程%s> parse %s' %(os.getpid(),res['url']))
    parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))
    with open('db.txt','a') as f:
        f.write(parse_res)

if __name__ == '__main__':
    urls=[
        'https://www.baidu.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://help.github.com/',
        'http://www.sina.com.cn/'
    ]

    # p=Pool(3)
    # for url in urls:
    #     p.apply_async(get_page,args=(url,),callback=pasrse_page)
    # p.close()
    # p.join()

    p=ProcessPoolExecutor(3)
    for url in urls:
        p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果