我们常把 23 种经典的设计模式分为三类：创建型、结构型、行为型。前面我们已经学习了创建型和结构型，从今天起，我们开始学习行为型设计模式。我们知道，创建型设计模式主要解决“对象的创建”问题，结构型设计模式主要解决“类或对象的组合或组装”问题，那行为型设计模式主要解决的就是“类或对象之间的交互”问题。

行为型设计模式比较多，有 11 个，几乎占了 23 种经典设计模式的一半。它们分别是：观察者模式、模板模式、策略模式、职责链模式、状态模式、迭代器模式、访问者模式、备忘录模式、命令模式、解释器模式、中介模式。

今天，我们学习第一个行为型设计模式，也是在实际的开发中用得比较多的一种模式：观察者模式。根据应用场景的不同，观察者模式会对应不同的代码实现方式：有同步阻塞的实现方式，也有异步非阻塞的实现方式；有进程内的实现方式，也有跨进程的实现方式。今天我会重点讲解原理、实现、应用场景。下一节课，我会带你一块实现一个基于观察者模式的异步非阻塞的 EventBus，加深你对这个模式的理解。

1、什么是观察者模式

观察者模式（Observer Design Pattern）也被称为发布订阅模式（Publish-Subscribe Design Pattern）。在 GoF 的《设计模式》一书中，它的定义是这样的：

Define a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically.

翻译成中文就是：在对象之间定义一个一对多的依赖，当一个对象状态改变的时候，所有依赖的对象都会自动收到通知。

一般情况下，被依赖的对象叫作被观察者（Observable），依赖的对象叫作观察者（Observer）。不过，在实际的项目开发中，这两种对象的称呼是比较灵活的，有各种不同的叫法，比如：Subject-Observer、Publisher-Subscriber、Producer-Consumer、EventEmitter-EventListener、Dispatcher-Listener。不管怎么称呼，只要应用场景符合刚刚给出的定义，都可以看作观察者模式。

2、为什么要使用观察者模式

观察者模式的应用场景非常广泛，小到代码层面的解耦，大到架构层面的系统解耦，再或者一些产品的设计思路，都有这种模式的影子，比如，邮件订阅、RSS Feeds，本质上都是观察者模式。

一个对象状态改变给其他对象通知的问题，而且要考虑到易用和低耦合，保证高度的协作。
一个对象（目标对象）的状态发生改变，所有的依赖对象（观察者对象）都将得到通知，进行广播通知。

如果需要在系统中创建一个触发链，A对象的行为将影响B对象，B对象的行为将影响C对象……，可以使用观察者模式创建一种链式触发机制。

3、例子

3.1、同步阻塞（简单）

image.png

  1. import java.util.ArrayList;
  2. import java.util.List;
  4. public class Main {
  5.     public static void main(String[] args) {
  6.         ConcreteSubject subject = new ConcreteSubject();
  8.         subject.registerObserver(new ConcreteObserverOne());
  9.         subject.registerObserver(new ConcreteObserverTwo());
  11.         subject.notifyObservers(new Message());
  12.     }
  13. }
  15. class Message {
  16.     Integer code;
  17.     String msg;
  18. }
  20. interface Subject {
  21.     void registerObserver(Observer observer);
  23.     void removeObserver(Observer observer);
  25.     void notifyObservers(Message message);
  26. }
  28. interface Observer {
  29.     void update(Message message);
  30. }
  32. class ConcreteSubject implements Subject {
  33.     private List<Observer> observers = new ArrayList<Observer>();
  35.     @Override
  36.     public void registerObserver(Observer observer) {
  37.         observers.add(observer);
  38.     }
  40.     @Override
  41.     public void removeObserver(Observer observer) {
  42.         observers.remove(observer);
  43.     }
  45.     @Override
  46.     public void notifyObservers(Message message) {
  47.         for (Observer observer : observers) {
  48.             observer.update(message);
  49.         }
  50.     }
  52. }
  54. class ConcreteObserverOne implements Observer {
  55.     @Override
  56.     public void update(Message message) {
  57.         //TODO: 获取消息通知，执行自己的逻辑...
  58.         System.out.println("ConcreteObserverOne is notified.");
  59.     }
  60. }
  62. class ConcreteObserverTwo implements Observer {
  63.     @Override
  64.     public void update(Message message) {
  65.         //TODO: 获取消息通知，执行自己的逻辑...
  66.         System.out.println("ConcreteObserverTwo is notified.");
  67.     }
  68. }

image.png

3.2、异步非阻塞（中等）

对于异步非阻塞观察者模式，如果只是实现一个简易版本，不考虑任何通用性、复用性，实际上是非常容易的。

有两种实现方式。其中一种是：在每个观察者的 update() 函数中创建一个新的线程执行代码逻辑；另一种是：在被观察者发布消息函数中使用线程池来执行每个观察者的 update() 函数。

