python: Worker Pool Pattern

geovi

311人浏览 · 2026-06-24 20:04:07

geovi · 2026-06-24 20:04:07 发布

项目经构：

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("珠宝生产系统全部执行完成")

输出：