python: Monitor Pattern

博客园 - ®Geovin Du Dream Park™

go: Functional Options Pattern go:Timing Functions Pattern python: Timing Functions Pattern go: Steady-State Pattern python: Steady-State Pattern go: Handshaking Pattern python: Handshaking Pattern go: Fail-Fast Pattern python: Fail-Fast Pattern I go: Deadline Pattern python: Deadline Pattern go: Circuit-Breaker Pattern python: Circuit-Breaker Pattern go: Bulkheads Pattern python: Bulkheads Pattern go: Push & Pull Pattern python: Push & Pull Pattern python: Publish/Subscribe Pattern II go: Publish/Subscribe Pattern python: Publish/Subscribe Pattern go: Futures & Promises Pattern python: Futures & Promises Pattern go: Worker Pool Pattern go：Pipeline Pattern python: Worker Pool Pattern python: Pipeline Pattern go: Fan-Out Pattern python: Fan-Out Pattern go: Fan-In Pattern python: Fan-In Pattern Fan-In

®Geovin Du Dream Park™ · 2026-05-17 · via 博客园 - ®Geovin Du Dream Park™

项目结构：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：监控模式 Monitor 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/17 20:30 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : settings.py

# 系统全局配置
class SystemConfig(object):
    """
    生产任务配置
    """
    TOTAL_PRODUCTION_TASKS = 10  # 总生产数量
    WORKER1_TASKS = 5            # 工匠1 生产数量
    WORKER2_TASKS = 5            # 工匠2 生产数量

    # 珠宝类型配置
    WORKER1_JEWELRY = "钻戒"
    WORKER1_MATERIAL = "钻石+铂金"
    WORKER2_JEWELRY = "黄金项链"
    WORKER2_MATERIAL = "足金999"

    # 模拟耗时配置（秒）
    PRODUCE_MIN_DELAY = 0
    PRODUCE_MAX_DELAY = 0.5
    CHECK_MIN_DELAY = 0.3
    CHECK_MAX_DELAY = 0.8

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：监控模式 Monitor 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/17 20:31 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : logger.py

from datetime import datetime

class SystemLogger:
    """
    统一日志工具类
    """
    @staticmethod
    def log(message: str) -> None:
        """
        带时间戳的标准日志输出
        :param message:
        :return:
        """
        timestamp = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
        print(f"{timestamp} {message}")

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：监控模式 Monitor 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/17 20:30 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : jewelry.py



class Jewelry:
    """
    珠宝实体类
    """
    def __init__(self, jewelry_id: int, name: str, material: str):
        """

        :param jewelry_id:
        :param name:
        :param material:
        """
        self.id = jewelry_id      # 珠宝ID
        self.name = name          # 名称
        self.material = material  # 材质

    def __repr__(self):
        """

        :return:
        """
        return f"ID={self.id} | {self.name} | {self.material}"

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：监控模式 Monitor 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/17 20:33 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : monitor.py

import threading
from MonitorPattern.entity.jewelry import Jewelry
from MonitorPattern.utils.logger import SystemLogger
from MonitorPattern.config.settings import SystemConfig

class JewelryMonitor:
    """
    监控器：线程安全的生产-消费缓冲区（Monitor 模式标准实现）
    """
    _instance = None  # 单例模式
    _lock = threading.Lock()

    def __new__(cls):
        """
        单例：全局唯一监控器
        """
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance.init()
            return cls._instance

    def init(self):
        """
        线程同步
        :return:
        """
        self.mutex = threading.Lock()
        self.cond = threading.Condition(self.mutex)

        # 共享资源
        self.jewelry_queue: list[Jewelry] = []
        self.jewelry_id = 1

        # 任务计数
        self.produced_count = 0
        self.checked_count = 0
        self.total_tasks = SystemConfig.TOTAL_PRODUCTION_TASKS

    def produce(self, worker_name: str, jewelry_name: str, material: str) -> bool:
        """
        生产：原子操作，线程安全"
        :param worker_name: 
        :param jewelry_name: 
        :param material: 
        :return: 
        """
        with self.mutex:
            # 创建珠宝实体
            jewelry = Jewelry(self.jewelry_id, jewelry_name, material)
            self.jewelry_queue.append(jewelry)

            # 日志
            SystemLogger.log(f"[{worker_name}] 生产完成: {jewelry_name} | {material}")

            # 计数
            self.jewelry_id += 1
            self.produced_count += 1

            # 唤醒消费者
            self.cond.notify()

            # 返回是否生产完成
            return self.produced_count == self.total_tasks

    def consume(self, inspector_name: str) -> bool:
        """
        消费：原子操作，线程安全
        :param inspector_name:
        :return:
        """
        with self.mutex:
            # 等待条件
            while not self.jewelry_queue and self.produced_count < self.total_tasks:
                self.cond.wait()

            # 消费
            if self.jewelry_queue:
                jewelry = self.jewelry_queue.pop(0)
                SystemLogger.log(f"[{inspector_name}] 质检完成: {jewelry}")
                self.checked_count += 1

            # 返回是否全部质检完成
            return self.checked_count == self.total_tasks



# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：监控模式 Monitor 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/17 20:35 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : producer.py
import time
import random
from MonitorPattern.core.monitor import JewelryMonitor
from MonitorPattern.utils.logger import SystemLogger
from MonitorPattern.config.settings import SystemConfig

class JewelryProducer:
    """
    生产者
    """
    def __init__(self, name: str, jewelry_name: str, material: str, task_num: int):
        """

        :param name:
        :param jewelry_name:
        :param material:
        :param task_num:
        """
        self.name = name
        self.jewelry_name = jewelry_name
        self.material = material
        self.task_num = task_num
        self.monitor = JewelryMonitor()

    def run(self):
        """
        执行生产任务
        :return:
        """
        for _ in range(self.task_num):
            # 模拟生产耗时
            time.sleep(random.uniform(
                SystemConfig.PRODUCE_MIN_DELAY,
                SystemConfig.PRODUCE_MAX_DELAY
            ))
            # 生产
            finished = self.monitor.produce(
                self.name,
                self.jewelry_name,
                self.material
            )
            # 完成标记
            if finished:
                SystemLogger.log(f"[{self.name}] ✅ 全部生产任务完成")


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：监控模式 Monitor 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/17 20:36 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : consumer.py

import time
import random
from MonitorPattern.core.monitor import JewelryMonitor
from MonitorPattern.utils.logger import SystemLogger
from MonitorPattern.config.settings import SystemConfig

class JewelryConsumer:
    """
    消费者：
    """
    def __init__(self, name: str):
        """

        :param name:
        """
        self.name = name
        self.monitor = JewelryMonitor()

    def run(self):
        """
        执行质检任务
        :return:
        """
        while True:
            # 模拟质检耗时
            time.sleep(random.uniform(
                SystemConfig.CHECK_MIN_DELAY,
                SystemConfig.CHECK_MAX_DELAY
            ))
            # 质检
            finished = self.monitor.consume(self.name)
            if finished:
                SystemLogger.log(f"[{self.name}] ✅ 全部质检任务完成")
                break

调用：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：监控模式 Monitor 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/17 20:38 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : MonitorBll.py
"""
Monitor/
├── config/                # 配置层
│   └── settings.py        # 全局配置
├── core/                  # 核心业务层
│   ├── monitor.py         # Monitor 模式实现（线程安全监控器）
│   ├── producer.py        # 生产者（工匠）
│   └── consumer.py        # 消费者（质检员）
├── entity/                # 实体层
│   └── jewelry.py         # 珠宝实体类
├── utils/                 # 工具层
│   └── logger.py          # 日志工具（统一时间戳）
└── main.py                # 系统入口
"""
import threading
from MonitorPattern.utils.logger import SystemLogger
from MonitorPattern.core.producer import JewelryProducer
from MonitorPattern.core.consumer import JewelryConsumer
from MonitorPattern.config.settings import SystemConfig


class MonitorBll(object):
    """

    """
    def demo(self):
        """

        :return:
        """
        # 日志：系统启动
        SystemLogger.log("=== 珠宝生产-质检系统 启动 ===")

        # ========== 初始化角色 ==========
        # 工匠1
        worker1 = JewelryProducer(
            name="工匠1",
            jewelry_name=SystemConfig.WORKER1_JEWELRY,
            material=SystemConfig.WORKER1_MATERIAL,
            task_num=SystemConfig.WORKER1_TASKS
        )

        # 工匠2
        worker2 = JewelryProducer(
            name="工匠2",
            jewelry_name=SystemConfig.WORKER2_JEWELRY,
            material=SystemConfig.WORKER2_MATERIAL,
            task_num=SystemConfig.WORKER2_TASKS
        )

        # 质检员
        inspector = JewelryConsumer(name="质检员")

        # ========== 创建线程 ==========
        t1 = threading.Thread(target=worker1.run, name="工匠1")
        t2 = threading.Thread(target=worker2.run, name="工匠2")
        t3 = threading.Thread(target=inspector.run, name="质检员")

        # ========== 启动 ==========
        t1.start()
        t2.start()
        t3.start()

        # ========== 等待结束 ==========
        t1.join()
        t2.join()
        t3.join()

        # 系统完成
        SystemLogger.log("=== 🎉 珠宝生产-质检全流程完成 ===")

输出：

哲学管理(学)人生, 文学艺术生活, 自动(计算机学)物理(学)工作, 生物(学)化学逆境, 历史(学)测绘(学)时间, 经济(学)数学金钱(理财), 心理(学)医学情绪, 诗词美容情感, 美学建筑(学)家园, 解构建构(分析)整合学习, 智商情商(IQ、EQ)运筹(学)生存.---Geovin Du(涂聚文）

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

博客园 - ®Geovin Du Dream Park™