线程池实现

线程池是一个线程管理技术，创建一个或者多个线程进行管理，避免线程的创建和销毁带来的开销线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度.

• 线程池的优点
  • 降低资源消耗；通过重复利用已创建的线程降低创建和销毁造成的消耗。
  • 提高响应速度，不必等待线程的创建，正常情况下(没有任务进入队列的情况)不需要等待。
  • 线程管理，线程统一由线程池管理，随取随用。

自定义线程池实现

from threading import Thread
from queue import Queue
import time
import random
import threading


# 自定义线程池
class MyThreadPool(Thread):

    def __init__(self, thread_num, task_queue):
        super().__init__()
        self.daemon = True
        self.thread_num = thread_num
        self.task_queue = task_queue
        self.threads = None
        # 初始化的时候启动线程池
        self.__start()

    def run(self):
        while True:
            target, args, kwargs = self.task_queue.get()
            target(*args, **kwargs)
            # 队列计数器减一
            self.task_queue.task_done()

    # 重写父类join 方法,将等待结束条件修改为: 任务队列执行完成
    def join(self):
        self.task_queue.join()

    def __start(self):
        self.threads = self.__create_threads()
        for thread in self.threads:
            thread.start()

    def __create_threads(self):
        return [MyThreadPool(0, self.task_queue) for _ in range(self.thread_num)]

    def join_all(self):
        for thread in self.threads:
            thread.join()

    def submit_task(self, target, args=(), kwargs={}):
        self.task_queue.put((target, args, kwargs))


def work(name, no, **kwargs):
    print(f"【子线程{no}-开始】({threading.current_thread()}) ")
    print(f"【子线程{no}-data】 name={name}, no={no}, kwargs={kwargs}")
    time.sleep(random.randint(1, 10))
    print(f"【子线程{no}-结束】({threading.current_thread()})")


q = Queue()

t_pool = MyThreadPool(3, q)
for i in range(5):
    t_pool.submit_task(work, (f"任务{i}", i), {"test": 1})

t_pool.join_all()

【运行效果】: