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);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52

• 编写主程序，启动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
  3
  4
  5
  6
  7

• 执行结果如下：
  

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 真实应用场景

1. observer模式，在GUI工具包和事件监听器中大量使用。例如，java AWT中的button(Subject)和 ActionListener(observer) 是用观察者模式构建的。
2. 社交媒体、RSS 提要、电子邮件订阅、公众号等，使用observer模式向关注或订阅的用户推送最新消息
3. 手机应用商店，应用如果有更新，将使用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);
  }
  1
  2
  3
  4
  5
  6
  7

• 定义Observer接口：

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

• 实现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);
              }
          }
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35

• 实现短信告警、电话告警两种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);
          }
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23

• 使用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);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14

• 执行结果如下：
  

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

1. 问题一：内存泄漏

   • 在observer模式中，subject持有对已经注册的observer的强引用，使得observer不会被垃圾回收
   • 如果observer不再需要接收subject的通知，但却没有正确地从subject中注销，则将发生内存泄漏
   • 此时，subject持有对observer的强引用，observer及其引用其他对象都将无法被垃圾回收

2. 问题二：性能下降

   • 不感兴趣的observer没有从subject中注册自己，将增加subject发送消息的工作量，导致性能下降

3. 解决办法：

   • 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);
    }
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14

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

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

• 这样的设计既可以使用拉模式，让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;
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26

• 实现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);
          }
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19

• 测试程序

  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);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17

• 执行结果如下：
  

• 通过这个示例程序的顿悟： Subject和observer之间的一对多关系，并非是对应多个不同类型的observer，同一类型的多个observer也行

3.5 其他

• 博客The Observer Pattern in Java还提出，Observer接口不完美，且Observable类容易开发者重写方法而破坏线程安全
• 所以，在JDK 9中，Observer接口被弃用，推荐基于ProperyChangeListener接口实现
• 由于笔者使用的是JDK 8，所以无法验证，后续有机会可以体验一下ProperyChangeListener接口

4. 参考链接

• Observer Pattern | Set 1 (Introduction) & Observer Pattern | Set 2 (Implementation)
• 推模式、拉模式：《JAVA与模式》之观察者模式
• PropertyChangeSupport：The Observer Pattern in Java