当前位置：首页 > news >正文

Python: Condition Variable Pattern

news 2026/5/10 21:36:44

项目结构：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Condition Variable  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/10 21:12 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : settings.py
# 项目全局配置
TOTAL_BATCH = 3  # 每个工匠生产批次
ARTISAN_COUNT = 2  # 工匠数量
PRODUCE_DELAY = 0.5  # 生产耗时
INSPECT_DELAY = 0.3  # 质检耗时
START_WAIT_DELAY = 0.2  # 启动等待

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Condition Variable  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/10 21:13 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : jewelry.py
import threading
from dataclasses import dataclass
from typing import List@dataclass
class Jewelry:"""珠宝实体（数据封装）数据模型"""artisan_id: intbatch: intdef __str__(self):return f"钻石#{self.artisan_id}-{self.batch}"class JewelryInventory:"""共享库存（线程安全封装）"""def __init__(self):self._box: List[Jewelry] = []self._condition = threading.Condition()@propertydef condition(self):return self._condition@propertydef stock(self):return self._box.copy()@propertydef stock_count(self):return len(self._box)def is_empty(self):return len(self._box) == 0def put(self, jewelry: Jewelry):"""放入珠宝:param jewelry::return:"""self._box.append(jewelry)def take(self) -> Jewelry:"""取出珠宝:return:"""return self._box.pop(0)

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Condition Variable  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/10 21:14 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : producer.py
from ConditionVariablePattern.model.jewelry import JewelryInventory, Jewelryclass ProductionService:"""生产业务"""def __init__(self, inventory: JewelryInventory):""":param inventory:"""self.inventory = inventorydef produce(self, artisan_id: int, batch: int) -> Jewelry:"""生产并放入珠宝:param artisan_id::param batch::return:"""jewelry = Jewelry(artisan_id, batch)with self.inventory.condition:self.inventory.put(jewelry)print(f"✅ 工匠放入珠宝：{jewelry} | 当前库存：{self.inventory.stock_count}")self.inventory.condition.notify()return jewelry# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Condition Variable  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/10 21:14 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : consumer.py
from ConditionVariablePattern.model.jewelry import JewelryInventoryclass InspectionService:"""质检业务"""def __init__(self, inventory: JewelryInventory):""":param inventory:"""self.inventory = inventoryself.complete_count = 0def inspect(self, total_need: int) -> bool:"""等待珠宝并质检:param total_need::return:"""with self.inventory.condition:# 等待库存非空while self.inventory.is_empty():print("🔍 珠宝箱为空，质检员等待...")self.inventory.condition.wait()# 取出并质检jewelry = self.inventory.take()self.complete_count += 1print(f"🛡️  质检员取出珠宝：{jewelry} | 当前库存：{self.inventory.stock_count}")return self.complete_count >= total_need

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述： Condition Variable  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/10 21:20 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : artisan_thread.pyimport threading
import time
from ConditionVariablePattern.config.settings import TOTAL_BATCH, PRODUCE_DELAY
from ConditionVariablePattern.service.producer import ProductionServiceclass ArtisanThread(threading.Thread):"""工匠线程"""def __init__(self, artisan_id: int, service: ProductionService):""":param artisan_id::param service:"""super().__init__()self.artisan_id = artisan_idself.service = servicedef run(self):""":return:"""for batch in range(1, TOTAL_BATCH + 1):self.service.produce(self.artisan_id, batch)time.sleep(PRODUCE_DELAY)# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述： Condition Variable  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/10 21:22 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : inspector_thread.py
import threading
import time
from ConditionVariablePattern.config.settings import INSPECT_DELAY, ARTISAN_COUNT, TOTAL_BATCH
from ConditionVariablePattern.service.consumer import InspectionServiceclass InspectorThread(threading.Thread):"""质检员线程"""def __init__(self, service: InspectionService):""":param service:"""super().__init__()self.service = serviceself.total_need = ARTISAN_COUNT * TOTAL_BATCHdef run(self):""":return:"""while True:finished = self.service.inspect(self.total_need)if finished:breaktime.sleep(INSPECT_DELAY)

调用：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Condition Variable  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/10 21:19 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : ConditionVariableBll.py
'''
ConditionVariablePattern/
├── config/             # 配置层
│   └── settings.py     # 全局配置
├── model/              # 数据模型层
│   └── jewelry.py      # 珠宝、库存实体
├── service/            # 业务逻辑层
│   ├── producer.py     # 生产业务
│   └── consumer.py     # 质检业务
├── thread/             # 线程服务层
│   ├── artisan_thread.py  # 工匠线程
│   └── inspector_thread.py # 质检员线程
└── main.py             # 程序入口
'''
import time
from ConditionVariablePattern.config.settings import START_WAIT_DELAY
from ConditionVariablePattern.model.jewelry import JewelryInventory
from ConditionVariablePattern.service.producer import ProductionService
from ConditionVariablePattern.service.consumer import InspectionService
from ConditionVariablePattern.thread.artisan_thread import ArtisanThread
from ConditionVariablePattern.thread.inspector_thread import InspectorThreadclass ConditionVariableBll(object):""""""def demo(self):""":return:"""# 1. 初始化共享资源inventory = JewelryInventory()# 2. 初始化业务服务production_service = ProductionService(inventory)inspection_service = InspectionService(inventory)# 3. 启动质检员inspector = InspectorThread(inspection_service)inspector.start()time.sleep(START_WAIT_DELAY)# 4. 启动工匠artisans = [ArtisanThread(1, production_service),ArtisanThread(2, production_service)]for t in artisans:t.start()# 5. 等待所有线程完成for t in artisans:t.join()inspector.join()print("\n🎉 所有珠宝加工、质检完成！")

输出：