第6章 6.4 设计模式：单例/工厂/观察者

📖 这是系列教程「Python 从入门到精通」的第 29 章
- 上一篇：第 6 章 6.3 魔术方法与魔术常量
- 下一篇：第 6 章 6.5 综合实战：MVC 框架迷你版

🎯 开场：为什么你的代码总是不够"优雅"？

上两章我们学了 Python 的魔术方法和魔术常量，就像学会了武功的内功心法。但光有内功还不够，你还需要一些武林秘籍——别人总结好的、反复验证过的代码结构。

你有没有遇到过这些情况：

• 项目里到处要读配置，结果每个地方都 new Config() 一遍，内存里一堆重复对象……
• 程序要根据不同情况创建不同对象，if/else 写了几十行，改一下要改一堆地方……
• 某个数据变了，需要手动通知几十个地方更新代码，漏一个就出 bug……

这些问题，设计模式能帮你优雅地解决。

今天我们学三个最常用的：

| 模式 | 一句话理解 | 像什么 |
|------|-----------|--------|\n\n\n\n\n\n
| 单例模式 | 全局就一个实例 | 一个城市只有一个市长 |
| 工厂模式 | 批量生产"差不多"的对象 | 富士康流水线 |
| 观察者模式 | 一变多个跟着变 | 粉丝等博主发微博 |

学完这章，你写的代码会从「能跑」升级到「优雅可维护」。

🧱 基础：三个模式到底是什么？

1. 单例模式：全局就一个实例

是什么

单例模式（Singleton）保证一个类只有一个实例，不管你 new 多少次，拿到的都是同一个对象。

生活类比

一个城市只有一个市长。不管你在哪个区问"市长是谁"，答案都是同一个人。你不能"新建"一个市长。

为什么用

• 配置管理：整个程序共享一份配置，不会有多份不一致的问题
• 数据库连接：一个数据库连接对象到处用，不用每次都新建断开
• 日志器：一个日志对象记录所有日志

怎么用

class Singleton:
"""单例模式基类"""
_instance = None

def __new__(cls):
    if cls._instance is None:
        cls._instance = super().__new__(cls)
        cls._instance._initialized = False
    return cls._instance

def __init__(self):
    if self._initialized:
        return
    self._initialized = True
    self.value = "我是唯一的实例"


# 测试一下
a = Singleton()
b = Singleton()
c = Singleton()

print(f"a 的 id: {id(a)}")
print(f"b 的 id: {id(b)}")
print(f"c 的 id: {id(c)}")
print(f"a is b is c: {a is b is c}")
print(f"a.value = {a.value}")

输出：

a 的 id: 140234567890
b 的 id: 140234567890
c 的 id: 140234567890
a is b is c: True
a.value = 我是唯一的实例

解释：三个变量指向同一个对象（id 相同），修改 a 的 value，b 和 c 也会看到新值。

2. 工厂模式：批量生产"差不多"的对象

是什么

工厂模式（Factory）不直接 new 对象，而是通过一个"工厂函数"来创建。你告诉工厂你要什么，它给你生产出来。

生活类比

去富士康买手机。你不需要知道流水线怎么焊接，你只要说"我要 iPhone 16"，工厂给你生产一个。你不需要懂螺丝拧几圈、屏幕怎么装。

为什么用

• 解耦：调用者不知道具体类名，只知道"我要一个xx"
• 集中管理：创建逻辑在一个地方，改一处全生效
• 扩展方便：加新产品不用改调用方代码

怎么用

# 产品类：动物
class Dog:
def speak(self):
    return "汪汪汪"

class Cat:
def speak(self):
    return "喵喵喵"

class Duck:
def speak(self):
    return "嘎嘎嘎"


# 工厂函数
def animal_factory(animal_type):
"""根据类型创建动物实例"""
animals = {
    "dog": Dog,
    "cat": Cat,
    "duck": Duck,
}

animal_class = animals.get(animal_type.lower())
if animal_class is None:
    raise ValueError(f"不支持的动物类型: {animal_type}")

return animal_class()


# 使用工厂
pet1 = animal_factory("dog")
pet2 = animal_factory("cat")
pet3 = animal_factory("duck")

print(f"pet1 是 {type(pet1).__name__}，它说：{pet1.speak()}")
print(f"pet2 是 {type(pet2).__name__}，它说：{pet2.speak()}")
print(f"pet3 是 {type(pet3).__name__}，它说：{pet3.speak()}")

输出：

pet1 是 Dog，它说：汪汪汪
pet2 是 Cat，它说：喵喵喵
pet3 是 Duck，它说：嘎嘎嘎

解释：你不需要知道 Dog/Cat/Duck 类是怎么定义的，只要会调用 animal_factory("dog") 就行。

3. 观察者模式：一变多个跟着变

是什么

观察者模式（Observer）当一个对象（主题/Subject）状态改变时，自动通知所有订阅了它的对象（观察者/Observer）。

生活类比

微博博主发微博，粉丝都收到推送。你关注了博主，他一发微博你就收到通知。你不需要每秒刷新看他有没有发新内容。

为什么用

• 事件驱动：数据变了，自动触发更新，不用手动一个个通知
• 解耦：主题和观察者互不相知，通过"订阅"机制连接
• 实时性强：变化后立即通知，不遗漏

怎么用

from typing import Callable, List


class NewsAgency:
"""新闻社（主题）"""

def __init__(self):
    self._subscribers: List[Callable] = []
    self._latest_news = ""

def subscribe(self, callback: Callable):
    """订阅新闻"""
    self._subscribers.append(callback)
    print(f"✅ 新订阅者加入，当前共 {len(self._subscribers)} 个订阅者")

def unsubscribe(self, callback: Callable):
    """取消订阅"""
    if callback in self._subscribers:
        self._subscribers.remove(callback)
        print(f"❌ 一个订阅者退出，当前共 {len(self._subscribers)} 个订阅者")

def publish_news(self, news: str):
    """发布新闻，自动通知所有订阅者"""
    self._latest_news = news
    print(f"\n📰 发布新闻：{news}")
    self._notify_all()

def _notify_all(self):
    """通知所有订阅者"""
    for callback in self._subscribers:
        callback(self._latest_news)


# 定义两个订阅者（观察者）
def subscriber_xiaoming(news):
print(f"  📱 小明收到：{news}")

def subscriber_xiaoli(news):
print(f"  📱 小李收到：{news}")


# 测试一下
agency = NewsAgency()

agency.subscribe(subscriber_xiaoming)
agency.subscribe(subscriber_xiaoli)

agency.publish_news("Python 4.0 正式发布！")
agency.publish_news("PHP 终于能写前端了！")

agency.unsubscribe(subscriber_xiaoli)

agency.publish_news("JavaScript 25岁了！")

输出：

✅ 新订阅者加入，当前共 1 个订阅者
✅ 新订阅者加入，当前共 2 个订阅者

📰 发布新闻：Python 4.0 正式发布！
📱 小明收到：Python 4.0 正式发布！
📱 小李收到：Python 4.0 正式发布！

📰 发布新闻：PHP 终于能写前端了！
📱 小明收到：PHP 终于能写前端了！
📱 小李收到：PHP 终于能写前端了！

❌ 一个订阅者退出，当前共 1 个订阅者

📰 发布新闻：JavaScript 25岁了！
📱 小明收到：JavaScript 25岁了！

解释：小李退订后，就不再收到新闻了。发布者不需要知道有多少订阅者，订阅者也不需要知道新闻社怎么运作。

🔥 实战：三个递进小项目

项目 1（5 分钟）：用单例模式管理应用配置

场景：你写了一个小工具，需要读取数据库配置、API 密钥等配置。全局只能有一份配置。

class Config:
"""应用配置类（单例模式）"""
_instance = None

def __new__(cls):
    if cls._instance is None:
        cls._instance = super().__new__(cls)
        cls._instance._initialized = False
    return cls._instance

def __init__(self):
    if self._initialized:
        return
    self._initialized = True
    self.database_url = "localhost:5432"
    self.api_key = "sk-test-123456"
    self.debug_mode = True

def get(self, key):
    """获取配置项"""
    return getattr(self, key, None)


# 模拟程序不同地方读取配置
def init_database():
config = Config()
print(f"[数据库模块] 连接地址: {config.get('database_url')}")

def use_api():
config = Config()
print(f"[API模块] 使用密钥: {config.get('api_key')}")

def check_env():
config = Config()
print(f"[环境模块] 调试模式: {config.get('debug_mode')}")


# 运行
print("=== 程序启动 ===")
init_database()
use_api()
check_env()

print("\n=== 验证是同一实例 ===")
c1 = Config()
c2 = Config()
print(f"c1 is c2: {c1 is c2}")

预期输出：

=== 程序启动 ===
[数据库模块] 连接地址: localhost:5432
[API模块] 使用密钥: sk-test-123456
[环境模块] 调试模式: True

=== 验证是同一实例 ===
c1 is c2: True

一句话解释：三个模块都用 Config() 获取配置，但拿到的是同一个实例，保证配置一致。

项目 2（15 分钟）：用工厂模式做一个简单的日志处理器

场景：你有一个数据分析脚本，要根据配置生成不同格式的日志（JSON 格式给程序读、文本格式给人看）。

import json
from datetime import datetime
from typing import Dict


# ============ 产品类 ============
class LogFormatter:
"""日志格式化器基类"""
def format(self, level: str, message: str) -> str:
    raise NotImplementedError


class JsonFormatter(LogFormatter):
"""JSON 格式（给程序读）"""
def format(self, level: str, message: str) -> str:
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "level": level,
        "message": message
    }
    return json.dumps(log_entry, ensure_ascii=False)


class TextFormatter(LogFormatter):
"""文本格式（给人看）"""
def format(self, level: str, message: str) -> str:
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return f"[{timestamp}] [{level}] {message}"


class HtmlFormatter(LogFormatter):
"""HTML 格式（给浏览器看）"""
def format(self, level: str, message: str) -> str:
    color_map = {
        "INFO": "blue",
        "WARNING": "orange",
        "ERROR": "red"
    }
    color = color_map.get(level, "black")
    return f'<span style="color:{color}">[{level}] {message}</span>'


# ============ 工厂 ============
class FormatterFactory:
"""日志格式化器工厂"""
_formatters = {
    "json": JsonFormatter,
    "text": TextFormatter,
    "html": HtmlFormatter
}

@classmethod
def create(cls, format_type: str) -> LogFormatter:
    formatter_class = cls._formatters.get(format_type.lower())
    if not formatter_class:
        available = ", ".join(cls._formatters.keys())
        raise ValueError(f"不支持的格式: {format_type}，可用: {available}")
    return formatter_class()

@classmethod
def register(cls, name: str, formatter_class):
    """注册新的格式化器（扩展用）"""
    cls._formatters[name.lower()] = formatter_class


# ============ 日志器（使用工厂） ============
class Logger:
def __init__(self, format_type: str = "text"):
    self.formatter = FormatterFactory.create(format_type)

def log(self, level: str, message: str):
    output = self.formatter.format(level, message)
    print(output)


# ============ 测试 ============
if __name__ == "__main__":
print("=== JSON 格式（给程序读）===")
logger_json = Logger("json")
logger_json.log("INFO", "用户登录成功")
logger_json.log("ERROR", "数据库连接超时")

print("\n=== 文本格式（给人看）===")
logger_text = Logger("text")
logger_text.log("WARNING", "磁盘空间不足 20%")

print("\n=== HTML 格式（给浏览器看）===")
logger_html = Logger("html")
logger_html.log("INFO", "页面加载完成")

预期输出：

