当前位置：网站首页>观察者（observer）模式（一）

观察者（observer）模式（一）

2022-07-31 01:21:00 【晓之木初】

1. 引子

1.1 不完美的实现方案

公司业务发展壮大，集群监控也逐渐走向自动化：上报集群重要指标，实时监控集群状态，异常时进行自动告警
老大说：你去写一个告警程序，集群状态异常时，以短信和电话的形式通知运维人员

新来的可能会这样写（程序简化了，能表述出编程思路就行）：

public class AlertApplication {
      
    private final MessageAlarm messageAlarm;
    private final TelephoneAlarm telephoneAlarm;

    public AlertApplication(MessageAlarm messageAlarm, TelephoneAlarm telephoneAlarm) {
      
        this.messageAlarm = messageAlarm;
        this.telephoneAlarm = telephoneAlarm;
    }

    // 收到来自实时监控的指标数据，根据阈值确定是否需要进行告警
    public void metricData(double memory, double cpu) {
      
        // 打印日志
        System.out.printf("集群内存使用: %.2fGB, cpu使用率: %.2f%%\n", memory, cpu * 100);
        if (memory >= Threshold.MAX_MEMORY.getThreshold()) {
      
            String msg = String.format("集群内存使用量: %.2fGB, 超过阈值: %.2fGB", memory, Threshold.MAX_MEMORY.getThreshold());
            messageAlarm.sendMessage(msg);
            telephoneAlarm.ringUp(msg);
        }
        if (cpu >= Threshold.MAX_CPU.getThreshold()) {
      
            String msg = String.format("集群cpu使用率: %.2f%%, 超过阈值： %.2f%%", cpu * 100, Threshold.MAX_CPU.getThreshold() * 100);
            messageAlarm.alert(msg);
            telephoneAlarm.alert(msg);
        }
    }

}

enum Threshold {
      
    MAX_MEMORY(100),
    MAX_CPU(0.8);
    private double threshold;

    Threshold(double threshold) {
      
        this.threshold = threshold;
    }

    public double getThreshold() {
      
        return this.threshold;
    }
}

class MessageAlarm {
      
    public void alert(String msg) {
      
        System.out.println("短信告警: " + msg);
    }
}

class TelephoneAlarm {
      
    public void alert(String msg) {
      
        System.out.println("电话告警: " + msg);
    }
}

编写主程序，启动AlertApplication

public class Main {
      
    public static void main(String[] args) {
      
        AlertApplication application = new AlertApplication(new MessageAlarm(), new TelephoneAlarm());
        application.metricData(64.8, 0.45);
        application.metricData(120.26, 0.97);
    }
}

执行结果如下：

1.2 存在的问题

有经验的同事对这段代码做了如下评价（实际来自博客：Observer Pattern | Set 1 (Introduction)）：
- AlertApplication持有具体Alarm对象的引用，可以访问到超出其需要的更多额外信息，即使它只需要调用这些Alarm对象的alert()方法。（违反了迪米特原则？菜鸟不是很懂）
- 调用Alarm对象的alert(String msg)方法，是在使用具体对象共享数据，而非使用接口共享数据。这违背了一个重要的设计原则：
  Program to interfaces, not implementations
- AlertApplication与Alarm对象紧耦合，如果要添加或移除Alarm对象，需要修改AlertApplication，这明显违背了开闭原则
针对以上问题，自己体会最深的就是违反了开闭原则，代码不易维护

2. 使用observer模式

observer模式属于行为设计模式，其定义如下：
observer模式定义了对象之间的一对多依赖，当一个对象的状态发生变化，会自动通知并更新其他依赖对象
根据上面的场景，我们可以分析出：
- AlertApplication与Alarm对象之间存在一对多关系（one-to-many relationship），AlertApplication是one，Alarm对象是many
- 当集群处于异常状态时，AlertApplication需要自动调用（通知）Alarm对象
- 换句话说，Alarm对象是否执行告警动作，依赖于AlertApplication对象的状态是否发生改变（这里是指是否达到告警阈值）
不难发现，上述场景可以使用observer模式

2.1 概念解读

observer模式中，将一对多关系中的one叫做Subject（主题），many叫做Observer
但是，这里的Observer不能主动获取消息，而是等待Subject向他推送消息
- 就像医院排号看病一样，病人如果频繁询问医生或者护士现在多少号了，那治疗工作就没法进行下去了
- 需要通过叫号器，显示当前进度、通知下一个病人进入诊室就诊
- 这时，叫号器就是Subject，病人就是Observer
其实，observer模式还有很多其他的称呼：如发布-订阅(Publish/Subscribe)模式、模型-视图(Model/View)模式、源-监听器(Source/Listener)模式或从属者(Dependents)模式
我们熟悉的Java GUI中的各种Listener，就是源-监听器模式（简称事件监听模式）
微信公众号的订阅、银行活动推送等，使用发布-订阅模式来描述更加简洁易懂

