python: Publish/Subscribe Pattern II

博客园 - ®Geovin Du Dream Park™

go: Functional Options Pattern go:Timing Functions Pattern python: Timing Functions Pattern go: Steady-State Pattern python: Steady-State Pattern go: Handshaking Pattern python: Handshaking Pattern go: Fail-Fast Pattern python: Fail-Fast Pattern I go: Deadline Pattern python: Deadline Pattern go: Circuit-Breaker Pattern python: Circuit-Breaker Pattern go: Bulkheads Pattern python: Bulkheads Pattern go: Push & Pull Pattern python: Push & Pull Pattern go: Publish/Subscribe Pattern python: Publish/Subscribe Pattern go: Futures & Promises Pattern python: Futures & Promises Pattern go: Worker Pool Pattern go：Pipeline Pattern python: Worker Pool Pattern python: Pipeline Pattern go: Fan-Out Pattern python: Fan-Out Pattern go: Fan-In Pattern python: Fan-In Pattern Fan-In go: Producer Consumer Pattern python: Producer Consumer Pattern go: Parallelism Pattern python: Parallelism Pattern go: Reactor Pattern python: Reactor Pattern go: Generators Pattern python: speech to text offline python: Generators Pattern go: Coroutines Pattern python:Coroutines Pattern go: Broadcast Pattern python: Broadcast Pattern python: Bounded Parallelism Pattern go: Bounded Parallelism Pattern python: N-Barrier Pattern go: N-Barrier Pattern python: Semaphore Pattern go: Semaphore Pattern python: Read-Write Lock Pattern go: Read-Write Lock Pattern python: Monitor Pattern go: Monitor Pattern python: Mutex Pattern go: Lock/Mutex Pattern Python: Condition Variable Pattern go:Condition Variable Pattern python: Registry Pattern python: Interpreter Pattern go: Interpreter Pattern go: Registry Pattern go: Memento Pattern go: Command Pattern go: State Pattern go:Template Method Pattern go: Strategy Pattern go: Visitor Pattern go: Observer Pattern go: Mediator Pattern go: Iterator Pattern go: Chain of Responsibility Pattern go: Proxy Pattern go:Decorator Pattern go: Facade Pattern go: Flyweight Pattern go: Composite Pattern go: Singleton Pattern go: Prototype Pattern go: Bridge Pattern go: Adapter Pattern go: Builder Pattern 密码进行加盐哈希 using CSharp,Python,Go,Java go: Abstract Factory Pattern go: Model，Interface，DAL ，Factory，BLL using mysql go: Simple Factory Pattern go: Factory Method Pattern go: goLang 在Windows环境搭建Go语言开发环境 CSharp: Parallel Extensions cpp: class python: 蝴蝶跟随鼠标 javascript: 中国历史人物热力分布图using echart javascript: 中国历史人物热力图 python: 初养龙虾微信纯文字自动回复using workBuddy python: object 入门 python: Factory Method Pattern python: Simple Factory Pattern python: Abstract Factory Pattern 区域化代码 python: Null Object Pattern python: Builder Pattern python: Adapter Pattern

®Geovin Du Dream Park™ · 2026-06-25 · via 博客园 - ®Geovin Du Dream Park™

项目结构：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 22:00 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : settings.py
import os
from dotenv import load_dotenv

load_dotenv()

# 日志配置
LOG_LEVEL = "INFO"
LOG_DIR = "./logs"
LOG_ROTATE_DAYS = 7

# 消息持久化数据库
DB_PATH = "./jewelry_msg.db"

# 异步线程池
ASYNC_WORKER_NUM = 5

# Broker开关 False=本地内存  True=RabbitMQ远程队列
USE_RABBITMQ = False

