项目结构：
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 20:09
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : exceptions.py
class BusinessException(Exception):
"""
业务基础异常
"""
pass
class ServiceTimeoutException(BusinessException):
"""
服务超时异常（截止日期模式触发）
"""
pass
class ServiceInvokeException(BusinessException):
"""
服务调用失败异常
"""
pass
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 20:10
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : logger.py
import logging
import sys
def get_logger(name: str) -> logging.Logger:
"""
企业日志
:param name:
:return:
"""
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
formatter = logging.Formatter(
"%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
logger.propagate = False
return logger
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 20:10
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : deadline.py
import threading
from typing import Callable, Any, Generic, TypeVar
from DeadlinePattern.common.exceptions import ServiceTimeoutException, ServiceInvokeException
from DeadlinePattern.common.logger import get_logger
logger = get_logger("DeadlinePattern")
T = TypeVar("T")
class DeadlineExecutor(Generic[T]):
"""
截止日期模式（企业级核心）
企业级截止日期模式执行器
职责：统一处理超时任务，超时终止等待，保证系统稳定性
"""
def __init__(self, timeout_seconds: float):
self.timeout = timeout_seconds
self.result: T | None = None
self.error: Exception | None = None
def _run(self, func: Callable[[], T]):
try:
self.result = func()
except Exception as e:
self.error = e
def execute(self, func: Callable[[], T]) -> T:
thread = threading.Thread(target=self._run, args=(func,))
thread.start()
thread.join(self.timeout)
if thread.is_alive():
logger.error(f"任务超时，已终止等待，超时={self.timeout}s")
raise ServiceTimeoutException("超时：响应概率极低，主动终止等待")
if self.error:
raise ServiceInvokeException(f"服务执行异常：{str(self.error)}")
logger.info("任务执行完成")
return self.result
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 20:08
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : config.py
# 全局超时配置（截止日期模式超时时间）
GEM_IDENTIFY_TIMEOUT_SECONDS = 3
# 业务配置
ORDER_ID_PREFIX = "JEWELRY_ORDER"
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 20:13
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : models.py
from dataclasses import dataclass
from typing import Optional
@dataclass
class GemIdentifyResult:
"""
宝石鉴定结果实体
"""
gem_type: str
carat: float
color: str
clarity: str
identify_no: str
@dataclass
class JewelryOrder:
"""
珠宝订单实体
"""
order_id: str
identify_result: Optional[GemIdentifyResult] = None
material_approved: bool = False
produce_finished: bool = False
quality_passed: bool = False
delivered: bool = False
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 21:00
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : gem_identify_service.py
import random
import time
from DeadlinePattern.domain.models import GemIdentifyResult
from DeadlinePattern.common.logger import get_logger
logger = get_logger("GemIdentifyService")
class GemIdentifyService:
"""
第三方宝石鉴定服务（外部不稳定依赖）
"""
def identify(self) -> GemIdentifyResult:
"""
:return:
"""
logger.info("开始调用第三方宝石鉴定服务...")
# 随机控制响应速度：50%快速，50%超长耗时触发截止期限
if random.random() > 0.5:
delay = 1
else:
delay = 10
time.sleep(delay)
res = GemIdentifyResult(
gem_type="钻石",
carat=1.0,
color="D",
clarity="VVS1",
identify_no=f"GEM_{random.randint(100000,999999)}"
)
logger.info(f"鉴定完成：{res}")
return res
'''
class GemIdentifyService:
"""
第三方宝石鉴定服务（外部不稳定依赖）
"""
def identify(self) -> GemIdentifyResult:
"""
:return:
"""
logger.info("开始调用第三方宝石鉴定服务...")
# 模拟不稳定：50% 1秒返回，50% 10秒返回
delay = 1 if random.random() > 0.5 else 10
time.sleep(delay)
result = GemIdentifyResult(
gem_type="钻石",
carat=1.0,
color="D",
clarity="VVS1",
identify_no=f"GEM_{random.randint(100000, 999999)}"
)
logger.info(f"鉴定完成：{result}")
return result
'''
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 21:02
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : material_service.py
from DeadlinePattern.common.logger import get_logger
logger = get_logger("MaterialService")
class MaterialService:
"""
"""
def approve(self) -> bool:
logger.info("原料采购审核中...")
return True
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 21:03
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : produce_service.py
from DeadlinePattern.common.logger import get_logger
logger = get_logger("ProduceService")
class ProduceService:
"""
"""
def produce(self) -> bool:
"""
:return:
"""
logger.info("工厂加工生产完成")
return True
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 21:04
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : quality_service.py
from DeadlinePattern.common.logger import get_logger
logger = get_logger("QualityService")
class QualityService:
"""
"""
def check(self) -> bool:
"""
:return:
"""
logger.info("成品质检合格")
return True
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 21:05
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : deliver_service.py
from DeadlinePattern.common.logger import get_logger
logger = get_logger("DeliverService")
class DeliverService:
"""
"""
def deliver(self) -> bool:
"""
:return:
"""
logger.info("订单已交付")
return True
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Deadline 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/6/29 20:08
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : workflow_manager.py
from DeadlinePattern.domain.models import JewelryOrder, GemIdentifyResult
from DeadlinePattern.service.external.gem_identify_service import GemIdentifyService
from DeadlinePattern.service.internal.material_service import MaterialService
from DeadlinePattern.service.internal.produce_service import ProduceService
from DeadlinePattern.service.internal.quality_service import QualityService
from DeadlinePattern.service.internal.deliver_service import DeliverService
from DeadlinePattern.common.deadline import DeadlineExecutor
from DeadlinePattern.common.exceptions import ServiceTimeoutException
from DeadlinePattern.common.logger import get_logger
from DeadlinePattern.config.config import GEM_IDENTIFY_TIMEOUT_SECONDS
logger = get_logger("WorkflowManager")
class JewelryWorkflowManager:
"""
流程编排器（核心）
珠宝全流程管理器
职责：统一编排所有服务，不实现业务逻辑
"""
def __init__(self):
self.identify = GemIdentifyService()
self.material = MaterialService()
self.produce = ProduceService()
self.quality = QualityService()
self.deliver = DeliverService()
#
def run(self, order: JewelryOrder):
"""
:param order:
:return:
"""
logger.info(f"开始执行订单全流程：{order.order_id}")
try:
exec = DeadlineExecutor[GemIdentifyResult](GEM_IDENTIFY_TIMEOUT_SECONDS)
order.identify_result = exec.execute(self.identify.identify)
except ServiceTimeoutException:
logger.error("鉴定超时，启动人工复核兜底流程")
order.material_approved = self.material.approve()
order.produce_finished = self.produce.produce()
order.quality_passed = self.quality.check()
order.delivered = self.deliver.deliver()
logger.info(f"订单全流程执行完成：{order.order_id}")
return order
'''
class JewelryWorkflowManager:
"""
流程编排器（核心）
珠宝全流程管理器
职责：统一编排所有服务，不实现业务逻辑
"""
def __init__(self):
# 依赖注入（可扩展、可测试）
self.identify_service = GemIdentifyService()
self.material_service = MaterialService()
self.produce_service = ProduceService()
self.quality_service = QualityService()
self.deliver_service = DeliverService()
def execute_full_workflow(self, order: JewelryOrder) -> JewelryOrder:
"""
:param order:
:return:
"""
logger.info(f"开始执行订单全流程：{order.order_id}")
# ===================== 截止日期模式：鉴定服务 =====================
try:
executor = DeadlineExecutor[GemIdentifyResult](GEM_IDENTIFY_TIMEOUT_SECONDS)
identify_result = executor.execute(self.identify_service.identify)
order.identify_result = identify_result
except ServiceTimeoutException:
logger.error("鉴定超时，启动人工复核兜底流程")
# 企业级兜底逻辑，不阻塞主流程
# ===================== 内部稳定流程 =====================
order.material_approved = self.material_service.approve()
order.produce_finished = self.produce_service.produce()
order.quality_passed = self.quality_service.check()
order.delivered = self.deliver_service.deliver()
logger.info(f"订单全流程执行完成：{order.order_id}")
return order
'''
调用：
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# 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/6/29 21:07
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : DeadlineBll.py
import time
import random
from DeadlinePattern.domain.models import JewelryOrder
from DeadlinePattern.core.workflow_manager import JewelryWorkflowManager
from DeadlinePattern.common.logger import get_logger
from DeadlinePattern.config.config import ORDER_ID_PREFIX
logger = get_logger("DeadlineBll")
class DeadlineBll(object):
"""
"""
def demo(self):
"""
:return:
"""
print("===== 展示【Deadline Pattern（截止期限模式）】示例 =====")
logger.info("=== 珠宝定制系统启动 ===")
order = JewelryOrder(f"{ORDER_ID_PREFIX}_{random.randint(1000, 9999)}")
logger.info(f"订单创建：{order}")
# 这里现在一定正常
manager = JewelryWorkflowManager()
final = manager.run(order)
logger.info("=" * 60)
logger.info(f"最终订单状态：{final}")
logger.info("=== 系统运行完成 ===")
'''
logger.info("=== 珠宝定制系统启动 ===")
# 1. 创建订单
order = JewelryOrder(
order_id=f"{ORDER_ID_PREFIX}_{random.randint(1000, 9999)}"
)
logger.info(f"订单创建：{order}")
# 2. 执行全流程
manager = JewelryWorkflowManager()
final_order = manager.execute_full_workflow(order)
# 3. 最终结果
logger.info("=" * 60)
logger.info(f"最终订单状态：{final_order}")
logger.info("=== 系统运行完成 ===")
time.sleep(1)
'''
输出：
哲学管理(学)人生, 文学艺术生活, 自动(计算机学)物理(学)工作, 生物(学)化学逆境, 历史(学)测绘(学)时间, 经济(学)数学金钱(理财), 心理(学)医学情绪, 诗词美容情感, 美学建筑(学)家园, 解构建构(分析)整合学习, 智商情商(IQ、EQ)运筹(学)生存.---Geovin Du(涂聚文）
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。