Python常用模块threading与Thread线程池实战指南

线程池核心概念与运行机制

线程池的运行遵循典型的生产者与消费者模型：

• 主线程（生产者）：仅负责向线程池内部的任务队列提交待执行函数与参数，无需感知底层的具体执行路径。
• 线程池（消费者）：预先创建固定数量的工作线程常驻后台，持续轮询任务队列。一旦获取到待处理对象，立即分配给空闲的工作线程执行。

该设计有效避免了高频线程创建与上下文切换带来的性能损耗，特别适合IO密集型并发场景。

基于Queue与Threading自定义线程池

借助Python内置的threading与queue模块，可以轻量级构建支持异步提交的线程池类。核心逻辑包含初始化守护线程、循环消费队列以及同步等待任务完成。

# -*- coding: utf-8 -*-
from threading import Thread
from queue import Queue
import time

class ThreadPool:
    def __init__(self, n):
        self.queue = Queue()
        # 创建指定数量的守护线程并启动
        for i in range(n):
            Thread(target=self.worker, daemon=True).start()

    # 线程工作循环
    def worker(self):
        while True:
            # 阻塞获取队列中的任务元组
            func, args, kwargs = self.queue.get()
            try:
                func(*args, **kwargs)
            finally:
                self.queue.task_done()

    # 异步提交任务
    def apply_async(self, target, args=(), kwargs=None):
        if kwargs is None:
            kwargs = {}
        self.queue.put((target, args, kwargs))

    # 阻塞主线程直到所有任务完成
    def join(self):
        self.queue.join()

def task_func(x):
    print(f'正在执行任务 {x}')
    time.sleep(2)
    print(f'任务 {x} 执行完毕')

# 初始化含2个工作线程的池
t = ThreadPool(2)
for i in range(5):
    t.apply_async(task_func, args=(i,))
print('全部任务已入队，等待执行结束...')
t.join()
print('所有线程池任务已完成')