# RabbitMQ连接参数
RABBIT_HOST = os.getenv("RABBIT_HOST", "127.0.0.1")
RABBIT_PORT = int(os.getenv("RABBIT_PORT", 5672))
RABBIT_USER = os.getenv("RABBIT_USER", "guest")
RABBIT_PWD = os.getenv("RABBIT_PWD", "guest")
RABBIT_VHOST = "/"


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:15 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : topics.py
"""
统一消息主题配置，杜绝硬编码
"""
class Topic:
    """

    """
    # 珠宝全业务流程主题
    RAW_MATERIAL_PURCHASE = "RawMaterialPurchase"
    JEWELRY_PRODUCTION = "JewelryProduction"
    QUALITY_INSPECTION = "QualityInspection"
    STORE_DISTRIBUTION = "StoreDistribution"
    JEWELRY_SALE = "JewelrySale"
    AFTER_SALES_CARE = "AfterSalesCare"



# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:16 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : models.py
from dataclasses import dataclass
from typing import Optional

"""
珠宝行业领域消息模型
"""
@dataclass
class JewelryMessage:
    """
    基础消息
    """
    title: str
    content: str
    department: str

@dataclass
class RawMaterialPurchaseMsg(JewelryMessage):
    """

    """
    material: str
    quantity: str

@dataclass
class JewelryProductionMsg(JewelryMessage):
    """

    """
    product_count: int
    product_type: str

@dataclass
class QualityInspectionMsg(JewelryMessage):
    """

    """
    qualified_count: int
    has_certificate: bool

@dataclass
class StoreDistributionMsg(JewelryMessage):
    """

    """
    store_name: str
    distribute_count: int

@dataclass
class JewelrySaleMsg(JewelryMessage):
    """

    """
    product_name: str
    amount: float
    store: str

@dataclass
class AfterSalesCareMsg(JewelryMessage):
    """

    """
    customer_name: str
    service_type: str



# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 22:01 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : log_config.py
import logging
from logging.handlers import TimedRotatingFileHandler
import os
from PublishSubscribePattern.config.settings import LOG_LEVEL, LOG_DIR, LOG_ROTATE_DAYS

def init_logger():
    if not os.path.exists(LOG_DIR):
        os.makedirs(LOG_DIR)

    log_format = logging.Formatter(
        "[%(asctime)s] [%(levelname)s] [%(name)s] %(filename)s:%(lineno)d - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )
    root_log = logging.getLogger()
    root_log.setLevel(LOG_LEVEL)

    # 控制台输出
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(log_format)
    root_log.addHandler(console_handler)

    # 按日切割日志文件
    file_handler = TimedRotatingFileHandler(
        filename=os.path.join(LOG_DIR, "jewelry_pubsub.log"),
        when="D",
        interval=1,
        backupCount=LOG_ROTATE_DAYS,
        encoding="utf-8"
    )
    file_handler.setFormatter(log_format)
    root_log.addHandler(file_handler)
    return root_log

logger = init_logger()


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 22:06 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : async_publisher.py
from concurrent.futures import ThreadPoolExecutor
from PublishSubscribePattern.config.settings import ASYNC_WORKER_NUM
from PublishSubscribePattern.logger.log_config import logger

class AsyncPublisher:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=ASYNC_WORKER_NUM, thread_name_prefix="msg-worker")

    def publish_async(self, publish_func, topic, msg):
        def task():
            try:
                publish_func(topic, msg)
            except Exception as e:
                logger.error(f"异步发布失败 topic:{topic} error:{str(e)}", exc_info=True)
        self.executor.submit(task)

