python: Semaphore Pattern

geovindu

发布于 2026-06-18 17:12:21

1130

项目结构：

# encoding: utf-8
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：企业级信号量模式 Semaphore 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/5/19 05:18
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : settings.py
 
"""
全局配置中心
"""
 
# ===================== 资源并发配额（信号量值） =====================
# 独占资源 = 1，共享资源 = 最大并发数
RESOURCE_QUOTA = {
    "vip_workshop": 1,       # VIP定制工坊
    "smelter": 2,            # 黄金熔炉
    "cutter": 3,             # 钻石切割台
    "inlay_machine": 1,      # 镶嵌设备
    "inspect_station": 2     # 质检工位
}
 
# ===================== 工序耗时配置（秒） =====================
PROCESS_TIME = {
    "vip_custom": 3,
    "smelt": 1,
    "cut": 1,
    "inlay": 2,
    "inspect": 1
}
 
# ===================== 订单总配置 =====================
TOTAL_ORDERS = 14
ORDER_ASSIGN = {
    "vip": [14, 13],
    "smelt": [1, 2, 3],
    "cut": [4, 5, 6, 7],
    "inlay": [8, 9],
    "inspect": [10, 11, 12]
}
 
 
 
# encoding: utf-8
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：企业级信号量模式 Semaphore 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/5/19 05:18
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : logger.py
 
 
import threading
from datetime import datetime
 
 
class LoggerFactory:
    """
    日志工具：线程安全、统一格式
    """
    _lock = threading.Lock()
 
    @staticmethod
    def info(msg: str):
        """
        INFO级别日志输出
        :param msg:
        :return:
        """
        current_time = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
        with LoggerFactory._lock:
            print(f"{current_time} [INFO] {msg}")

# encoding: utf-8
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：企业级信号量模式 Semaphore 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/5/19 05:20
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : resources.py
 
 
"""
共享资源管理器：信号量统一创建、管理
"""
import threading
from SemaphorePattern.config.settings import RESOURCE_QUOTA
 
 
class ResourceManager:
    """
    单例资源管理器，保证全局唯一资源实例
    """
    _instance = None
    _lock = threading.Lock()
 
    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = super().__new__(cls)
                    cls._instance._init_resources()
        return cls._instance
 
    def _init_resources(self):
        """
        初始化所有生产资源信号量
        :return:
        """
        self.vip_workshop = threading.Semaphore(RESOURCE_QUOTA["vip_workshop"])
        self.smelter = threading.Semaphore(RESOURCE_QUOTA["smelter"])
        self.cutter = threading.Semaphore(RESOURCE_QUOTA["cutter"])
        self.inlay_machine = threading.Semaphore(RESOURCE_QUOTA["inlay_machine"])
        self.inspect_station = threading.Semaphore(RESOURCE_QUOTA["inspect_station"])
 
 
# 全局唯一资源实例
resource = ResourceManager()
 
 
 
 
# encoding: utf-8
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：企业级信号量模式 Semaphore 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/5/19 05:21
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : process.py
 
"""
生产工序服务：业务逻辑实现
"""
import time
from SemaphorePattern.config.settings import PROCESS_TIME
from SemaphorePattern.core.resources import resource
from SemaphorePattern.utils.logger import LoggerFactory
 
 
class JewelryProcessService:
    """
    珠宝生产工序服务类
    """
 
    @staticmethod
    def vip_custom(order_id: int):
        """
        VIP定制工序
        :param order_id:
        :return:
        """
        with resource.vip_workshop:
            LoggerFactory.info(f"[VIP定制] 订单{order_id} 开始专属工坊制作")
            time.sleep(PROCESS_TIME["vip_custom"])
            LoggerFactory.info(f"[VIP定制] 订单{order_id} 定制完成 ✨")
 
    @staticmethod
    def smelt(order_id: int):
        """
        熔炼工序
        :param order_id:
        :return:
        """
        with resource.smelter:
            LoggerFactory.info(f"[熔炼] 订单{order_id} 开始黄金熔炼 | 占用熔炉资源")
            time.sleep(PROCESS_TIME["smelt"])
            LoggerFactory.info(f"[熔炼] 订单{order_id} 熔炼完成 ✅")
 
    @staticmethod
    def cut(order_id: int):
        """
        切割工序
        :param order_id:
        :return:
        """
        with resource.cutter:
            LoggerFactory.info(f"[切割] 订单{order_id} 开始钻石切割 | 占用切割台")
            time.sleep(PROCESS_TIME["cut"])
            LoggerFactory.info(f"[切割] 订单{order_id} 切割完成 ✅")
 
    @staticmethod
    def inlay(order_id: int):
        """
        镶嵌工序
        :param order_id:
        :return:
        """
        with resource.inlay_machine:
            LoggerFactory.info(f"[镶嵌] 订单{order_id} 开始精密镶嵌 | 独占设备")
            time.sleep(PROCESS_TIME["inlay"])
            LoggerFactory.info(f"[镶嵌] 订单{order_id} 镶嵌完成 ✅")
 
    @staticmethod
    def inspect(order_id: int):
        """
        质检工序
        :param order_id:
        :return:
        """
        with resource.inspect_station:
            LoggerFactory.info(f"[质检] 订单{order_id} 开始品质检测")
            time.sleep(PROCESS_TIME["inspect"])
            LoggerFactory.info(f"[质检] 订单{order_id} 检测通过 ✅")
 
 
 
