Skip to content

设计模式:行为型-观察者模式

Published: at 15:57:08

根据应用场景的不同,观察者模式会对应不同的代码实现方式:有同步阻塞的实现方式,也有异步非阻塞的实现方式;有进程内的实现方式,也有跨进程的实现方式。今天我会重点讲解原理、实现、应用场景。

下一节课,我会带你一块实现一个基于观察者模式的异步非阻塞的 EventBus,加深你对这个模式的理解。

原理及应用场景剖析

观察者模式(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。不管怎么称呼,只要应用场景符合刚刚给出的定义,都可以看作观察者模式。

实际上,观察者模式是一个比较抽象的模式,根据不同的应用场景和需求,有完全不同的实现方式,待会我们会详细地讲到。现在,我们先来看其中最经典的一种实现方式。这也是在讲到这种模式的时候,很多书籍或资料给出的最常见的实现方式。具体的代码如下所示:

public interface Subject {
  void registerObserver(Observer observer);
  void removeObserver(Observer observer);
  void notifyObservers(Message message);
}

public interface Observer {
  void update(Message message);
}

public 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);
    }
  }

}

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

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

public class Demo {
  public static void main(String[] args) {
    ConcreteSubject subject = new ConcreteSubject();
    subject.registerObserver(new ConcreteObserverOne());
    subject.registerObserver(new ConcreteObserverTwo());
    subject.notifyObservers(new Message());
  }
}

实际上,上面的代码算是观察者模式的“模板代码”,只能反映大体的设计思路。在真实的软件开发中,并不需要照搬上面的模板代码。观察者模式的实现方法各式各样,函数、类的命名等会根据业务场景的不同有很大的差别,比如 register 函数还可以叫作 attach,remove 函数还可以叫作 detach 等等。

不过,万变不离其宗,设计思路都是差不多的。

原理和代码实现都非常简单,也比较好理解,不需要我过多的解释。

我们还是通过一个具体的例子来重点讲一下,什么情况下需要用到这种设计模式?或者说,这种设计模式能解决什么问题呢?

假设我们在开发一个 P2P 投资理财系统,用户注册成功之后,我们会给用户发放投资体验金。代码实现大致是下面这个样子的:

public class UserController {
  private UserService userService; // 依赖注入
  private PromotionService promotionService; // 依赖注入

  public Long register(String telephone, String password) {
    //省略输入参数的校验代码
    //省略userService.register()异常的try-catch代码
    long userId = userService.register(telephone, password);
    promotionService.issueNewUserExperienceCash(userId);
    return userId;
  }
}

虽然注册接口做了两件事情,注册和发放体验金,违反单一职责原则,但是,如果没有扩展和修改的需求,现在的代码实现是可以接受的。

如果非得用观察者模式,就需要引入更多的类和更加复杂的代码结构,反倒是一种过度设计。

相反,如果需求频繁变动,比如,用户注册成功之后,不再发放体验金,而是改为发放优惠券,并且还要给用户发送一封“欢迎注册成功”的站内信。

这种情况下,我们就需要频繁地修改 register() 函数中的代码,违反开闭原则。而且,如果注册成功之后需要执行的后续操作越来越多,那 register() 函数的逻辑会变得越来越复杂,也就影响到代码的可读性和可维护性。

这个时候,观察者模式就能派上用场了。利用观察者模式,我对上面的代码进行了重构。

重构之后的代码如下所示:

public interface RegObserver {
  void handleRegSuccess(long userId);
}

public class RegPromotionObserver implements RegObserver {
  private PromotionService promotionService; // 依赖注入

  @Override
  public void handleRegSuccess(long userId) {
    promotionService.issueNewUserExperienceCash(userId);
  }
}

public class RegNotificationObserver implements RegObserver {
  private NotificationService notificationService;

  @Override
  public void handleRegSuccess(long userId) {
    notificationService.sendInboxMessage(userId, "Welcome...");
  }
}