async_publisher = AsyncPublisher()


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:20 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : broker.py
from collections import defaultdict
from typing import Callable, Dict, List, Any
from .exceptions import SubscriberInvalidError
from PublishSubscribePattern.logger.log_config import logger
from .message_store import msg_store
"""
企业级 Pub/Sub 消息代理（单例、线程安全、可扩展）
"""
class PubSubBroker:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.topic_subscribers: Dict[str, List[Callable]] = defaultdict(list)
        return cls._instance

    def subscribe(self, topic: str, callback: Callable[[Any], None]) -> None:
        if not callable(callback):
            raise SubscriberInvalidError("订阅者必须传入可执行回调函数")
        self.topic_subscribers[topic].append(callback)
        logger.info(f"订阅注册成功 topic={topic} handler={callback.__name__}")

    def unsubscribe(self, topic: str, callback: Callable[[Any], None]) -> None:
        if callback in self.topic_subscribers[topic]:
            self.topic_subscribers[topic].remove(callback)
            logger.info(f"取消订阅 topic={topic} handler={callback.__name__}")

    def publish(self, topic: str, message: Any) -> None:
        subscribers = self.topic_subscribers.get(topic, [])
        msg_id = msg_store.save_publish(topic, message)
        logger.info(f"发布消息 topic={topic} msg_id={msg_id} 订阅者:{len(subscribers)}")

        if not subscribers:
            logger.warning(f"主题 {topic} 无订阅者")
            return

        # 只在最后标记一次消费
        success = True
        for cb in subscribers:
            try:
                cb(message)
            except Exception as e:
                success = False
                logger.error(f"订阅者 {cb.__name__} 处理失败: {e}", exc_info=True)

        if success:
            msg_store.mark_consumed(msg_id)

broker = PubSubBroker()

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:17 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : exceptions.py
"""
消息模块自定义异常
"""
class PubSubException(Exception):
    """
    基础异常
    """
    pass

class TopicNotExistError(PubSubException):
    """
    主题不存在异常
    """
    pass

class SubscriberInvalidError(PubSubException):
    """
    订阅者无效异常
    """
    pass




# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 22:07 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : message_store.py
import sqlite3
import json
import uuid
from datetime import datetime
from dataclasses import asdict
from PublishSubscribePattern.config.settings import DB_PATH
from PublishSubscribePattern.logger.log_config import logger

class MessageStore:
    def __init__(self):
        self.DB_PATH = DB_PATH
        self._create_table()

    # 每次操作都创建新连接（彻底解决多线程冲突）
    def _get_conn(self):
        return sqlite3.connect(self.DB_PATH, check_same_thread=False)

    def _create_table(self):
        conn = self._get_conn()
        cursor = conn.cursor()
        try:
            cursor.execute("""
            CREATE TABLE IF NOT EXISTS message_record (
                msg_id TEXT PRIMARY KEY,
                topic TEXT NOT NULL,
                msg_data TEXT NOT NULL,
                publish_time TEXT NOT NULL,
                consume_status INTEGER DEFAULT 0
            )
            """)
            conn.commit()
        finally:
            cursor.close()
            conn.close()

    def save_publish(self, topic: str, msg_obj):
        msg_id = str(uuid.uuid4())
        data = json.dumps(asdict(msg_obj), ensure_ascii=False)
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        conn = self._get_conn()
        cursor = conn.cursor()
        try:
            cursor.execute(
                "INSERT INTO message_record (msg_id, topic, msg_data, publish_time) VALUES (?,?,?,?)",
                (msg_id, topic, data, now)
            )
            conn.commit()
            logger.debug(f"消息入库成功: {msg_id}")
        except Exception as e:
            logger.error(f"消息保存失败: {e}")
        finally:
            cursor.close()
            conn.close()
        return msg_id

    def mark_consumed(self, msg_id: str):
        conn = self._get_conn()
        cursor = conn.cursor()
        try:
            cursor.execute(
                "UPDATE message_record SET consume_status=1 WHERE msg_id=?",
                (msg_id,)
            )
            conn.commit()
            logger.debug(f"消息已标记消费: {msg_id}")
        except Exception as e:
            logger.error(f"标记消费失败: {e}")
        finally:
            cursor.close()
            conn.close()

msg_store = MessageStore()



# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 22:04 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : rabbit_broker.py
import pika
import json
from dataclasses import asdict
from PublishSubscribePattern.config.settings import RABBIT_HOST, RABBIT_PORT, RABBIT_USER, RABBIT_PWD, RABBIT_VHOST
from PublishSubscribePattern.logger.log_config import logger
from .message_store import msg_store