# encoding: utf-8
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：企业级信号量模式 Semaphore 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/5/19 05:22
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : order_task.py
 
"""
订单任务管理：任务创建、线程管理
"""
import threading
from typing import List
from SemaphorePattern.config.settings import ORDER_ASSIGN
from SemaphorePattern.service.process import JewelryProcessService
 
 
class OrderTask:
    """
    订单任务生成器
    """
 
    @staticmethod
    def create_all_tasks() -> List[threading.Thread]:
        """
        创建所有生产任务线程
        :return:
        """
        threads = []
 
        # VIP定制任务
        for oid in ORDER_ASSIGN["vip"]:
            threads.append(threading.Thread(target=JewelryProcessService.vip_custom, args=(oid,)))
 
        # 熔炼任务
        for oid in ORDER_ASSIGN["smelt"]:
            threads.append(threading.Thread(target=JewelryProcessService.smelt, args=(oid,)))
 
        # 切割任务
        for oid in ORDER_ASSIGN["cut"]:
            threads.append(threading.Thread(target=JewelryProcessService.cut, args=(oid,)))
 
        # 镶嵌任务
        for oid in ORDER_ASSIGN["inlay"]:
            threads.append(threading.Thread(target=JewelryProcessService.inlay, args=(oid,)))
 
        # 质检任务
        for oid in ORDER_ASSIGN["inspect"]:
            threads.append(threading.Thread(target=JewelryProcessService.inspect, args=(oid,)))
 
        return threads
 
 
 
# encoding: utf-8
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：企业级信号量模式 Semaphore 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/5/19 05:24
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : engine.py
 
"""
珠宝工厂生产引擎：核心调度
"""
from typing import List
import threading
from SemaphorePattern.config.settings import TOTAL_ORDERS
from SemaphorePattern.task.order_task import OrderTask
from SemaphorePattern.utils.logger import LoggerFactory
 
 
class ProductionEngine:
    """
    生产调度引擎"
    """
 
 
    def __init__(self):
        self.task_threads: List[threading.Thread] = OrderTask.create_all_tasks()
 
    def start_production(self):
        """
        启动全流程生产
        :return:
        """
        LoggerFactory.info(f"========== 珠宝工厂开始生产，总订单数：{TOTAL_ORDERS} ==========")
 
        # 启动所有任务
        for t in self.task_threads:
            t.start()
 
        # 等待所有任务完成
        for t in self.task_threads:
            t.join()
 
        LoggerFactory.info("========== 所有订单生产完成 ==========")

调用：

# encoding: utf-8
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：业级信号量模式 Semaphore 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/5/19 05:25
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : SemaphoreBll.py
"""
Semaphore/
├── config/                # 配置层（全局配置、资源配额）
│   └── settings.py
├── core/                  # 核心层（信号量资源、工厂引擎）
│   ├── resources.py       # 共享资源管理
│   └── engine.py          # 生产引擎
├── service/               # 业务服务层（工序逻辑）
│   └── process.py
├── task/                  # 任务层（订单任务定义）
│   └── order_task.py
├── utils/                 # 工具层（日志、线程工具）
│   └── logger.py
└── main.py                # 项目入口
项目主入口
"""
from SemaphorePattern.core.engine import ProductionEngine
 
class SemaphoreBll(object):
    """
 
    """
    def demo(self):
        """
 
        :return:
        """
        # 启动工厂生产引擎
        factory = ProductionEngine()
        factory.start_production()

输出：