项目结构# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Push Pull 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/25 20:38 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : settings.py 全局配置端口、超时、珠宝行业固定常量、消息分隔符 # ZMQ 通信端口定义 PORT_RAW_MATERIAL 5555 PORT_PROCESS 5556 PORT_QUALITY 5557 PORT_SALE 5558 # 通信超时(ms) ZMQ_SOCKET_TIMEOUT 3000 # 珠宝行业基础枚举常量 JEWELRY_CATEGORY [ 钻石戒指, 黄金项链, 翡翠手镯, 铂金耳钉, 彩宝吊坠 ] QUALITY_GRADE [ S级(收藏), A级(精品), B级(常规), C级(特价) ] # 消息分隔符统一解析格式 MSG_SEP | KV_SEP : # 本地连接地址 LOCAL_TCP_ADDR tcp://localhost BIND_ADDR tcp://* # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Circuit-Breaker 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/28 10:39 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : exceptions.py 自定义异常体系 class ServiceException(Exception): 核心层自定义异常 业务服务基础异常 pass class ExternalServiceException(ServiceException): 外部服务调用异常 pass class CircuitBreakerOpenException(ServiceException): 断路器开启异常 pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Circuit-Breaker 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/28 10:44 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : base.py 断路器抽象基类依赖倒置原则面向接口编程 from abc import ABC, abstractmethod from enum import Enum class CircuitBreakerState(Enum): 核心层断路器抽象基类 断路器状态枚举 CLOSED CLOSED OPEN OPEN HALF_OPEN HALF_OPEN class BaseCircuitBreaker(ABC): 断路器抽象接口 abstractmethod def is_allowed(self) - bool: 是否允许请求通过 :return: pass abstractmethod def on_success(self)-None: 请求成功处理 :return: pass abstractmethod def on_failure(self)-None: 请求失败处理 :return: pass property abstractmethod def state(self)-CircuitBreakerState: 获取当前状态 :return: pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Circuit-Breaker 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/28 10:46 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : impl.py import time from .base import BaseCircuitBreaker, CircuitBreakerState from CircuitBreakerPattern.config.settings import CB_CONFIG class DefaultCircuitBreaker(BaseCircuitBreaker): 核心层断路器实现 默认断路器实现 def __init__( self, failure_thresholdCB_CONFIG.FAILURE_THRESHOLD, recovery_timeoutCB_CONFIG.RECOVERY_TIMEOUT, success_thresholdCB_CONFIG.SUCCESS_THRESHOLD ): self._failure_threshold failure_threshold self._recovery_timeout recovery_timeout self._success_threshold success_threshold self._state CircuitBreakerState.CLOSED self._failure_count 0 self._success_count 0 self._last_failure_time 0 property def state(self)-CircuitBreakerState: :return: return self._state def is_allowed(self) - bool: if self._state CircuitBreakerState.OPEN: now time.time() if now - self._last_failure_time self._recovery_timeout: self._state CircuitBreakerState.HALF_OPEN print( 断路器超时 → 进入半开状态) return True return False return True def on_success(self)- bool: :return: if self._state CircuitBreakerState.CLOSED: self._failure_count 0 elif self._state CircuitBreakerState.HALF_OPEN: self._success_count 1 if self._success_count self._success_threshold: print(✅ 半开状态连续成功 → 断路器关闭) self._reset() def on_failure(self)-None: :return: if self._state CircuitBreakerState.CLOSED: self._failure_count 1 print(f失败计数{self._failure_count}/{self._failure_threshold}) if self._failure_count self._failure_threshold: print( 连续失败达到阈值 → 断路器打开) self._open() elif self._state CircuitBreakerState.HALF_OPEN: print(❌ 半开状态请求失败 → 重新打开断路器) self._open() def _open(self)-None: 切换为开启状态 :return: self._state CircuitBreakerState.OPEN self._last_failure_time time.time() self._success_count 0 def _reset(self)- None: 重置为关闭状态 :return: self._state CircuitBreakerState.CLOSED self._failure_count 0 self._success_count 0 self._last_failure_time 0 # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Circuit-Breaker 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/28 10:49 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : decorators.py 通用装饰器 import functools from CircuitBreakerPattern.core.circuitbreaker.base import BaseCircuitBreaker, CircuitBreakerState from CircuitBreakerPattern.core.exceptions import ExternalServiceException from CircuitBreakerPattern.config.settings import JEWELRY_CONFIG def circuit_breaker(breaker: BaseCircuitBreaker): 工具层断路器装饰器 断路器装饰器通用、可复用 :param breaker: :return: def decorator(func): :param func: :return: functools.wraps(func) def wrapper(*args, **kwargs): :param args: :param kwargs: :return: if not breaker.is_allowed(): print(f⚠️ 断路器已打开 | 当前状态{breaker.state.value} → 使用兜底金价) return { status: fallback, message: 断路器已开启使用兜底数据, gold_price: JEWELRY_CONFIG.FALLBACK_GOLD_PRICE } try: result func(*args, **kwargs) breaker.on_success() print(f✅ 请求成功 | 断路器状态{breaker.state.value}) return result except ExternalServiceException as e: breaker.on_failure() print(f❌ 请求失败 | 断路器状态{breaker.state.value} | 异常{e}) return { status: error, message: 第三方金价服务调用失败, gold_price: JEWELRY_CONFIG.FALLBACK_GOLD_PRICE } return wrapper return decorator# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Circuit-Breaker 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/28 10:53 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : inventory_service.py 珠宝库存服务内部稳定服务 from dataclasses import dataclass dataclass class StockResult: sku: str stock: int status: str success class JewelryInventoryService: 业务服务层库存服务 库存服务单一职责仅处理库存校验 def check_diamond_stock(self, sku: str) - StockResult: 钻石库存校验 :param sku: :return: return StockResult(skusku, stock15) def check_gold_stock(self, weight: float) - dict: 黄金库存校验 :param weight: :return: return { weight: weight, stock: 充足, status: success } # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Circuit-Breaker 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/28 10:54 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : price_service.py 第三方金价服务外部依赖带断路器保护 import random import time from CircuitBreakerPattern.core.circuitbreaker.impl import DefaultCircuitBreaker from CircuitBreakerPattern.core.exceptions import ExternalServiceException from CircuitBreakerPattern.utils.decorators import circuit_breaker from CircuitBreakerPattern.config.settings import JEWELRY_CONFIG # 全局单例断路器生产可替换为Spring/DI容器管理 _gold_price_circuit_breaker DefaultCircuitBreaker() class GoldPriceService: 业务服务层金价服务 第三方实时金价服务 circuit_breaker(_gold_price_circuit_breaker) def get_realtime_price(self) - dict: 获取实时金价带熔断保护 :return: # 模拟第三方接口故障 if random.random() JEWELRY_CONFIG.FAILURE_RATE: raise ExternalServiceException(第三方金价接口超时/500错误) # 正常返回 price round(random.uniform(630, 650), 2) return { status: success, gold_price: price, update_time: time.strftime(%Y-%m-%d %H:%M:%S) } # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Circuit-Breaker 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/28 10:56 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : order_service.py 珠宝订单核心服务编排业务流程 import time from CircuitBreakerPattern.service.inventory_service import JewelryInventoryService from CircuitBreakerPattern.service.price_service import GoldPriceService class JewelryOrderService: 业务服务层订单服务 订单服务 依赖注入库存服务、金价服务不直接new对象 def __init__( self, inventory_service: JewelryInventoryService, price_service: GoldPriceService ): # 依赖注入松耦合、可测试、可替换 self.inventory inventory_service self.price price_service def create_gold_order(self, user_id: str, weight: float, sku: str) - dict: 创建黄金珠宝订单标准业务编排 :param user_id: :param weight: :param sku: :return: print(\n * 50) print(f[订单服务] 开始创建订单 | 用户{user_id} | 克重{weight}g) # 1. 校验库存 stock self.inventory.check_gold_stock(weight) print(f[库存服务] {stock}) # 2. 获取金价断路器保护 price_info self.price.get_realtime_price() print(f[金价服务] {price_info}) # 3. 计算金额 total round(weight * price_info[gold_price], 2) # 4. 生成订单 order { order_id: fORD{int(time.time())}, user_id: user_id, sku: sku, weight: weight, gold_price: price_info[gold_price], total_amount: total, order_status: PENDING_PAYMENT } print(f[订单服务] 订单创建成功{order}) return order调用# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Circuit-Breaker 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/28 10:57 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : CircuitBreakerBll.py 系统入口依赖注入 启动测试 import time from CircuitBreakerPattern.service.inventory_service import JewelryInventoryService from CircuitBreakerPattern.service.price_service import GoldPriceService from CircuitBreakerPattern.service.order_service import JewelryOrderService class CircuitBreakerBll(object): def init_dependencies(self): 依赖注入容器模拟Spring/DI 解耦所有服务通过注入创建而非硬编码 inventory_service JewelryInventoryService() price_service GoldPriceService() order_service JewelryOrderService(inventory_service, price_service) return order_service def demo(self): :return: print( 企业级珠宝订单系统断路器模式保护\n) order_service self.init_dependencies() # 模拟10次连续下单 for i in range(10): print(f\n 第 {i 1} 次下单请求 ) order order_service.create_gold_order( user_idfUSER_{i 1}, weight10.0, skuGOLD_999_001 ) print(f✅ 最终订单结果{order}\n) time.sleep(1)输出