class RabbitMQBroker:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.channel = None
            cls._instance.conn = None
            cls._instance.sub_map = {}
        return cls._instance

    def _connect(self):
        if self.conn and self.conn.is_open:
            return
        try:
            cred = pika.PlainCredentials(RABBIT_USER, RABBIT_PWD)
            params = pika.ConnectionParameters(
                host=RABBIT_HOST,
                port=RABBIT_PORT,
                virtual_host=RABBIT_VHOST,
                credentials=cred,
                heartbeat=60
            )
            self.conn = pika.BlockingConnection(params)
            self.channel = self.conn.channel()
            logger.info("RabbitMQ 连接成功")
        except Exception as e:
            logger.error(f"RabbitMQ 连接失败: {e}")
            raise

    def subscribe(self, topic: str, callback):
        self._connect()
        self.channel.exchange_declare(exchange=topic, exchange_type="fanout")
        res = self.channel.queue_declare(queue="", exclusive=True)
        queue_name = res.method.queue
        self.channel.queue_bind(exchange=topic, queue=queue_name)

        def wrapper(ch, method, props, body):
            raw = json.loads(body)
            callback(raw["data"])
            ch.basic_ack(delivery_tag=method.delivery_tag)

        self.channel.basic_consume(queue=queue_name, on_message_callback=wrapper)
        self.sub_map[(topic, queue_name)] = wrapper
        logger.info(f"RabbitMQ 订阅成功 topic={topic}")

    def publish(self, topic: str, msg_obj):
        self._connect()
        self.channel.exchange_declare(exchange=topic, exchange_type="fanout")
        msg_id = msg_store.save_publish(topic, msg_obj)
        payload = json.dumps({"msg_id": msg_id, "data": asdict(msg_obj)}, ensure_ascii=False)
        self.channel.basic_publish(exchange=topic, routing_key="", body=payload)
        logger.info(f"RabbitMQ 发布成功 topic={topic} msg_id={msg_id}")

# 延迟实例化
rabbit_broker = None

订阅者：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:24 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : customer.py
from PublishSubscribePattern.domain.models import JewelryMessage
from PublishSubscribePattern.logger.log_config import logger

class CustomerSubscriber:
    """

    """
    @staticmethod
    def handle(message: JewelryMessage):
        print(f"👥 客户部 | 接收: {message.title} | {message.content}")
        logger.info(f"👥 客户部 | 接收: {message.title} | {message.content}")




# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:21 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : finance.py
from PublishSubscribePattern.domain.models import JewelryMessage
from PublishSubscribePattern.logger.log_config import logger

class FinanceSubscriber:
    """

    """
    @staticmethod
    def handle(message: JewelryMessage):
        print(f"💰 财务部 | 接收: {message.title} | {message.content}")
        logger.info(f"💰 财务部 | 接收: {message.title} | {message.content}")



# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:22 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : quality.py
from PublishSubscribePattern.domain.models import JewelryMessage
from PublishSubscribePattern.logger.log_config import logger

class QualitySubscriber:
    @staticmethod
    def handle(message: JewelryMessage):
        print(f"🔍 质检部 | 接收: {message.title} | {message.content}")
        logger.info(f"🔍 质检部 | 接收: {message.title} | {message.content}")



# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:23 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : sales.py
from PublishSubscribePattern.domain.models import JewelryMessage
from PublishSubscribePattern.logger.log_config import logger

class SalesSubscriber:
    """

    """
    @staticmethod
    def handle(message: JewelryMessage):
        print(f"🛍️ 营销部 | 接收: {message.title} | {message.content}")
        logger.info(f"🛍️ 营销部 | 接收: {message.title} | {message.content}")



# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:23 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : store.py
from PublishSubscribePattern.domain.models import JewelryMessage
from PublishSubscribePattern.logger.log_config import logger


class StoreSubscriber:
    """

    """
    @staticmethod
    def handle(message: JewelryMessage):
        print(f"🏪 品牌门店 | 接收: {message.title} | {message.content}")
        logger.info(f"🏪 品牌门店 | 接收: {message.title} | {message.content}")


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:25 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : tech.py
from PublishSubscribePattern.domain.models import JewelryMessage
from PublishSubscribePattern.logger.log_config import logger


class TechSubscriber:
    """

    """

    @staticmethod
    def handle(message: JewelryMessage):
        print(f"🔧 资讯科技部 | 接收: {message.title} | {message.content}")
        logger.info(f"🔧 资讯科技部 | 接收: {message.title} | {message.content}")


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:21 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : warehouse.py
from PublishSubscribePattern.domain.models import JewelryMessage
from PublishSubscribePattern.logger.log_config import logger



class WarehouseSubscriber:
    """

    """
    @staticmethod
    def handle(message: JewelryMessage):
        print(f"🏬 仓库部 | 接收: {message.title} | {message.content}")
        logger.info(f"🏬 仓库部 | 接收消息: {message.title} | {message.content}")

发布者：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:34 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : after_sales.py
from PublishSubscribePattern.messaging.broker import broker
from PublishSubscribePattern.config.topics import Topic
from PublishSubscribePattern.domain.models import AfterSalesCareMsg
from PublishSubscribePattern.logger.log_config import logger
from PublishSubscribePattern.messaging.async_publisher import async_publisher
from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
from PublishSubscribePattern.config.settings import USE_RABBITMQ


def get_broker():
    if USE_RABBITMQ:
        from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
        return rabbit_broker
    return broker

class AfterSalesPublisher:
    """

    """

    @staticmethod
    def publish(customer_name: str, service_type: str):
        msg = AfterSalesCareMsg(
            title="客户售后保养申请",
            content=f"客户{customer_name}申请{service_type}服务，已受理",
            department="客服部",
            customer_name=customer_name,
            service_type=service_type
        )
        print(f"\n📢 发布: {Topic.AFTER_SALES_CARE}")
        broker.publish(Topic.AFTER_SALES_CARE, msg)
        target_broker = get_broker()
        async_publisher.publish_async(target_broker.publish, Topic.AFTER_SALES_CARE, msg)



# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:33 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : distribution.py
from PublishSubscribePattern.messaging.broker import broker
from PublishSubscribePattern.config.topics import Topic
from PublishSubscribePattern.domain.models import StoreDistributionMsg
from PublishSubscribePattern.logger.log_config import logger
from PublishSubscribePattern.messaging.async_publisher import async_publisher
from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
from PublishSubscribePattern.config.settings import USE_RABBITMQ



def get_broker():
    if USE_RABBITMQ:
        from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
        return rabbit_broker
    return broker

class DistributionPublisher:
    """

    """
    @staticmethod
    def publish(store_name: str, distribute_count: int):
        msg = StoreDistributionMsg(
            title="门店铺货完成",
            content=f"向{store_name}铺货{distribute_count}件珠宝，已出库发货",
            department="仓库部",
            store_name=store_name,
            distribute_count=distribute_count
        )
        print(f"\n📢 发布: {Topic.STORE_DISTRIBUTION}")
        broker.publish(Topic.STORE_DISTRIBUTION, msg)
        target_broker = get_broker()
        async_publisher.publish_async(target_broker.publish, Topic.STORE_DISTRIBUTION, msg)


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:32 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : inspection.py
from PublishSubscribePattern.messaging.broker import broker
from PublishSubscribePattern.config.topics import Topic
from PublishSubscribePattern.domain.models import QualityInspectionMsg
from PublishSubscribePattern.logger.log_config import logger
from PublishSubscribePattern.messaging.async_publisher import async_publisher
from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
from PublishSubscribePattern.config.settings import USE_RABBITMQ

def get_broker():
    if USE_RABBITMQ:
        from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
        return rabbit_broker
    return broker

class InspectionPublisher:
    """

    """

    @staticmethod
    def publish(qualified_count: int, has_certificate: bool):
        cert_str = "已出具证书" if has_certificate else "未出具证书"
        msg = QualityInspectionMsg(
            title="珠宝质量检测完成",
            content=f"总计{qualified_count}件珠宝，全部合格，{cert_str}",
            department="质检部",
            qualified_count=qualified_count,
            has_certificate=has_certificate
        )
        print(f"\n📢 发布: {Topic.QUALITY_INSPECTION}")
        broker.publish(Topic.QUALITY_INSPECTION, msg)
        target_broker = get_broker()
        async_publisher.publish_async(target_broker.publish, Topic.QUALITY_INSPECTION, msg)


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:27 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : production.py
from PublishSubscribePattern.messaging.broker import broker
from PublishSubscribePattern.config.topics import Topic
from PublishSubscribePattern.domain.models import JewelryProductionMsg
from PublishSubscribePattern.logger.log_config import logger
from PublishSubscribePattern.messaging.async_publisher import async_publisher
from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
from PublishSubscribePattern.config.settings import USE_RABBITMQ

