基于队列queue实现的线程池

本文通过文章同步功能推送至博客园，显示排版可能会有所错误，请见谅!

写在前文：在Python中给多进程提供了进程池类，对于线程，Python2并没有直接提供线程池类（Python3中提供了线程池功能），而线程池在并行中应用较广泛，因此实现一个进程池的功能十分必要。本文基于队列（queue）功能来实现线程池功能。

在Python3标准库中提供了线程池、进程池功能，推荐使用标准库。

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import ProcessPoolExecutor

实现代码：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
__auth__ = "SongWei"
import threading,queue,time
 
class Threadpool:
    '''基于队列queue实现的线程池'''
 
    def __init__(self,max_thread=1):
        '''创建进程队列'''
        self.queue = queue.Queue(maxsize=max_thread)
 
    def apply(self,target=None,args=(),callback=None,calljoin=True,**kwargs):
        ''':param callback 回调函数 当子线程函数运行结束后将返回值传入回调函数
            :param calljoin 布尔值  回调函数是否阻塞进程池 默认True 只有当目标函数和回调函数都执行结束后才视为该线程结束
            其他参数同threading.Thread类
            注意：只有当目标函数和回调函数都执行结束后，消息队列才会取回值（即回调函数会阻塞线程池）
        '''
        if not callback:
            callback = self._callback
        t = threading.Thread(target=self._decorate(target,callback,calljoin),args=args,**kwargs)
        self.queue.put(t)
        t.start()
 
    def join(self):
        '''
            当线程池中还有未执行结束的子线程时 阻塞主线程
            注意：当calljoin=False时 因回调函数在消息队列取回后才执行 故join不会等待回调函数
        '''
        while self.queue.qsize():
            time.sleep(0.05)
 
    def _decorate(self,target,callback,calljoin):
        ''':param target 接收一个目标函数
            :param callback 接受一个回调函数
            :param backjoin 布尔值 若为真 则当回调函数执行结束后才释放队列 否则 当目标函数执行结束后就会释放队列
            本函数本质上是一个装饰器，即运行目标函数后，执行队列取回(self.queque.get())，并将返回值作为参数执行回调函数。
        '''
        def wrapper(*args,**kwargs):
            res = target(*args,**kwargs)
            if calljoin:
                callback(res)
                self.queue.get()
            else:
                self.queue.get()
                callback(res)
            return res
        return wrapper
 
    def _callback(self,*args,**kwargs):
        '''没有传入回调函数时 什么也不干'''
        pass
 
调用示例：
result_list = []
def func(arg):
    print('正在等待执行%s' % arg)
    time.sleep(10)
    return arg
 
def back(res):
    print('我已经取回了数据:%s' % res)
    result_list.append(res)
 
pool = Threadpool(max_thread=20)
for i in range(40):
    pool.apply(target=func,args=(i,),callback=back)
pool.join()
print(result_list)