项目结构：
# 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🎉 所有珠宝加工、质检完成！")
输出：
哲学管理(学)人生, 文学艺术生活, 自动(计算机学)物理(学)工作, 生物(学)化学逆境, 历史(学)测绘(学)时间, 经济(学)数学金钱(理财), 心理(学)医学情绪, 诗词美容情感, 美学建筑(学)家园, 解构建构(分析)整合学习, 智商情商(IQ、EQ)运筹(学)生存.---Geovin Du(涂聚文）
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。