【Python】实现线程池

青旅半醒 2022-02-04 02:21 296阅读 0赞

一.线程池代码解释

1.初始化函数\_\_init\_\_():

2.线程池启动函数run():

3.创建线程函数generate\_thread()：

4.处理任务函数call():

5.关闭线程函数close():

6.立即终止线程函数terminal():

二.线程池代码块

三.线程池使用代码：

# 一.线程池代码解释 #

### 1.初始化函数\_\_init\_\_(): ###

1.  self.task:                   一个不限长度的队列，用于放待处理的任务，一端存放，另一端由线程取任务执行，初始为空
2.  self.max\_num:          线程的最大数量，默认为5
3.  self.generate\_list:     创建的线程存放在该列表中，初始为空，数量应小于self.max\_num
4.  self.free\_list:             空闲的线程存放在该列表中，初始为空
5.  self.isterminal:          一个标志位，表示进程是否终止，默认False：不终止；True：终止，不理解可以暂时先放一放。

### 2.线程池启动函数run(): ###

**传入参数：**

1.  待处理任务的函数名称
2.  传入参数（元组形式）
3.  回调函数的函数名（回调函数的作用：执行完待处理任务后，将其返回值传给回调函数作相应处理，反映任务的执行情况）

**执行逻辑：**

1.  将任务put到待处理任务队列self.task中
2.  若当前无空闲线程，且已创建的线程数未达到最大值，则执行generate\_thread()函数，创建新线程，否则不作任何操作，等待空闲线程去队列中取该任务执行

### 3.创建线程函数generate\_thread()： ###

1.  创建一个线程，将self.call()函数作为目标函数,则每一个self.call()都是一个独立的线程

### 4.处理任务函数call(): ###

1.  将当前线程加入self.generate\_list中
2.  取任务：从待处理任务队列中get任务
3.  只要取出的任务不是【任务完成符StopTask】，则处理任务并执行回调函数，否则销毁该线程
4.  任务处理完成后，如果【立即终止符self.isterminal】为True, 则该线程会在下一轮循环中被销毁；否则，获取队列中的任务继续处理任务

### 5.关闭线程函数close(): ###

**功能：**

*  该函数执行后，会等待所有任务执行完成后，销毁每一个线程

**实现逻辑：**

*  在所有任务完成后，向待处理任务队列中put【任务完成符StopTask】，在call函数中完成线程的销毁，创建了多少个线程，就put多少个StopTask.

### 6.立即终止线程函数terminal(): ###

**功能：**

*  该函数执行后，不会等待所有任务都执行完，而是等待各个线程完成当前的任务后，立即销毁所有线程。

**实现逻辑：**

1.  修改【立即终止符self.isterminal】为True；
2.  但是该方法不能终止以下类型的线程，即任务已经全部完成，但是还有一些线程处于68行代码处的阻塞状态，因此还需要向待处理任务队列中put【任务完成符StopTask】，直到self.generate\_list为空，即所有线程均被销毁。
3.  最后将任务队列中多余的StopTask清空，但是python3中的Queue已经没有清空clear()的方法的了.

# 二.线程池代码块 #

#!/usr/bin/env python
    # -*- coding:utf-8 -*-
    
    import queue
    import threading
    import time
    
    StopTask = object()
    
    
    class threadpool():
        def __init__(self, max_num=5):
            self.task = queue.Queue()
            self.max_num = max_num
    
            self.generate_list = []
            self.free_list = []
    
            self.isterminal = False
    
        def run(self, func, args, callback=None):
            """
            线程池启动函数
            :param func: 函数名称
            :param args: 函数参数
            :param callback: 回调函数
            :return:
            """
            task = (func, args, callback)
            self.task.put(task)
    
            if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
                self.generate_thread()
    
        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)
    
            get_task = self.task.get()
            while get_task != StopTask:
                # 执行任务
                func, args, callback = get_task
                try:
                    ret = func(*args)
                    status = True
                except Exception as e:
                    ret = e
                    status = False
                # 执行回调函数
                if callback:
                    try:
                        callback(status, ret)
                    except Exception:
                        pass
                # 处理任务or立即终止
                if self.isterminal:
                    # 立即终止
                    get_task = StopTask
                else:
                    # 空闲处理
                    self.free_list.append(current_thread)
                    get_task = self.task.get()
                    self.free_list.remove(current_thread)
            else:
                # 销毁线程
                self.generate_list.remove(current_thread)
    
        def close(self):
            """
            等待任务全部完成后，清空所有线程
            :return:
            """
            num = len(self.generate_list)
            while num:
                self.task.put(StopTask)
                num -= 1
    
        def terminal(self):
            """
            立即终止所有的线程
            :return:
            """
            self.isterminal = True
    
            while self.generate_list:
                self.task.put(StopTask)
    
            self.task.empty()

# 三.线程池使用代码： #

def func(i):
        """
        待处理任务
        :param i:
        :return:
        """
        time.sleep(0.5)
        print(i)
    
    
    def call(status, ret):
        """
        回调函数
        :param status: True:任务运行无误；False：报异常
        :param ret: 任务运行返回值/异常信息
        :return:
        """
        pass
    
    
    if __name__ == '__main__':
        #创建线程池
        pool = threadpool(5)
        #执行任务
        for i in range(20):
            pool.run(func, args=(i,), callback=call)
        #终止任务
        pool.close()