第一种

import java.util.ArrayList;
import java.util.List;

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

        subject.registerObserver(new ConcreteObserverOne());
        subject.registerObserver(new ConcreteObserverTwo());

        subject.notifyObservers(new Message());
    }
}

class Message {
    Integer code;
    String msg;
}

interface Subject {
    void registerObserver(Observer observer);

    void removeObserver(Observer observer);

    void notifyObservers(Message message);
}

interface Observer {
    void update(Message message);
}

class ConcreteSubject implements Subject {
    private List<Observer> observers = new ArrayList<Observer>();

    @Override
    public void registerObserver(Observer observer) {
        observers.add(observer);
    }

    @Override
    public void removeObserver(Observer observer) {
        observers.remove(observer);
    }

    @Override
    public void notifyObservers(Message message) {
        for (Observer observer : observers) {
            observer.update(message);
        }
    }

}

class ConcreteObserverOne implements Observer {
    @Override
    public void update(Message message) {
        // 获取消息通知，执行自己的逻辑...
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("ConcreteObserverOne is notified.");
            }
        });
        thread.start();
    }
}

class ConcreteObserverTwo implements Observer {
    @Override
    public void update(Message message) {
        // 获取消息通知，执行自己的逻辑...
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("ConcreteObserverTwo is notified.");
            }
        });
        thread.start();
    }
}

image.png

第二种

为了更直观的看到异步的效果，我增加了 ConcreteObserverThree 和 ConcreteObserverFour 两个类。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class Main {
    public static void main(String[] args) {
        // 定长线程池
//        ExecutorService executor = Executors.newFixedThreadPool(4); // 不推荐
        ExecutorService executor = new ThreadPoolExecutor(
                4,
                16,
                10L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());

        ConcreteSubject subject = new ConcreteSubject(executor);

        subject.registerObserver(new ConcreteObserverOne());
        subject.registerObserver(new ConcreteObserverTwo());
        subject.registerObserver(new ConcreteObserverThree());
        subject.registerObserver(new ConcreteObserverFour());

        subject.notifyObservers(new Message());

        executor.shutdown();
    }
}

class Message {
    Integer code;
    String msg;
}

interface Subject {
    void registerObserver(Observer observer);

    void removeObserver(Observer observer);

    void notifyObservers(Message message);
}

interface Observer {
    void update(Message message);
}

class ConcreteSubject implements Subject {
    private Executor executor;
    private List<Observer> observers = new ArrayList<Observer>();

    public ConcreteSubject(Executor executor) {
        this.executor = executor;
    }

    @Override
    public void registerObserver(Observer observer) {
        observers.add(observer);
    }

    @Override
    public void removeObserver(Observer observer) {
        observers.remove(observer);
    }

    @Override
    public void notifyObservers(Message message) {

        for (Observer observer : observers) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    observer.update(message);
                }
            });
        }
    }
}

class ConcreteObserverOne implements Observer {
    @Override
    public void update(Message message) {
        // 获取消息通知，执行自己的逻辑...
        System.out.println("ConcreteObserverOne is notified.");
    }
}

class ConcreteObserverTwo implements Observer {
    @Override
    public void update(Message message) {
        // 获取消息通知，执行自己的逻辑...
        System.out.println("ConcreteObserverTwo is notified.");
    }
}

class ConcreteObserverThree implements Observer {
    @Override
    public void update(Message message) {
        // 获取消息通知，执行自己的逻辑...
        System.out.println("ConcreteObserverThree is notified.");
    }
}

class ConcreteObserverFour implements Observer {
    @Override
    public void update(Message message) {
        // 获取消息通知，执行自己的逻辑...
        System.out.println("ConcreteObserverFour is notified.");
    }
}

image.png

总结

对于第一种实现方式，频繁地创建和销毁线程比较耗时，并且并发线程数无法控制，创建过多的线程会导致堆栈溢出。第二种实现方式，尽管利用了线程池解决了第一种实现方式的问题，但线程池、异步执行逻辑都耦合在了 register() 函数中，增加了这部分业务代码的维护成本。

如果我们的需求更加极端一点，需要在同步阻塞和异步非阻塞之间灵活切换，那就要不停地修改 UserController 的代码。除此之外，如果在项目中，不止一个业务模块需要用到异步非阻塞观察者模式，那这样的代码实现也无法做到复用。

我们知道，框架的作用有：隐藏实现细节，降低开发难度，做到代码复用，解耦业务与非业务代码，让程序员聚焦业务开发。针对异步非阻塞观察者模式，我们也可以将它抽象成框架来达到这样的效果，而这个框架就是我们这节课要讲的 EventBus。

