观察者模式

代码实现（Python版本）：

class Observer:def update(self, message):raise NotImplementedError("You should implement this method.")class Subject:def __init__(self):self._observers = []def attach(self, observer):"""附加观察者"""self._observers.append(observer)def detach(self, observer):"""移除观察者"""self._observers.remove(observer)def notify(self, message):"""通知所有观察者"""for observer in self._observers:observer.update(message)class ConcreteSubject(Subject):def __init__(self):super().__init__()self._state = Nonedef change_state(self, new_state):"""改变状态并通知观察者"""self._state = new_stateprint(f"Subject state changed to: {self._state}")self.notify(self._state)class ConcreteObserver(Observer):def __init__(self, name):self._name = namedef update(self, message):"""接收通知并处理"""print(f"Observer {self._name} received message: {message}")if __name__ == "__main__":subject = ConcreteSubject()observer1 = ConcreteObserver("Observer 1")observer2 = ConcreteObserver("Observer 2")subject.attach(observer1)  # 观察者1注册到主题subject.attach(observer2)  # 观察者2注册到主题subject.change_state("State 1")  # 改变状态并通知观察者subject.change_state("State 2")  # 改变状态并通知观察者subject.detach(observer1)  # 移除观察者1subject.change_state("State 3")  # 观察者2仍然会收到通知

当您运行这个程序时，您将看到观察者接收到主题状态变化的通知，具体输出如下：

Subject state changed to: State 1
Observer Observer 1 received message: State 1
Observer Observer 2 received message: State 1
Subject state changed to: State 2
Observer Observer 1 received message: State 2
Observer Observer 2 received message: State 2
Subject state changed to: State 3
Observer Observer 2 received message: State 3

观察者模式又叫发布/订阅模式（Publish/Subscribe)模式。

监听设计模式还可以分为推模式和拉模式。
下面是使用Python实现监听设计模式（观察者模式）的推模式和拉模式的示例代码。我们将定义一个主题（Subject）和观察者（Observer），并分别实现这两种模式。

推模式实现
在推模式中，主题在状态变化时主动将数据推送给观察者。

class Observer:def update(self, message):raise NotImplementedError("You should implement this method.")class Subject:def __init__(self):self._observers = []def attach(self, observer):self._observers.append(observer)def detach(self, observer):self._observers.remove(observer)def notify(self, message):for observer in self._observers:observer.update(message)class ConcreteSubject(Subject):def __init__(self):super().__init__()self._state = Nonedef change_state(self, new_state):self._state = new_stateprint(f"Subject state changed to: {self._state}")self.notify(self._state)  # 推模式：将状态推送给观察者class ConcreteObserver(Observer):def __init__(self, name):self._name = namedef update(self, message):print(f"Observer {self._name} received message: {message}")if __name__ == "__main__":subject = ConcreteSubject()observer1 = ConcreteObserver("Observer 1")observer2 = ConcreteObserver("Observer 2")subject.attach(observer1)subject.attach(observer2)subject.change_state("State 1")subject.change_state("State 2")

拉模式实现
在拉模式中，观察者在需要数据时主动请求主题提供数据。

class Observer:def update(self, subject):raise NotImplementedError("You should implement this method.")class Subject:def __init__(self):self._observers = []self._state = Nonedef attach(self, observer):self._observers.append(observer)def detach(self, observer):self._observers.remove(observer)def set_state(self, new_state):self._state = new_stateprint(f"Subject state changed to: {self._state}")self.notify()  # 通知观察者def notify(self):for observer in self._observers:observer.update(self)  # 拉模式：观察者请求状态class ConcreteSubject(Subject):def get_state(self):return self._stateclass ConcreteObserver(Observer):def __init__(self, name):self._name = namedef update(self, subject):state = subject.get_state()  # 拉模式：主动请求状态print(f"Observer {self._name} received state: {state}")if __name__ == "__main__":subject = ConcreteSubject()observer1 = ConcreteObserver("Observer 1")observer2 = ConcreteObserver("Observer 2")subject.attach(observer1)subject.attach(observer2)subject.set_state("State 1")subject.set_state("State 2")

案例，登录异常的监测与提醒

import time
from collections import defaultdict# 抽象观察者
class Observer:def update(self, message):raise NotImplementedError("You should implement this method.")# 抽象主题
class Subject:def __init__(self):self._observers = []def attach(self, observer):self._observers.append(observer)def detach(self, observer):self._observers.remove(observer)def notify(self, message):for observer in self._observers:observer.update(message)# 具体主题
class LoginMonitor(Subject):def __init__(self):super().__init__()self.login_attempts = defaultdict(list)  # 存储登录尝试self.threshold = 3  # 失败尝试阈值def record_login_attempt(self, username, ip_address, success):"""记录登录尝试"""timestamp = time.time()self.login_attempts[username].append((timestamp, ip_address, success))self.check_for_anomalies(username)def check_for_anomalies(self, username):"""检查异常登录行为"""attempts = self.login_attempts[username]failed_attempts = [attempt for attempt in attempts if not attempt[2]]# 检查连续失败的登录尝试if len(failed_attempts) >= self.threshold:last_failed_time = failed_attempts[-1][0]if time.time() - last_failed_time < 300:  # 5分钟内的失败尝试self.notify(f"Alert: Multiple failed login attempts detected for user '{username}'.")# 具体观察者
class EmailAlert(Observer):def update(self, message):print(f"Email Alert: {message}")class SMSAlert(Observer):def update(self, message):print(f"SMS Alert: {message}")# 示例使用
if __name__ == "__main__":monitor = LoginMonitor()# 添加观察者email_alert = EmailAlert()sms_alert = SMSAlert()monitor.attach(email_alert)monitor.attach(sms_alert)# 模拟登录尝试monitor.record_login_attempt("user1", "192.168.1.1", False)monitor.record_login_attempt("user1", "192.168.1.1", False)monitor.record_login_attempt("user1", "192.168.1.1", False)  # 第三次失败，触发提醒monitor.record_login_attempt("user1", "192.168.1.1", True)   # 成功登录

在之前的代码示例中，login_attempts是一个 defaultdict，用于存储每个用户的登录尝试记录。具体来说，它的结构是一个字典，其中键是用户名，值是一个列表，列表中包含该用户的每次登录尝试的详细信息。

详细解释：
类型：defaultdict(list) 是 Python 的一个字典子类，允许在访问不存在的键时自动创建一个空列表。这样可以方便地向列表中添加元素，而不需要先检查键是否存在。

用途：login_attempts 用于记录每个用户的登录尝试，包括时间戳、IP 地址和登录结果（成功或失败）。例如，记录的每个登录尝试可以是一个元组，包含 (timestamp, ip_address, success)。