=== JSON 格式（给程序读）===
{"timestamp": "2024-01-15T10:30:45.123456", "level": "INFO", "message": "用户登录成功"}
{"timestamp": "2024-01-15T10:30:45.123456", "level": "ERROR", "message": "数据库连接超时"}

=== 文本格式（给人看）===
[2024-01-15 10:30:45] [WARNING] 磁盘空间不足 20%

=== HTML 格式（给浏览器看）===
<span style="color:blue">[INFO] 页面加载完成</span>

一句话解释：换一种格式日志，不用改日志器代码，只要换工厂创建的不同格式化器就行。

项目 3（15 分钟）：观察者模式 + 工厂模式做一个「价格监控小工具」

场景：你开网店，想监控几个商品的价格，价格变动时自动通知你（邮件/短信/日志）。

from typing import Callable, List, Dict
from datetime import datetime
import json


# ============ 观察者：通知渠道 ============
class NotificationChannel:
"""通知渠道基类"""
def send(self, product_name: str, old_price: float, new_price: float):
    raise NotImplementedError


class EmailChannel(NotificationChannel):
def send(self, product_name: str, old_price: float, new_price: float):
    change = new_price - old_price
    pct = (change / old_price * 100) if old_price else 0
    print(f"📧 [邮件通知] {product_name}: ¥{old_price} → ¥{new_price} ({pct:+.1f}%)")


class SMSChannel(NotificationChannel):
def send(self, product_name: str, old_price: float, new_price: float):
    change = new_price - old_price
    direction = "📈上涨" if change > 0 else "📉下降"
    print(f"📱 [短信通知] {product_name} {direction}至 ¥{new_price}")


class LogChannel(NotificationChannel):
def send(self, product_name: str, old_price: float, new_price: float):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    record = {
        "time": timestamp,
        "product": product_name,
        "old_price": old_price,
        "new_price": new_price
    }
    print(f"📝 [日志] {json.dumps(record, ensure_ascii=False)}")


# ============ 工厂：创建通知渠道 ============
class ChannelFactory:
"""通知渠道工厂"""
_channels = {
    "email": EmailChannel,
    "sms": SMSChannel,
    "log": LogChannel
}

@classmethod
def create(cls, channel_type: str) -> NotificationChannel:
    channel_class = cls._channels.get(channel_type.lower())
    if not channel_class:

        raise ValueError(f"不支持的渠道类型: {channel_type}")
    return channel_class()


# ============ 主题：价格监控器 ============
class PriceMonitor:
"""商品价格监控器（主题）"""

def __init__(self, product_name: str, initial_price: float):
    self.product_name = product_name
    self._price = initial_price
    self._subscribers: List[NotificationChannel] = []

@property
def price(self) -> float:
    return self._price

@price.setter
def price(self, new_price: float):
    if new_price == self._price:
        return  # 价格没变，不通知

    old_price = self._price
    self._price = new_price
    print(f"\n🔔 价格变动检测: {self.product_name} ¥{old_price} → ¥{new_price}")
    self._notify_subscribers(old_price, new_price)

def subscribe(self, channel: NotificationChannel):
    self._subscribers.append(channel)

def unsubscribe(self, channel: NotificationChannel):
    if channel in self._subscribers:
        self._subscribers.remove(channel)

def _notify_subscribers(self, old_price: float, new_price: float):
    for channel in self._subscribers:
        channel.send(self.product_name, old_price, new_price)


# ============ 运行演示 ============
if __name__ == "__main__":
# 创建监控器
monitor = PriceMonitor("iPhone 16 Pro", 8999.0)

# 创建通知渠道
email = ChannelFactory.create("email")
sms = ChannelFactory.create("sms")
log = ChannelFactory.create("log")

# 订阅
monitor.subscribe(email)
monitor.subscribe(sms)
monitor.subscribe(log)

print("=== 开始监控 ===")

# 价格变动，触发通知
monitor.price = 8799.0  # 降价
monitor.price = 8799.0  # 没变，不通知

# 退订短信
monitor.unsubscribe(sms)

monitor.price = 9299.0  # 涨价，只有邮件和日志收到