public class UserController {
  private UserService userService; // 依赖注入
  private List<RegObserver> regObservers = new ArrayList<>();

  // 一次性设置好,之后也不可能动态的修改
  public void setRegObservers(List<RegObserver> observers) {
    regObservers.addAll(observers);
  }

  public Long register(String telephone, String password) {
    //省略输入参数的校验代码
    //省略userService.register()异常的try-catch代码
    long userId = userService.register(telephone, password);

    for (RegObserver observer : regObservers) {
      observer.handleRegSuccess(userId);
    }

    return userId;
  }
}

当我们需要添加新的观察者的时候,比如,用户注册成功之后,推送用户注册信息给大数据征信系统,基于观察者模式的代码实现,UserController 类的 register() 函数完全不需要修改,只需要再添加一个实现了 RegObserver 接口的类,并且通过 setRegObservers() 函数将它注册到 UserController 类中即可。

不过,你可能会说,当我们把发送体验金替换为发送优惠券的时候,需要修改 RegPromotionObserver 类中 handleRegSuccess() 函数的代码,这还是违反开闭原则呀?你说得没错,不过,相对于 register() 函数来说,handleRegSuccess() 函数的逻辑要简单很多,修改更不容易出错,引入 bug 的风险更低。

前面我们已经学习了很多设计模式,不知道你有没有发现,实际上,设计模式要干的事情就是解耦。

创建型模式是将创建和使用代码解耦,结构型模式是将不同功能代码解耦,行为型模式是将不同的行为代码解耦。

具体到观察者模式,它是将观察者和被观察者代码解耦。

借助设计模式,我们利用更好的代码结构,将一大坨代码拆分成职责更单一的小类,让其满足开闭原则、高内聚松耦合等特性,以此来控制和应对代码的复杂性,提高代码的可扩展性。

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

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

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

之前讲到的实现方式,从刚刚的分类方式上来看,它是一种同步阻塞的实现方式。观察者和被观察者代码在同一个线程内执行,被观察者一直阻塞,直到所有的观察者代码都执行完成之后,才执行后续的代码。

对照上面讲到的用户注册的例子,register() 函数依次调用执行每个观察者的 handleRegSuccess() 函数,等到都执行完成之后,才会返回结果给客户端。

如果注册接口是一个调用比较频繁的接口,对性能非常敏感,希望接口的响应时间尽可能短,那我们可以将同步阻塞的实现方式改为异步非阻塞的实现方式,以此来减少响应时间。

具体来讲,当 userService.register() 函数执行完成之后,我们启动一个新的线程来执行观察者的 handleRegSuccess() 函数,这样 userController.register() 函数就不需要等到所有的 handleRegSuccess() 函数都执行完成之后才返回结果给客户端。userController.register() 函数从执行 3 个 SQL 语句才返回,减少到只需要执行 1 个 SQL 语句就返回,响应时间粗略来讲减少为原来的 1/3。

那如何实现一个异步非阻塞的观察者模式呢?简单一点的做法是,在每个 handleRegSuccess() 函数中,创建一个新的线程执行代码。

不过,我们还有更加优雅的实现方式,那就是基于 EventBus 来实现。

如何实现一个异步非阻塞的EventBus框架?

同步阻塞是最经典的实现方式,主要是为了代码解耦;

异步非阻塞除了能实现代码解耦之外,还能提高代码的执行效率;进程间的观察者模式解耦更加彻底,一般是基于消息队列来实现,用来实现不同进程间的被观察者和观察者之间的交互。

异步非阻塞观察者模式的简易实现

我们有两种实现方式。

其中一种是:在每个 handleRegSuccess() 函数中创建一个新的线程执行代码逻辑;另一种是:在 UserController 的 register() 函数中使用线程池来执行每个观察者的 handleRegSuccess() 函数。两种实现方式的具体代码如下所示:

