项目结构：
# 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: int batch: int def __str__(self): return f"钻石#{self.artisan_id}-{self.batch}" class JewelryInventory: """ 共享库存（线程安全封装） """ def __init__(self): self._box: List[Jewelry] = [] self._condition = threading.Condition() @property def condition(self): return self._condition @property def stock(self): return self._box.copy() @property def stock_count(self): return len(self._box) def is_empty(self): return len(self._box) == 0 def 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, Jewelry class ProductionService: """ 生产业务 """ def __init__(self, inventory: JewelryInventory): """ :param inventory: """ self.inventory = inventory def 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 JewelryInventory class InspectionService: """ 质检业务 """ def __init__(self, inventory: JewelryInventory): """ :param inventory: """ self.inventory = inventory self.complete_count = 0 def 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 += 1 print(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.py import threading import time from ConditionVariablePattern.config.settings import TOTAL_BATCH, PRODUCE_DELAY from ConditionVariablePattern.service.producer import ProductionService class ArtisanThread(threading.Thread): """ 工匠线程 """ def __init__(self, artisan_id: int, service: ProductionService): """ :param artisan_id: :param service: """ super().__init__() self.artisan_id = artisan_id self.service = service def 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 InspectionService class InspectorThread(threading.Thread): """ 质检员线程 """ def __init__(self, service: InspectionService): """ :param service: """ super().__init__() self.service = service self.total_need = ARTISAN_COUNT * TOTAL_BATCH def run(self): """ :return: """ while True: finished = self.service.inspect(self.total_need) if finished: break time.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 InspectorThread class 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🎉 所有珠宝加工、质检完成！")输出：