Google Guava EventBus 就是一个比较著名的 EventBus 框架，它不仅仅支持异步非阻塞模式，同时也支持同步阻塞模式。

3.3、Google Guava EventBus（困难）

待续

4、总结

框架的作用有：隐藏实现细节，降低开发难度，做到代码复用，解耦业务与非业务代码，让程序员聚焦业务开发。针对异步非阻塞观察者模式，我们也可以将它抽象成框架来达到这样的效果，而这个框架就是我们这节课讲的 EventBus。EventBus 翻译为“事件总线”，它提供了实现观察者模式的骨架代码。我们可以基于此框架，非常容易地在自己的业务场景中实现观察者模式，不需要从零开始开发。

很多人觉得做业务开发没有技术挑战，实际上，做业务开发也会涉及很多非业务功能的开发，比如今天讲到的 EventBus。在平时的业务开发中，我们要善于抽象这些非业务的、可复用的功能，并积极地把它们实现成通用的框架。

4.1、优缺点

1）优点

观察者和被观察者是抽象耦合的；
建立了一套触发机制；

2）缺点

如果一个被观察者对象有很多的直接和间接的观察者的话，将所有的观察者都通知到会花费很多时间；
如果在观察者和观察目标之间有循环依赖的话，观察目标会触发它们之间进行循环调用，可能导致系统崩溃；
观察者模式没有相应的机制让观察者知道所观察的目标对象是怎么发生变化的，而仅仅只是知道观察目标发生了变化；

4.2、基于不同应用场景的不同实现方式

小到代码层面的解耦，大到架构层面的系统解耦，再或者一些产品的设计思路，都有这种模式的影子，比如，邮件订阅、RSS Feeds，本质上都是观察者模式。

不同的应用场景和需求下，这个模式也有截然不同的实现方式，开篇的时候我们也提到，有同步阻塞的实现方式，也有异步非阻塞的实现方式；有进程内的实现方式，也有跨进程的实现方式。

之前讲到的实现方式，从刚刚的分类方式上来看，它是一种同步阻塞的实现方式。观察者和被观察者代码在同一个线程内执行，被观察者一直阻塞，直到所有的观察者代码都执行完成之后，才执行后续的代码。对照上面讲到的用户注册的例子，register() 函数依次调用执行每个观察者的 handleRegSuccess() 函数，等到都执行完成之后，才会返回结果给客户端。

如果注册接口是一个调用比较频繁的接口，对性能非常敏感，希望接口的响应时间尽可能短，那我们可以将同步阻塞的实现方式改为异步非阻塞的实现方式，以此来减少响应时间。具体来讲，当 userService.register() 函数执行完成之后，我们启动一个新的线程来执行观察者的 handleRegSuccess() 函数，这样 userController.register() 函数就不需要等到所有的 handleRegSuccess() 函数都执行完成之后才返回结果给客户端。userController.register() 函数从执行 3 个 SQL 语句才返回，减少到只需要执行 1 个 SQL 语句就返回，响应时间粗略来讲减少为原来的 1/3。

那如何实现一个异步非阻塞的观察者模式呢？简单一点的做法是，在每个 handleRegSuccess() 函数中，创建一个新的线程执行代码。不过，我们还有更加优雅的实现方式，那就是基于 EventBus 来实现。今天，我们就不展开讲解了。在下一讲中，我会用一节课的时间，借鉴 Google Guava EventBus 框架的设计思想，手把手带你开发一个支持异步非阻塞的 EventBus 框架。它可以复用在任何需要异步非阻塞观察者模式的应用场景中。

刚刚讲到的两个场景，不管是同步阻塞实现方式还是异步非阻塞实现方式，都是进程内的实现方式。如果用户注册成功之后，我们需要发送用户信息给大数据征信系统，而大数据征信系统是一个独立的系统，跟它之间的交互是跨不同进程的，那如何实现一个跨进程的观察者模式呢？

如果大数据征信系统提供了发送用户注册信息的 RPC 接口，我们仍然可以沿用之前的实现思路，在 handleRegSuccess() 函数中调用 RPC 接口来发送数据。但是，我们还有更加优雅、更加常用的一种实现方式，那就是基于消息队列（Message Queue，比如 ActiveMQ）来实现。

当然，这种实现方式也有弊端，那就是需要引入一个新的系统（消息队列），增加了维护成本。不过，它的好处也非常明显。在原来的实现方式中，观察者需要注册到被观察者中，被观察者需要依次遍历观察者来发送消息。而基于消息队列的实现方式，被观察者和观察者解耦更加彻底，两部分的耦合更小。被观察者完全不感知观察者，同理，观察者也完全不感知被观察者。被观察者只管发送消息到消息队列，观察者只管从消息队列中读取消息来执行相应的逻辑。

4.3、关于观察者模式的一些问题

image.png