Python: Condition Variable Pattern

项目结构# 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 所有珠宝加工、质检完成)输出