python: Worker Pool Pattern

项目经构：

Worker Pool（工作池）是控制并发的经典模式：
固定数量的「工人」（线程 / 进程），不随任务数量无限创建
工人从「任务队列」中领取任务执行
核心价值：限制并发数、优化资源、稳定可控、优雅关闭

# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:20 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : settings.py """ 全局配置文件：所有可变参数统一管理 符合企业级：配置与代码分离 """ import os # 工作池配置 WORKER_COUNT = 3 # 固定生产线数量（可动态修改） TASK_QUEUE_MAXSIZE = 100 # 任务队列最大长度 # 业务模拟配置 SIMULATE_TASK_DELAY = True # 是否模拟任务耗时 RANDOM_DELAY_RANGE = (0.5, 2.5) # 模拟IO延迟 # 日志配置 LOG_LEVEL = "INFO" LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s" # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:20 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : exceptions.py """ 统一异常体系：职责单一 专门处理工作池 + 业务异常 """ class WorkerPoolException(Exception): """ 工作池基础异常 """ pass class TaskExecutionException(WorkerPoolException): """ 任务执行失败异常 """ pass class TaskNotFoundException(WorkerPoolException): """ 任务不存在异常 """ pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:21 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : logger.py """ 日志工具：全局唯一日志实例 职责单一：只做日志输出 """ import logging from WorkerPoolPattern.config.settings import LOG_LEVEL, LOG_FORMAT def get_logger(name: str = "jewelry_production") -> logging.Logger: logger = logging.getLogger(name) logger.setLevel(LOG_LEVEL) if not logger.handlers: handler = logging.StreamHandler() # Windows 必加：解决中文/编码报错 handler.stream.reconfigure(encoding='utf-8') formatter = logging.Formatter(LOG_FORMAT) handler.setFormatter(formatter) logger.addHandler(handler) return logger # 全局可用的日志对象 logger = get_logger() # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:23 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : worker.py """ 核心工作池：通用、可扩展、线程安全 职责单一：只管理工人 + 任务队列 不耦合任何珠宝业务，可直接复用于其他项目 """ import queue import threading from typing import Callable, Any from WorkerPoolPattern.utils.logger import logger from WorkerPoolPattern.core.exceptions import TaskExecutionException class Worker(threading.Thread): """ 单个工作者：无限从队列取任务执行 """ def __init__(self, task_queue: queue.Queue, worker_id: int): super().__init__(daemon=True) self.queue = task_queue self.worker_id = worker_id self.name = f"Worker-{worker_id}" def run(self): """ :return: """ logger.info(f"工人【{self.worker_id}】已启动，等待任务...") while True: try: task_func, task_id, args, kwargs = self.queue.get() logger.info(f"工人【{self.worker_id}】领取任务【{task_id}】") try: task_func(*args, **kwargs) logger.info(f"工人【{self.worker_id}】完成任务【{task_id}】") except Exception as e: logger.error(f"任务【{task_id}】执行失败：{str(e)}") raise TaskExecutionException(f"任务执行异常：{e}") from e finally: self.queue.task_done() except Exception as e: logger.error(f"工人【{self.worker_id}】异常：{str(e)}") class WorkerPool: """ 工作池管理器：职责单一，只负责启动/管理/等待工人 """ def __init__(self, worker_count: int, queue_maxsize: int = 0): self.worker_count = worker_count self.task_queue = queue.Queue(maxsize=queue_maxsize) self.workers: list[Worker] = [] def start(self): """ 启动所有工作者 :return: """ logger.info(f"启动工作池，共 {self.worker_count} 个工作者") for i in range(1, self.worker_count + 1): worker = Worker(self.task_queue, i) worker.start() self.workers.append(worker) def submit(self, task_id: str, task_func: Callable, *args, **kwargs) -> None: """ 提交任务到队列（通用接口） :param task_id: :param task_func: :param args: :param kwargs: :return: """ self.task_queue.put((task_func, task_id, args, kwargs)) def wait_completion(self): """ 等待所有任务执行完成（优雅关闭） :return: """ self.task_queue.join() logger.info("所有任务已完成，工作池优雅关闭")

# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:25 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : check_tasks.py """原料质检 / 成品质检""" import time import random from WorkerPoolPattern.config.settings import RANDOM_DELAY_RANGE from WorkerPoolPattern.utils.logger import logger def raw_material_check(order_id: str): """ :param order_id: :return: """ logger.info(f"[{order_id}] 原料质检：钻石4C、真伪、纯度检测") time.sleep(random.uniform(*RANDOM_DELAY_RANGE)) def finished_goods_check(order_id: str): """ :param order_id: :return: """ logger.info(f"[{order_id}] 成品质检：工艺、成色、尺寸、镶口精度全检") time.sleep(random.uniform(*RANDOM_DELAY_RANGE)) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:27 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : process_tasks.py """首饰加工、设计、镶嵌""" import time import random from WorkerPoolPattern.config.settings import RANDOM_DELAY_RANGE from WorkerPoolPattern.utils.logger import logger def jewelry_process(order_id: str): """ :param order_id: :return: """ logger.info(f"[{order_id}] 首饰加工：设计 → 执模 → 镶嵌 → 抛光 → 电金") time.sleep(random.uniform(*RANDOM_DELAY_RANGE)) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:28 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : logistics_tasks.py """库存录入、物流发货""" import time import random from WorkerPoolPattern.config.settings import RANDOM_DELAY_RANGE from WorkerPoolPattern.utils.logger import logger def inventory_record(order_id: str): """ :param order_id: :return: """ logger.info(f"[{order_id}] 库存录入：生成唯一防伪码、ERP入库、标签打印") time.sleep(random.uniform(*RANDOM_DELAY_RANGE)) def order_delivery(order_id: str): """ :param order_id: :return: """ logger.info(f"[{order_id}] 订单发货：物流下单、保价、短信通知") time.sleep(random.uniform(*RANDOM_DELAY_RANGE)) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:29 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : production_service.py """ 珠宝生产业务服务 只负责：接收订单 → 编排任务 → 提交到工作池 """ from WorkerPoolPattern.tasks import JEWELRY_FULL_PROCESS from WorkerPoolPattern.core.worker import WorkerPool from WorkerPoolPattern.utils.logger import logger class JewelryProductionService(object): """ 业务编排层（职责单一） """ def __init__(self, pool: WorkerPool): self.pool = pool # 依赖注入：解耦 def create_order_task(self, order_id: str): """ 为单个订单创建全流程任务链 :param order_id: :return: """ logger.info(f"📦 开始创建订单【{order_id}】的全流程任务") for task_name, task_func in JEWELRY_FULL_PROCESS: task_id = f"{order_id}-{task_name}" self.pool.submit(task_id, task_func, order_id) def batch_create_orders(self, order_count: int): """ 批量创建订单任务 :param order_count: :return: """ logger.info(f"📋 开始批量创建 {order_count} 个珠宝生产订单") for i in range(1, order_count + 1): order_id = f"珠宝订单-{i:03d}" self.create_order_task(order_id)

调用：

# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Worker Pool Pattern 工作池模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/21 8:31 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : WorkerPoolBll.py """ 项目主入口 职责单一：只负责启动 → 执行业务 → 等待结束 """ from WorkerPoolPattern.config.settings import WORKER_COUNT, TASK_QUEUE_MAXSIZE from WorkerPoolPattern.core.worker import WorkerPool from WorkerPoolPattern.service.production_service import JewelryProductionService from WorkerPoolPattern.utils.logger import logger class WorkerPoolBll(object): """ """ def demo(self): """ :return: """ logger.info("=" * 60) logger.info("珠宝企业级生产系统启动（Worker Pool 模式）") logger.info("=" * 60) pool = WorkerPool( worker_count=WORKER_COUNT, queue_maxsize=TASK_QUEUE_MAXSIZE ) pool.start() production_service = JewelryProductionService(pool) production_service.batch_create_orders(order_count=10) pool.wait_completion() logger.info("珠宝生产系统全部执行完成")

输出：