def get_broker():
    if USE_RABBITMQ:
        from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
        return rabbit_broker
    return broker

def get_broker():
    return rabbit_broker if USE_RABBITMQ else broker


class ProductionPublisher:
    """

    """

    @staticmethod
    def publish(product_type: str, count: int):
        msg = JewelryProductionMsg(
            title="珠宝生产完成",
            content=f"{product_type}{count}件，待质检",
            department="生产部",
            product_count=count,
            product_type=product_type
        )
        print(f"\n📢 发布: {Topic.JEWELRY_PRODUCTION}")
        broker.publish(Topic.JEWELRY_PRODUCTION, msg)
        target_broker = get_broker()
        async_publisher.publish_async(target_broker.publish, Topic.JEWELRY_PRODUCTION, msg)


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:27 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : purchase.py
from PublishSubscribePattern.messaging.broker import broker
from PublishSubscribePattern.config.topics import Topic
from PublishSubscribePattern.domain.models import RawMaterialPurchaseMsg
from PublishSubscribePattern.logger.log_config import logger
from PublishSubscribePattern.messaging.async_publisher import async_publisher
from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
from PublishSubscribePattern.config.settings import USE_RABBITMQ

def get_broker():
    if USE_RABBITMQ:
        from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
        return rabbit_broker
    return broker

class PurchasePublisher:
    """

    """

    @staticmethod
    def publish(material: str, quantity: str):
        msg = RawMaterialPurchaseMsg(
            title="原材料采购完成",
            content=f"{material} {quantity}，待入库",
            department="采购部",
            material=material,
            quantity=quantity
        )
        print(f"\n📢 发布: {Topic.RAW_MATERIAL_PURCHASE}")
        broker.publish(Topic.RAW_MATERIAL_PURCHASE, msg)
        target_broker = get_broker()
        async_publisher.publish_async(target_broker.publish, Topic.RAW_MATERIAL_PURCHASE, msg)


# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:33 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : sale.py
from PublishSubscribePattern.messaging.broker import broker
from PublishSubscribePattern.config.topics import Topic
from PublishSubscribePattern.domain.models import JewelrySaleMsg
from PublishSubscribePattern.logger.log_config import logger
from PublishSubscribePattern.messaging.async_publisher import async_publisher
from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
from PublishSubscribePattern.config.settings import USE_RABBITMQ

def get_broker():
    if USE_RABBITMQ:
        from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker
        return rabbit_broker
    return broker

class SalePublisher:
    """

    """
    @staticmethod
    def publish(product_name: str, amount: float, store: str):
        msg = JewelrySaleMsg(
            title="珠宝销售成功",
            content=f"{store}售出{product_name}，收款{amount:.2f}元，库存已扣减",
            department="门店",
            product_name=product_name,
            amount=amount,
            store=store
        )
        print(f"\n📢 发布: {Topic.JEWELRY_SALE}")
        broker.publish(Topic.JEWELRY_SALE, msg)
        target_broker = get_broker()
        async_publisher.publish_async(target_broker.publish, Topic.JEWELRY_SALE, msg)

调用：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Publish/Subscribe 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/23 21:29 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : PublishSubscribeBll.py
'''

'''
from PublishSubscribePattern.logger.log_config import logger
from PublishSubscribePattern.config.settings import USE_RABBITMQ
from PublishSubscribePattern.config.topics import Topic

# 切换消息Broker
if USE_RABBITMQ:
    from PublishSubscribePattern.messaging.rabbit_broker import rabbit_broker as broker
else:
    from PublishSubscribePattern.messaging.broker import broker