2.2 真实应用场景

observer模式，在GUI工具包和事件监听器中大量使用。例如，java AWT中的button(Subject)和 ActionListener(observer) 是用观察者模式构建的。
社交媒体、RSS 提要、电子邮件订阅、公众号等，使用observer模式向关注或订阅的用户推送最新消息
手机应用商店，应用如果有更新，将使用observer模式通知所有用户

2.3 UML图

observer模式的UML图如下：
Subject（抽象主题）：Subject一般为接口或抽象类，提供添加、删除、通知observer对象的三个抽象方法
ConcreteSubject（具体主题）：内部使用集合存储注册的observer，实现Subject中的抽象方法，以便在内部状态发生变化时，通知所有注册过的observer对象。
Observer（抽象观察者）：Observer一般为借口或抽象类，为Subject提供通知自己的notify()方法
- 还可以定义为update()方法，二者都是subject向observer传递信息的接口）
ConcreteObserver（具体观察者）：实现notify()方法，在Subject状态边变化时，做出相应的反应

2.4 使用observer模式实现需求

定义Subject接口：

public interface Subject {
      
    void addObserver(Observer observer);

    void deleteObserver(Observer observer);

    void notifyObservers(double cpu, double memory);
}

定义Observer接口：

public interface Observer {
      
    void notify(double cpu, double memory);
}

实现AlertApplication对应的Subject：

public class AlertApplicationSubject implements Subject {
      
    private final List<Observer> observers;

    public AlertApplicationSubject() {
      
        this.observers = new ArrayList<>();
    }

    @Override
    public void addObserver(Observer observer) {
      
        if (observer == null && observers.contains(observer)) {
      
            return;
        }
        observers.add(observer);
    }

    @Override
    public void deleteObserver(Observer observer) {
      
        if (observer == null) {
      
            return;
        }
        observers.remove(observer);
    }

    @Override
    public void notifyObservers(double cpu, double memory) {
      
        System.out.printf("集群当前cpu使用率: %.2f%%, 内存使用量: %.2fGB\n", cpu * 100, memory);
        // 当cpu或memory超过阈值，通知observer
        // observer接收到信息后，自动告警
        if (cpu > 0.8 || memory > 100) {
      
            for (Observer observer : observers) {
      
                observer.notify(cpu, memory);
            }
        }
    }
}

实现短信告警、电话告警两种observer：

public class MessageAlarmObserver implements Observer{
      
    @Override
    public void notify(double cpu, double memory) {
      
        if (cpu > 0.8 ) {
      
            System.out.printf("短信告警: 集群cpu使用率%.2f%%, 超过阈值80%%\n", cpu*100);
        }
        if (memory > 100) {
      
            System.out.printf("短信告警: 集群内存使用量%.2fGB, 超过阈值100GB\n", memory);
        }
    }
}

public class PhoneAlarmObserver implements Observer{
      
    @Override
    public void notify(double cpu, double memory) {
      
        if (cpu > 0.8 ) {
      
            System.out.printf("电话告警: 集群cpu使用率%.2f%%, 超过阈值80%%\n", cpu * 100);
        }
        if (memory > 100) {
      
            System.out.printf("电话告警: 集群内存使用量%.2fGB, 超过阈值100GB\n", memory);
        }
    }
}

使用observer模式实现的集群监控告警程序

public class Main {
      
    public static void main(String[] args) {
      
        Subject subject = new AlertApplicationSubject();

        Observer messageAlarm = new MessageAlarmObserver();
        Observer phoneAlarm = new PhoneAlarmObserver();
        subject.addObserver(messageAlarm);
        subject.addObserver(phoneAlarm);

        // 采集集群监控数据
        subject.notifyObservers(0.6, 45);
        subject.notifyObservers(0.5, 127);
    }
}

执行结果如下：

2.5 observer模式的优缺点

2.5.1 优点

自己的理解： 相对第一个版本的代码实现，使用observer模式符合迪米特原则、接口编程原则、开闭原则
专业的评价： observer模式实现了交互对象之间的松耦合
松耦合的对象可以灵活应对不断变化的需求，且交互对象无需拥有其他对象的额外信息
松耦合具体指：
- Subject只需要知道observer对象实现了Observer接口
- 添加或删除observer无需修改Subject
- 可以相互独立地重用subject和observer对象（例如，可以直接调用observer的相关方法）

2.5.2 缺点

由于需要显式地注册和注销observer，Lapsed listener problem 将导致内存泄漏