预期输出：

=== 开始监控 ===

🔔 价格变动检测: iPhone 16 Pro ¥8999.0 → ¥8799.0
📧 [邮件通知] iPhone 16 Pro: ¥8999.0 → ¥8799.0 (-2.2%)
📱 [短信通知] iPhone 16 Pro 📉下降至 ¥8799.0
📝 [日志] {"time": "2024-01-15 10:30:45", "product": "iPhone 16 Pro", "old_price": 8999.0, "new_price": 8799.0}

🔔 价格变动检测: iPhone 16 Pro ¥8799.0 → ¥9299.0
📧 [邮件通知] iPhone 16 Pro: ¥8799.0 → ¥9299.0 (+5.7%)
📝 [日志] {"time": "2024-01-15 10:30:45", "product": "iPhone 16 Pro", "old_price": 8799.0, "new_price": 9299.0}

一句话解释：价格一变动，所有订阅的渠道自动收到通知，退订的渠道不再收到。工厂模式让添加新渠道超简单。

💪 进阶：常见坑 + 性能小贴士

坑 1：单例模式在多线程环境下可能创建多个实例

# ❌ 错误示例：多线程下不安全
class UnsafeSingleton:
_instance = None

def __new__(cls):
    if cls._instance is None:  # 两个线程可能同时通过这里
        cls._instance = super().__new__(cls)
    return cls._instance

# ✅ 正确示例：加锁保证线程安全
import threading

class SafeSingleton:
_instance = None
_lock = threading.Lock()

def __new__(cls):
    if cls._instance is None:
        with cls._lock:
            if cls._instance is None:  # 双重检查锁定
                cls._instance = super().__new__(cls)
    return cls._instance

坑 2：工厂模式返回的是类不是实例

# ❌ 错误示例：返回了类，没调用 ()
result = animal_factory("dog")
print(result.speak())  # TypeError: Dog.speak() 缺少 self

# ✅ 正确示例：工厂返回的是实例
result = animal_factory("dog")
print(result.speak())  # 汪汪汪

坑 3：观察者模式中观察者引用了已删除的主题

# ❌ 错误示例：观察者持有主题引用，主题删除后出问题
class Observer:
def __init__(self, monitor):
    self.monitor = monitor  # 保存引用
    self.monitor.subscribe(self)  # 订阅

# 如果 monitor 被 del，观察者还存着引用

# ✅ 正确示例：使用弱引用或在主题中处理清理
import weakref

class Monitor:
def __init__(self):
    self._subscribers = []  # 简化版，实际可用 weakref

坑 4：工厂模式硬编码了所有产品

# ❌ 错误示例：每次加新产品都要改工厂
def animal_factory(animal_type):
if animal_type == "dog":
    return Dog()
elif animal_type == "cat":
    return Cat()
# 每次加新动物都要改这里

# ✅ 正确示例：注册制，加新产品不改工厂代码
class FormatterFactory:
_formatters = {}

@classmethod
def register(cls, name, formatter_class):
    cls._formatters[name] = formatter_class

@classmethod
def create(cls, name):
    return cls._formatters[name]()

坑 5：单例模式 init 被多次调用

# ❌ 错误示例：每次 getInstance() 都可能触发 __init__
class BadSingleton:
def __init__(self):
    # 这个打印每次 new 都会执行
    print("初始化中...")

# ❌ 错误示例：没有加 _initialized 标记

# ✅ 正确示例：加标志位防止重复初始化
class GoodSingleton:
_instance = None

def __new__(cls):
    if cls._instance is None:
        cls._instance = super().__new__(cls)
        cls._instance._initialized = False
    return cls._instance

def __init__(self):
    if self._initialized:
        return
    self._initialized = True
    print("初始化完成（只一次）")

性能小贴士：观察者模式用列表推导式批量订阅

# 普通方式：一个个订阅
monitor.subscribe(ChannelFactory.create("email"))
monitor.subscribe(ChannelFactory.create("sms"))
monitor.subscribe(ChannelFactory.create("log"))