from PublishSubscribePattern.messaging.broker import broker
from PublishSubscribePattern.config.topics import Topic

# 订阅者
from PublishSubscribePattern.service.subscribers.warehouse import WarehouseSubscriber
from PublishSubscribePattern.service.subscribers.finance import FinanceSubscriber
from PublishSubscribePattern.service.subscribers.quality import QualitySubscriber
from PublishSubscribePattern.service.subscribers.sales import SalesSubscriber
from PublishSubscribePattern.service.subscribers.store import StoreSubscriber
from PublishSubscribePattern.service.subscribers.customer import CustomerSubscriber
from PublishSubscribePattern.service.subscribers.tech import TechSubscriber

# 发布者
from PublishSubscribePattern.service.publishers.purchase import PurchasePublisher
from PublishSubscribePattern.service.publishers.production import ProductionPublisher
from PublishSubscribePattern.service.publishers.inspection import InspectionPublisher
from PublishSubscribePattern.service.publishers.distribution import DistributionPublisher
from PublishSubscribePattern.service.publishers.sale import SalePublisher
from PublishSubscribePattern.service.publishers.after_sales import AfterSalesPublisher


class PublishSubscribeBll(object):
    """

    """

    def register_subscribers(self):
        """统一注册订阅者"""
        broker.subscribe(Topic.RAW_MATERIAL_PURCHASE, WarehouseSubscriber.handle)
        broker.subscribe(Topic.RAW_MATERIAL_PURCHASE, FinanceSubscriber.handle)

        broker.subscribe(Topic.JEWELRY_PRODUCTION, QualitySubscriber.handle)
        broker.subscribe(Topic.JEWELRY_PRODUCTION, WarehouseSubscriber.handle)

        broker.subscribe(Topic.QUALITY_INSPECTION, SalesSubscriber.handle)
        broker.subscribe(Topic.QUALITY_INSPECTION, WarehouseSubscriber.handle)
        broker.subscribe(Topic.QUALITY_INSPECTION, FinanceSubscriber.handle)

        broker.subscribe(Topic.STORE_DISTRIBUTION, StoreSubscriber.handle)
        broker.subscribe(Topic.STORE_DISTRIBUTION, SalesSubscriber.handle)

        broker.subscribe(Topic.JEWELRY_SALE, FinanceSubscriber.handle)
        broker.subscribe(Topic.JEWELRY_SALE, WarehouseSubscriber.handle)

        broker.subscribe(Topic.AFTER_SALES_CARE, CustomerSubscriber.handle)
        broker.subscribe(Topic.AFTER_SALES_CARE, TechSubscriber.handle)

    def demo(self):
        """

        :return:
        """
        print("=" * 70)
        print("珠宝企业 - 企业级发布/订阅系统启动")
        print("=" * 70)

        # 1. 注册订阅
        self.register_subscribers()

        # 2. 执行业务流程
        PurchasePublisher.publish("18K金+钻石", "500克+100颗")
        ProductionPublisher.publish("钻石戒指+金项链", 80)
        InspectionPublisher.publish(80, True)
        DistributionPublisher.publish("北京/上海门店", 80)
        SalePublisher.publish("钻石戒指", 12800.0, "上海店")
        AfterSalesPublisher.publish("张先生", "钻石清洗保养")

        # 3. 测试取消订阅
        print("\n--- 取消仓库订阅销售主题 ---")
        broker.unsubscribe(Topic.JEWELRY_SALE, WarehouseSubscriber.handle)
        SalePublisher.publish("金项链", 3800.0, "北京店")

输出：

81fda515b5026f6b8a8b72b88aba1adc

哲学管理(学)人生, 文学艺术生活, 自动(计算机学)物理(学)工作, 生物(学)化学逆境, 历史(学)测绘(学)时间, 经济(学)数学金钱(理财), 心理(学)医学情绪, 诗词美容情感, 美学建筑(学)家园, 解构建构(分析)整合学习, 智商情商(IQ、EQ)运筹(学)生存.---Geovin Du(涂聚文）

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

博客园 - ®Geovin Du Dream Park™