关于Lapsed listener problem

问题一：内存泄漏
- 在observer模式中，subject持有对已经注册的observer的强引用，使得observer不会被垃圾回收
- 如果observer不再需要接收subject的通知，但却没有正确地从subject中注销，则将发生内存泄漏
- 此时，subject持有对observer的强引用，observer及其引用其他对象都将无法被垃圾回收
问题二：性能下降
- 不感兴趣的observer没有从subject中注册自己，将增加subject发送消息的工作量，导致性能下降
解决办法：
- subject持有observer的弱引用，而非强引用，使得observer不再工作后（只被弱引用关联），无需注销就能被垃圾回收
- 自己的疑问：这不靠谱啊，为了不让observer被垃圾回收，不得另外找个地方给它创建一个强引用？不然，不知啥时候就被垃圾回收了

3. Java内置的observer模式

3.1 两种模式（推模式 vs 拉模式）

推模式

上面的代码实现中，subject知道observer需要哪些数据，并通过notify()方法主动向observer传递数据，属于push模式（推模式）
这样的设计使得observer难以复用，因为observer的notify()方法需要根据实际需求定义参数，很可能无法兼顾其他需求场景
例如，上面的示例代码，observer只适合用于监控cpu和memory的场景。如果切换成发送广告邮件的场景，则无法适用

拉模式

既然subject无法准确判断observer需要什么数据，那干脆就把自身作为入参，让observer按需按需获取
这样的模式，被叫做pull模式（拉模式）
Java内置的observer模式，在我个人看来是拉模式 + 推模式的完美结合

3.2 java内置的observer模式

Java提供了Observable类，对应Subject；Observer接口，对应观察者

其中，Observable类非常简单

包含一个存储observer的Vector对象obs，一个标识状态是否变化的布尔值changed
提供了用于添加、删除、计数observer的同步方法，用于更新、重置、获取状态的同步方法

但是，其notifyObservers()却只是局部同步，并非整体同步 —— 这样的设计存在问题？欢迎讨论

public void notifyObservers(Object arg) {
        
    Object[] arrLocal;

    synchronized (this) {
        
        if (!changed)
            return;
        arrLocal = obs.toArray();
        clearChanged();
    }
    // 如注释说的一样，没有对这部分代码进行同步，容易出现：
    // 新添加的observer无法收到正在进行的通知，最近移除的observer会错误地收到通知
    for (int i = arrLocal.length-1; i>=0; i--)
        ((Observer)arrLocal[i]).update(this, arg);
}

Observer接口只有一个update(Observable o, Object arg)方法

public interface Observer {
      
    void update(Observable o, Object arg);
}

这样的设计既可以使用拉模式，让observer主动从Subject获取数据；又可以基于Object arg使用推模式，主动向observer传递数据

3.3 使用实战

使用Java自带的observer模式，实现看病叫号的需求

继承Observable类实现叫号器

public class Caller extends Observable {
      
    private final int room; // 诊室
    private int number; // 记录当前的就诊序号

    public Caller(int room) {
      
        super(); // 初始化存储observer的Vector
        this.room = room;
    }

    public void call(int number) {
      
        // 就诊序号发生变化，开始叫号
        if (number != this.number) {
      
            this.number = number; // 记录最新的就诊序号
            setChanged(); // 将状态更新为true，表示状态发生变化，以触发notifyObservers()方法
            notifyObservers(); // 调用notifyObservers()通知就诊的病人
        }
    }

    public int getNumber() {
      
        return number;
    }

    public int getRoom() {
      
        return room;
    }
}

实现Observer接口，创建Patient类

public class Patient implements Observer {
      
    private final int number;
    private final String name;

    public Patient(int number, String name) {
      
        this.number = number;
        this.name = name;
    }

    @Override
    public void update(Observable o, Object arg) {
      
        // 获取诊室号和就诊序号，如果是自己则做出回应
        int room = ((Caller) o).getRoom();
        int number = ((Caller) o).getNumber();
        if (number == this.number) {
      
            System.out.printf("我是%d号病人: %s，轮到我去%d诊室就诊\n", number, name, room);
        }
    }
}

测试程序

public class Main {
      
    public static void main(String[] args) {
      
        Caller caller = new Caller(7);
        // 添加已经到场的病人
        Patient patient1 = new Patient(3, "张三");
        caller.addObserver(patient1);
        Patient patient2 = new Patient(1, "王二");
        caller.addObserver(patient2);
        Patient patient3 = new Patient(4, "李四");
        caller.addObserver(patient3);

        // 开始叫号
        caller.call(1);
        caller.call(2); // 没有对应的病人，无任何响应
        caller.call(3);
    }
}