Python轻量级线程池实现原理与完整代码

Python轻量级线程池原理与架构设计

线程池的核心目标是避免频繁创建和销毁线程带来的系统开销，同时限制并发数量防止资源耗尽。该方案采用生产者-消费者模式：主线程负责将待执行任务推入统一队列，固定数量的工作线程从队列中阻塞式获取任务并调用对应函数执行。整体架构依赖标准threading库完成，无需引入第三方依赖。

第1步：定义任务数据结构

首先封装一个轻量级的任务载体类，用于携带目标函数及其参数。保持数据隔离便于后续扩展参数类型。
代码入口：class TaskInfo
关键参数：func（可调用的函数对象）、parm0/1/2（传递的三个位置参数）

import threading
import time
import signal
import os

class TaskInfo:
    def __init__(self):
        self.func = None
        self.parm0 = None
        self.parm1 = None
        self.parm2 = None

第2步：构建带同步控制的任务队列

线程安全是并发编程的基础。队列需同时使用互斥锁（Lock）保证读写原子性，以及信号量（Semaphore）充当计数器驱动阻塞等待。当队列为空时，工作线程不会轮询消耗CPU，而是进入休眠直到新任务到达。
入口方法：append() 添加任务，fetch() 取出任务
重要参数：self.mutex 保护列表操作，self.sem 控制线程唤醒与阻塞

class TaskList:
    def __init__(self):
        self.tl = []
        self.mutex = threading.Lock()
        self.sem = threading.Semaphore(0)

    def append(self, ti):
        # 加锁插入后释放锁，随后释放信号量唤醒等待线程
        with self.mutex:
            self.tl.append(ti)
        self.sem.release()

    def fetch(self):
        # 先等待信号量（任务到达），再加锁弹出队首元素
        self.sem.acquire()
        with self.mutex:
            ti = self.tl.pop(0)
        return ti

第3步：实现工作线程与线程池对象

工作线程继承自threading.Thread，其run方法为无限循环，持续调用队列的获取方法。只要进程未中断，线程将自动处理分配到的任务。线程池类仅负责批量实例化并启动这些工作线程。
入口：class WorkerThread 与 class ThreadPool
操作步骤：pool.start() 触发所有线程进入就绪状态

class WorkerThread(threading.Thread):
    def __init__(self, task_queue):
        super().__init__()
        self.task_queue = task_queue

    def run(self):
        while True:
            task = self.task_queue.fetch()
            if task and task.func:
                task.func(task.parm0, task.parm1, task.parm2)

class ThreadPool:
    def __init__(self, thread_count, task_queue):
        self.threads = []
        for _ in range(thread_count):
            self.threads.append(WorkerThread(task_queue))

    def start(self):
        for t in self.threads:
            t.start()

第4步：启动服务与主循环分发

在主程序中绑定操作系统信号以便在终端中断时正常终止进程。创建队列与线程池实例后，主线程以固定时间间隔（本例为2秒）生成任务并注入队列。此阶段观察控制台输出即可确认线程调度是否生效。
入口：脚本根路径直接运行
完成状态判断：终端持续打印count:X, thread_name:Thread-X且无阻塞报错，说明并发通道已打通。

def worker_func(parm0=None, parm1=None, parm2=None):
    print(f'count:{parm0}, thread_name:{threading.current_thread().name}')

def cleanup(signum, frame):
    print(f'Received signal {signum}, exiting gracefully.')
    # 注意：os._exit(0) 会强制终止进程而不执行普通清理逻辑，适用于快速退避场景
    os._exit(0)

if __name__ == '__main__':
    signal.signal(signal.SIGINT, cleanup)
    signal.signal(signal.SIGQUIT, cleanup)
    signal.signal(signal.SIGTERM, cleanup)

    task_queue = TaskList()
    pool = ThreadPool(6, task_queue)
    pool.start()

    count = 0
    try:
        while True:
            task = TaskInfo()
            task.parm0 = count
            task.func = worker_func
            task_queue.append(task)
            count += 1
            time.sleep(2)
    except KeyboardInterrupt:
        cleanup(None, None)

运行状态验证与安全退出

运行该脚本后，系统会预创建6个工作线程。每当time.sleep(2)结束，主线程将新任务压入队列，任意一个空闲的工作线程都会立即响应并执行worker_func。由于信号量初始值为0，首次运行时工作线程会自动阻塞，直到第一个任务进入队列后才被唤醒，体现了高效的低开销等待机制。若需停止服务，按下Ctrl+C或发送SIGTERM将触发信号处理器，程序输出退出信息后直接终止。建议在涉及文件写入或网络连接的长生命周期业务中替换os._exit()，改用标志位配合join()实现平滑关闭。

总结

以上就是Python轻量级线程池实现原理与代码优化的详细内容，更多关于Python多线程调度与并发架构的资料请关注本站其它相关文章！