项目结构：
# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:49 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : constants.py from enum import Enum class BusinessEventType(Enum): """ 珠宝行业全部业务事件类型枚举，统一管理 """ RAW_MATERIAL = "raw_material" # 原料采购验收 PROCESS_CHECK = "process_check" # 加工质检 STORE_SALE = "sale" # 门店销售结算 AFTER_SALE_REPAIR = "repair" # 售后维修 INVENTORY_CHECK = "inventory" # 库存盘点 # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:49 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : exceptions.py class HandlerNotFoundError(Exception): """ 事件无对应处理器异常 """ pass class InvalidEventDataError(Exception): """ 事件请求数据非法异常 """ pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:50 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : logger.py import logging import sys def get_logger(name: str = "jewelry_reactor") -> logging.Logger: """ :param name: :return: """ logger = logging.getLogger(name) logger.setLevel(logging.INFO) if not logger.handlers: fmt = logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S") sh = logging.StreamHandler(sys.stdout) sh.setFormatter(fmt) logger.addHandler(sh) return logger # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:53 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : base_handler.py from abc import ABC, abstractmethod from ReactorPattern.event import BusinessEvent class BaseHandler(ABC): """ 所有业务处理器抽象基类，统一规范接口 """ @abstractmethod def handle(self, event: BusinessEvent) -> None: """ 处理对应业务事件 :param event: :return: """ pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:54 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : material_handler.py from .base_handler import BaseHandler from ReactorPattern.event import BusinessEvent from ReactorPattern.common import get_logger logger = get_logger("material_handler") class MaterialAcceptHandler(BaseHandler): """ 原料采购 """ def handle(self, event: BusinessEvent) -> None: """ :param event: :return: """ payload = event.payload mat_type = payload["material_type"] purity = payload["purity"] weight = payload["weight"] logger.info(f"===== 原料采购验收业务开始 =====") logger.info(f"原料：{mat_type} | 纯度：{purity} | 重量：{weight}g") logger.info(f"{mat_type}原料验收合格，入库完成") # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:56 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : process_handler.py from .base_handler import BaseHandler from ReactorPattern.event import BusinessEvent from ReactorPattern.common import get_logger logger = get_logger("process_handler") class ProcessCheckHandler(BaseHandler): """ 加工质检 """ def handle(self, event: BusinessEvent) -> None: """ :param event: :return: """ payload = event.payload style = payload["style"] cert_no = payload["cert_no"] craft = payload["craft"] logger.info(f"===== 珠宝加工质检业务开始 =====") logger.info(f"款式：{style} | GIA证书号：{cert_no} | 工艺标准：{craft}") logger.info(f"{style}工艺、证书全部合规，成品入库") # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:57 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : sale_handler.py from .base_handler import BaseHandler from ReactorPattern.event import BusinessEvent from ReactorPattern.common import get_logger logger = get_logger("sale_handler") class StoreSaleHandler(BaseHandler): """ 门店销售结算 """ def handle(self, event: BusinessEvent) -> None: """ :param event: :return: """ payload = event.payload product = payload["product_name"] price = payload["unit_price"] qty = payload["quantity"] order_id = payload["order_id"] total = price * qty logger.info(f"===== 门店销售结算业务开始 =====") logger.info(f"商品：{product} | 单价：{price}元 | 数量：{qty}件") logger.info(f"订单{order_id}，结算总价：{total}元，销售完成") # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:58 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : repair_handler.py from .base_handler import BaseHandler from ReactorPattern.event import BusinessEvent from ReactorPattern.common import get_logger logger = get_logger("repair_handler") class AfterSaleRepairHandler(BaseHandler): """ 售后维修 """ def handle(self, event: BusinessEvent) -> None: """ :param event: :return: """ payload = event.payload customer = payload["customer_name"] jewel = payload["jewelry_name"] item = payload["repair_item"] logger.info(f"===== 售后维修业务开始 =====") logger.info(f"客户：{customer} | 饰品：{jewel} | 维修项目：{item}") logger.info(f"{item}服务完工，客户可取件") # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 20:59 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : inventory_handler.py from .base_handler import BaseHandler from ReactorPattern.event import BusinessEvent from ReactorPattern.common import get_logger logger = get_logger("inventory_handler") class InventoryCheckHandler(BaseHandler): """ 库存盘点 """ def handle(self, event: BusinessEvent) -> None: """ :param event: :return: """ payload = event.payload inv_type = payload["inventory_type"] total_num = payload["total_num"] total_val = payload["total_value"] logger.info(f"===== 库存盘点业务开始 =====") logger.info(f"盘点类目：{inv_type} | 在库总数：{total_num}件 | 库存总价值：{total_val}万元") logger.info(f"账实核对一致，库存台账更新完成")# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 21:00 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : demultiplexer.py from typing import Dict from ReactorPattern.common import BusinessEventType, HandlerNotFoundError from ReactorPattern.handler import BaseHandler class EventDemultiplexer: """ 事件分离器：维护事件类型与处理器映射，负责匹配分发 """ def __init__(self): self._handler_map: Dict[BusinessEventType, BaseHandler] = {} def register_handler(self, event_type: BusinessEventType, handler: BaseHandler) -> None: """ 注册事件与处理器绑定关系，扩展新增业务只需要注册 :param event_type: :param handler: :return: """ self._handler_map[event_type] = handler def get_handler(self, event_type: BusinessEventType) -> BaseHandler: """ 根据事件类型查找对应处理器，无则抛异常 :param event_type: :return: """ if event_type not in self._handler_map: raise HandlerNotFoundError(f"未注册处理器：{event_type.value}") return self._handler_map[event_type] # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 21:02 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : reactor_core.py from ReactorPattern.event import BusinessEvent from ReactorPattern.reactor import EventDemultiplexer from ReactorPattern.common import get_logger logger = get_logger("reactor_core") class JewelryReactor: """ 珠宝行业Reactor反应器，统一接收并发事件、调用分离器分发 """ def __init__(self, demultiplexer: EventDemultiplexer): self._demux = demultiplexer def dispatch(self, event: BusinessEvent) -> None: """ 对外统一入口：接收任意业务事件，自动分发处理 :param event: :return: """ logger.info(f"Reactor收到业务事件：{event.event_type.value}") handler = self._demux.get_handler(event.event_type) handler.handle(event) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 21:05 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : app.py from .common import BusinessEventType, get_logger from .event import BusinessEvent from .reactor import EventDemultiplexer, JewelryReactor from .handler import ( MaterialAcceptHandler, ProcessCheckHandler, StoreSaleHandler, AfterSaleRepairHandler, InventoryCheckHandler ) logger = get_logger("app") def build_reactor() -> JewelryReactor: """ 依赖组装：统一注册所有业务处理器，工厂化构建Reactor :return: """ demux = EventDemultiplexer() # 注册所有珠宝业务处理器 demux.register_handler(BusinessEventType.RAW_MATERIAL, MaterialAcceptHandler()) demux.register_handler(BusinessEventType.PROCESS_CHECK, ProcessCheckHandler()) demux.register_handler(BusinessEventType.STORE_SALE, StoreSaleHandler()) demux.register_handler(BusinessEventType.AFTER_SALE_REPAIR, AfterSaleRepairHandler()) demux.register_handler(BusinessEventType.INVENTORY_CHECK, InventoryCheckHandler()) reactor = JewelryReactor(demultiplexer=demux) return reactor def simulate_concurrent_business_requests(reactor: JewelryReactor): """ 模拟多并发业务请求，统一封装为标准Event对象送入Reactor :param reactor: :return: """ event_list = [ BusinessEvent( event_type=BusinessEventType.RAW_MATERIAL, payload={"material_type": "足金999", "purity": "99.9%", "weight": 800} ), BusinessEvent( event_type=BusinessEventType.PROCESS_CHECK, payload={"style": "1克拉钻戒", "cert_no": "GIA987654321", "craft": "微镶精工"} ), BusinessEvent( event_type=BusinessEventType.STORE_SALE, payload={"product_name": "古法传承金镯", "unit_price": 15680, "quantity": 1, "order_id": "JEW20260615001"} ), BusinessEvent( event_type=BusinessEventType.AFTER_SALE_REPAIR, payload={"customer_name": "李女士", "jewelry_name": "18K金项链", "repair_item": "断裂焊接+抛光翻新"} ), BusinessEvent( event_type=BusinessEventType.INVENTORY_CHECK, payload={"inventory_type": "钻石成品库存", "total_num": 1680, "total_value": 12600} ) ] # 循环推送所有并发事件给Reactor分发 for evt in event_list: reactor.dispatch(evt)调用：
# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Reactor 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/15 21:07 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : ReactorBll.py from ReactorPattern.common import get_logger from ReactorPattern.reactor import JewelryReactor from ReactorPattern.app import build_reactor, simulate_concurrent_business_requests logger = get_logger("ReactorBll") class ReactorBll(object): """ Reactor 业务逻辑封装层 对外提供统一调用入口，隐藏内部构建、分发、调度细节 """ def __init__(self): # 内部初始化 Reactor 实例（一次创建，多次复用） self._reactor: JewelryReactor = build_reactor() def demo(self): """ 对外演示方法：一键执行全套珠宝行业并发业务流程 :return: None """ logger.info("========== ReactorBll.demo() 开始执行 ==========") # 执行并发业务模拟 simulate_concurrent_business_requests(self._reactor) logger.info("========== ReactorBll.demo() 执行完成 ==========") def get_reactor(self) -> JewelryReactor: """ 提供内部 reactor 实例，方便外部扩展调用（可选） """ return self._reactor输出：