# 高效方式：批量订阅
channels = [ChannelFactory.create(t) for t in ["email", "sms", "log"]]
for ch in channels:
monitor.subscribe(ch)

调试技巧：用 repr 看清楚对象

class Singleton:
_instance = None

def __new__(cls):
    if cls._instance is None:
        cls._instance = super().__new__(cls)
    return cls._instance

def __repr__(self):
    return f"<Singleton id={id(self)}>"


# 调试时打印对象看清楚了
s1 = Singleton()
s2 = Singleton()
print(f"s1: {s1}")
print(f"s2: {s2}")
print(f"s1 is s2: {s1 is s2}")

✏️ 练习题

练习 1（2 分钟）：验证单例模式

• 输入：连续创建 5 个 Config 实例
• 预期输出：所有实例的 id 相同
• 提示：用 id() 函数查看内存地址

练习 2（2 分钟）：添加新动物类型

• 输入：在项目 1 的 animal_factory 中添加 "pig" → 猪叫 "哼哼哼"
• 预期输出：animal_factory("pig").speak() 输出 "哼哼哼"
• 提示：在字典里加一行 "pig": Pig 就行

练习 3（3 分钟）：给 Logger 加 XML 格式

• 输入：参考 JsonFormatter 写一个 XmlFormatter
• 预期输出：Logger("xml").log("INFO", "测试") 输出 XML 字符串
• 提示：XML 格式是 <log><level>INFO</level><message>测试</message></log>

练习 4（5 分钟）：价格监控加微信通知

• 输入：添加 WechatChannel，输出 "📱 [微信通知] {product} 最新价: ¥{new_price}"
• 预期输出：订阅后价格变动，微信渠道能收到
• 提示：先在 ChannelFactory 注册新渠道类

练习 5（5 分钟）：找 bug

• 输入：以下代码运行后会输出什么？为什么？
• 代码：

class BadSingleton:
def __new__(cls):
    return super().__new__(cls)

a = BadSingleton()
b = BadSingleton()
print(a is b)

• 预期输出：分析输出结果，解释为什么单例没生效
• 提示：没有保存实例，所以每次 new 都是新对象

作业：做一个「学习提醒小工具」

需求描述：做一个学习计划提醒工具，帮助你按时学习。

功能点：
1. 单例模式：一个全局的 StudyPlanner 实例管理所有学习计划
2. 工厂模式：ReminderFactory 创建不同类型的提醒（邮件/微信/控制台）
3. 观察者模式：学习计划状态变化（开始/完成/延期）自动通知所有订阅渠道

具体实现：
- StudyPlan 类有：课程名、计划日期、状态（pending/completed/delayed）
- 状态变化时，自动通过已订阅的渠道发送提醒
- 支持添加/删除/修改学习计划

加分项：
1. 用装饰器或魔术方法让代码更优雅
2. 加一个「统计功能」：显示本周完成/未完成计划数

验收标准：
- 能跑起来
- 操作学习计划后能看到提醒输出
- 代码有注释

📚 总结 + 资源

本文学了 3 个核心点：
1. 单例模式：全局唯一实例，用 __new__ 控制，保证一致性好
2. 工厂模式：把"创建对象"封装起来，换产品不用改调用代码
3. 观察者模式：一变自动通知多者，用订阅机制解耦

延伸学习资源：

1. 📖 《Head First 设计模式》—— 最适合新手的模式书，用图和故事讲清楚
2. 🎬 Python Design Patterns Tutorial —— 有代码有图解，中文友好
3. 📝 Python 官方文档 - Design Patterns —— 官方示例

互动钩子：

你在项目中用过这三种设计模式吗？比如日志系统用单例、创建对象用工厂、用户操作用观察者？评论区聊聊你的经验，老粉优先回复！

👉 下一篇我们将把这三个模式组合起来，手把手带你写一个 MVC 迷你框架，真正体验「模式组合」的力量。敬请期待！