python线程池实现

Posted rianley 的博客

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python线程池实现相关的知识,希望对你有一定的参考价值。

简单实现一个线程池:

import queue
import threading
import time

class ThreadPool(object):  #创建线程池类

    def __init__(self, max_num=20):  #创建一个最大长度为20的队列
        self.queue = queue.Queue(max_num)  #创建一个队列
        for i in range(max_num):  #循环把线程对象加入到队列中
            self.queue.put(threading.Thread)  #把线程的类名放进去,执行完这个Queue

    def get_thread(self):  #定义方法从队列里获取线程
        return self.queue.get()  #在队列中获取值

    def add_thread(self):  #线程执行完任务后,在队列里添加线程
        self.queue.put(threading.Thread)




def func(pool,a1):
    time.sleep(1)
    print(a1)
    pool.add_thread()  #线程执行完任务后,队列里再加一个线程

p = ThreadPool(10)  #执行init方法;  一次最多执行10个线程

for i in range(100):
    thread = p.get_thread()  #线程池10个线程,每一次循环拿走一个拿到类名,没有就等待
    t = thread(target=func, args=(p, i,))  #创建线程;  线程执行func函数的这个任务;args是给函数传入参数
    t.start()  #激活线程

  

复杂线程池

线程池要点:
1,创建线程池时,是在需要执行线程的时候创建线程,而不是创建好最大队列等待执行
2,创建一个回调函数,检查出剩余队列的任务,当线程执行完函数的时候通知线程池,
3,使用线程池时让其循环获取任务,并执行
4,线程池,让其自行的去激活线程,执行完成后,关闭退出

 

import queue
import threading
import time
import contextlib

StopEvent = object()


class ThreadPool(object):

    def __init__(self, max_num):
        self.q = queue.Queue()  # 最多创建的线程数(线程池最大容量)
        self.max_num = max_num

        self.terminal = False  #如果为True 终止所有线程,不在获取新任务
        self.generate_list = []  # 真实创建的线程列表
        self.free_list = []# 空闲线程数量

    def run(self, func, args, callback=None):
        """
        线程池执行一个任务
        :param func: 任务函数
        :param args: 任务函数所需参数
        :param callback: 任务执行失败或成功后执行的回调函数,回调函数有两个参数1、任务函数执行状态;2、任务函数返回值(默认为None,即:不执行回调函数)
        :return: 如果线程池已经终止,则返回True否则None
        """

        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
            self.generate_thread()  #创建线程
        w = (func, args, callback,)  #把参数封装成元祖
        self.q.put(w)  #添加到任务队列

    def generate_thread(self):
        """
        创建一个线程
        """
        t = threading.Thread(target=self.call)
        t.start()

    def call(self):
        """
        循环去获取任务函数并执行任务函数
        """
        current_thread = threading.currentThread  # 获取当前线程
        self.generate_list.append(current_thread)  #添加到已经创建的线程里

        event = self.q.get()  # 取任务并执行
        while event != StopEvent:  # 是元组=》是任务;如果不为停止信号  执行任务

            func, arguments, callback = event  #解开任务包; 分别取出值
            try:
                result = func(*arguments)  #运行函数,把结果赋值给result
                status = True  #运行结果是否正常
            except Exception as e:
                status = False  #表示运行不正常
                result = e  #结果为错误信息

            if callback is not None:  #是否存在回调函数
                try:
                    callback(status, result)  #执行回调函数
                except Exception as e:
                    pass

            if self.terminal:  # 默认为False,如果调用terminal方法
                event = StopEvent  #等于全局变量,表示停止信号
            else:
                # self.free_list.append(current_thread)  #执行完毕任务,添加到闲置列表
                # event = self.q.get()  #获取任务
                # self.free_list.remove(current_thread)  # 获取到任务之后,从闲置列表中删除;不是元组,就不是任务
                with self.worker_state(self.free_list, current_thread):
                    event = self.q.get()

        else:
            self.generate_list.remove(current_thread)  #如果收到终止信号,就从已经创建的线程列表中删除

    def close(self):  #终止线程
        num = len(self.generate_list)  #获取总共创建的线程数
        while num:
            self.q.put(StopEvent)  #添加停止信号,有多少线程添加多少表示终止的信号
            num -= 1


    def terminate(self):   #终止线程(清空队列)

        self.terminal = True  #把默认的False更改成True

        while self.generate_list:  #如果有已经创建线程存活
            self.q.put(StopEvent)  #有几个线程就发几个终止信号
        self.q.empty()  #清空队列

    @contextlib.contextmanager
    def worker_state(self, state_list, worker_thread):
        state_list.append(worker_thread)
        try:
            yield
        finally:
            state_list.remove(worker_thread)




def work(i):
    print(i)

pool = ThreadPool(10)
for item in range(50):
    pool.run(func=work, args=(item,))
# 将任务放在队列中
#      着手开始处理任务
#         - 创建线程
#                 - 有空闲线程,择不再创建线程
#                 - 不能高于线程池的限制
#                 - 根据任务个数判断
#         - 线程去队列中取任务

pool.terminate()

  

本文参考oldboy吴佩琦老师(详细地址:)http://www.cnblogs.com/wupeiqi/articles/4839959.html

 

以上是关于python线程池实现的主要内容,如果未能解决你的问题,请参考以下文章

Motan在服务provider端用于处理request的线程池

python全栈脱产第37天------进程池与线程池协程gevent模块单线程下实现并发的套接字通信

python 之 线程池实现并发

Python 37 进程池与线程池 协程

122 Python程序中的多进程和多线程

使用concurrent.futures模块并发,实现进程池线程池