// 第一种实现方式,其他类代码不变,就没有再重复罗列
public class RegPromotionObserver implements RegObserver {
  private PromotionService promotionService; // 依赖注入

  @Override
  public void handleRegSuccess(Long userId) {
    Thread thread = new Thread(new Runnable() {
      @Override
      public void run() {
        promotionService.issueNewUserExperienceCash(userId);
      }
    });
    thread.start();
  }
}

// 第二种实现方式,其他类代码不变,就没有再重复罗列
public class UserController {
  private UserService userService; // 依赖注入
  private List<RegObserver> regObservers = new ArrayList<>();
  private Executor executor;

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

  public void setRegObservers(List<RegObserver> observers) {
    regObservers.addAll(observers);
  }

  public Long register(String telephone, String password) {
    //省略输入参数的校验代码
    //省略userService.register()异常的try-catch代码
    long userId = userService.register(telephone, password);

    for (RegObserver observer : regObservers) {
      executor.execute(new Runnable() {
        @Override
        public void run() {
          observer.handleRegSuccess(userId);
        }
      });
    }

    return userId;
  }
}

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

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

我们知道,框架的作用有:隐藏实现细节,降低开发难度,做到代码复用,解耦业务与非业务代码,让程序员聚焦业务开发。

针对异步非阻塞观察者模式,我们也可以将它抽象成框架来达到这样的效果,而这个框架就是我们这节课要讲的 EventBus。

EventBus 框架功能需求介绍

EventBus 翻译为“事件总线”,它提供了实现观察者模式的骨架代码。

我们可以基于此框架,非常容易地在自己的业务场景中实现观察者模式,不需要从零开始开发。其中,Google Guava EventBus 就是一个比较著名的 EventBus 框架,它不仅仅支持异步非阻塞模式,同时也支持同步阻塞模式。

现在,我们就通过例子来看一下,Guava EventBus 具有哪些功能。还是上节课那个用户注册的例子,我们用 Guava EventBus 重新实现一下,代码如下所示:

public class UserController {
  private UserService userService; // 依赖注入

  private EventBus eventBus;
  private static final int DEFAULT_EVENTBUS_THREAD_POOL_SIZE = 20;

  public UserController() {
    //eventBus = new EventBus(); // 同步阻塞模式
    eventBus = new AsyncEventBus(Executors.newFixedThreadPool(DEFAULT_EVENTBUS_THREAD_POOL_SIZE)); // 异步非阻塞模式
  }

  public void setRegObservers(List<Object> observers) {
    for (Object observer : observers) {
      eventBus.register(observer);
    }
  }

  public Long register(String telephone, String password) {
    //省略输入参数的校验代码
    //省略userService.register()异常的try-catch代码
    long userId = userService.register(telephone, password);

    eventBus.post(userId);

    return userId;
  }
}

public class RegPromotionObserver {
  private PromotionService promotionService; // 依赖注入

  @Subscribe
  public void handleRegSuccess(Long userId) {
    promotionService.issueNewUserExperienceCash(userId);
  }
}

public class RegNotificationObserver {
  private NotificationService notificationService;

  @Subscribe
  public void handleRegSuccess(Long userId) {
    notificationService.sendInboxMessage(userId, "...");
  }
}

利用 EventBus 框架实现的观察者模式,跟从零开始编写的观察者模式相比,从大的流程上来说,实现思路大致一样,都需要定义 Observer,并且通过 register() 函数注册 Observer,也都需要通过调用某个函数(比如,EventBus 中的 post() 函数)来给 Observer 发送消息(在 EventBus 中消息被称作事件 event)。

但在实现细节方面,它们又有些区别。基于 EventBus,我们不需要定义 Observer 接口,任意类型的对象都可以注册到 EventBus 中,通过 @Subscribe 注解来标明类中哪个函数可以接收被观察者发送的消息。

Guava EventBus 的几个主要的类和函数。

设计模式系列阅读目录