20、观察者模式进阶：异步观察者模式（使用 asyncio 实现非阻塞通知）

好，咱们继续聊观察者模式。前面讲的是同步版本，说白了就是「一个通知，全员响应，响应完了才结束」。这在简单场景下没问题，但一旦观察者里有个慢操作——比如写日志、发邮件、调远程 API——整个流程就被卡住了。

我几年前接手过一个监控系统，主题一推送状态变化，十几个观察者依次执行。其中有个观察者负责把数据写入 Elasticsearch，偶尔网络抖动，写入慢了几秒，结果整个事件循环都堵死了。用户那边点了个按钮，界面卡住不动，体验极差。

那怎么办？用异步。Python 的 asyncio 就是干这个的。它能让观察者各自跑自己的协程，互不阻塞。说白了，就是「你发你的通知，我干我的活，谁也别等谁」。

为什么需要异步观察者？

同步模式有个硬伤：通知顺序 = 执行顺序 = 阻塞顺序。你想想看，如果观察者 A 要花 3 秒，观察者 B 要花 2 秒，那总耗时至少 5 秒。这还只是两个，要是二十个呢？

异步模式的核心思路是：主题只管「发布事件」，不等待观察者处理完。每个观察者拿到事件后，自己调度协程去执行。这样主题可以立刻返回，继续处理下一个事件。

基础实现：用 asyncio 改造观察者

先看一个最简单的异步观察者模式。我习惯把观察者的更新方法改成协程，主题的 notify 方法也改成异步。

import asyncio
from typing import List, Callable, Awaitable

class AsyncObserver:
    """异步观察者基类"""
    async def update(self, data):
        raise NotImplementedError

class AsyncSubject:
    """异步主题"""
    def __init__(self):
        self._observers: List[AsyncObserver] = []

    def attach(self, observer: AsyncObserver):
        self._observers.append(observer)

    def detach(self, observer: AsyncObserver):
        self._observers.remove(observer)

    async def notify(self, data):
        # 并发执行所有观察者的 update 协程
        tasks = [observer.update(data) for observer in self._observers]
        await asyncio.gather(*tasks)

这里用了 asyncio.gather，它会并发启动所有协程。注意，不是顺序等待，而是一起跑。哪个先完成无所谓，反正互不依赖。

实际案例：日志监控系统

我拿之前那个监控系统举例。主题是「日志事件源」，观察者有：写入数据库、发送告警邮件、更新实时仪表盘。

class LogEventSubject(AsyncSubject):
    """日志事件主题"""
    pass

class DatabaseWriter(AsyncObserver):
    async def update(self, data):
        print(f"[DB] 写入日志: {data}")
        await asyncio.sleep(2)  # 模拟写入耗时
        print(f"[DB] 写入完成")

class EmailAlert(AsyncObserver):
    async def update(self, data):
        print(f"[Email] 发送告警: {data}")
        await asyncio.sleep(3)  # 模拟发送耗时
        print(f"[Email] 发送完成")

class DashboardUpdater(AsyncObserver):
    async def update(self, data):
        print(f"[Dashboard] 更新面板: {data}")
        await asyncio.sleep(1)
        print(f"[Dashboard] 更新完成")

async def main():
    subject = LogEventSubject()
    subject.attach(DatabaseWriter())
    subject.attach(EmailAlert())
    subject.attach(DashboardUpdater())

    print("=== 开始通知 ===")
    await subject.notify("系统异常: CPU 使用率 95%")
    print("=== 通知结束 ===")

asyncio.run(main())

运行结果会怎样？你会看到三个观察者几乎同时开始执行，然后各自在 1~3 秒后完成。总耗时大约 3 秒（最慢的那个），而不是 1+2+3=6 秒。这就是并发的威力。

异步观察者的核心流程图

下面这张图展示了异步观察者模式的核心流程。注意看「并发执行」这个关键步骤。

进阶技巧：错误隔离与超时控制

异步模式有个坑：如果一个观察者抛异常，会不会影响其他观察者？默认情况下，asyncio.gather 会传播异常。也就是说，一个挂了，整个 gather 都挂。

我曾经踩过这个坑。有个观察者因为网络超时抛了异常，结果所有观察者都被取消了。日志没写进去，告警也没发出去。那怎么办？用 return_exceptions=True。

async def notify(self, data):
    tasks = [observer.update(data) for observer in self._observers]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"观察者 {i} 执行失败: {result}")

这样，异常会被收集起来，不会中断其他观察者。你还可以针对每个异常做重试或降级处理。

更灵活的实现：回调式异步观察者

有时候你不想定义类，只想注册一个简单的异步函数。那可以用回调式实现。

from typing import List, Callable, Awaitable

class AsyncCallbackSubject:
    def __init__(self):
        self._callbacks: List[Callable[[any], Awaitable[None]]] = []

    def attach(self, callback: Callable[[any], Awaitable[None]]):
        self._callbacks.append(callback)

    async def notify(self, data):
        tasks = [cb(data) for cb in self._callbacks]
        await asyncio.gather(*tasks, return_exceptions=True)

# 使用
async def log_to_file(data):
    async with aiofiles.open("log.txt", "a") as f:
        await f.write(f"{data}\n")

async def send_alert(data):
    async with aiohttp.ClientSession() as session:
        await session.post("http://alert-api", json={"msg": data})

subject = AsyncCallbackSubject()
subject.attach(log_to_file)
subject.attach(send_alert)

这种方式更轻量，适合快速集成。我个人的习惯是：如果观察者逻辑简单，用回调；如果逻辑复杂、需要状态管理，用类。

性能对比：同步 vs 异步

场景	同步模式耗时	异步模式耗时	提升比例
3 个观察者（1s, 2s, 3s）	6s	~3s	50%
5 个观察者（各 2s）	10s	~2s	80%
10 个观察者（混合 1~5s）	~30s	~5s	83%

数据很直观：观察者越多，异步优势越明显。尤其是当观察者里有网络 IO 或磁盘 IO 时，异步几乎是必选项。

避坑指南

最后分享几个我踩过的坑：

• 不要混用同步和异步：如果一个观察者是同步的，它会阻塞整个事件循环。我建议要么全异步，要么全同步，别搞混搭。
• 注意协程泄漏：如果观察者里创建了协程但没有 await，它会被静默丢弃。用 asyncio.create_task 时要确保有引用。
• 超时处理：用 asyncio.wait_for 给每个观察者设置超时，防止某个观察者卡死。

async def notify_with_timeout(self, data, timeout=5):
    tasks = []
    for observer in self._observers:
        task = asyncio.create_task(observer.update(data))
        tasks.append(task)
    # 等待所有任务，但每个任务最多等 timeout 秒
    done, pending = await asyncio.wait(tasks, timeout=timeout)
    for task in pending:
        task.cancel()  # 超时的任务取消掉
        print(f"观察者超时，已取消")

嗯，异步观察者模式就聊到这儿。它不是什么高深的技术，但用好了能大幅提升系统的响应性和吞吐量。你想想看，一个通知发出去，所有观察者各干各的，互不干扰——这才是观察者模式该有的样子。