10k

设计模式之美-课程笔记36-观察者模式

Published on

创建型设计模式： 解决对象的创建问题；

结构型设计模式：解决类或对象的组合或组装问题；

行为型设计模式：解决类与对象之间的交互问题。

行为型设计模式有：观察者模式、模板模式、策略模式、职责链模式、状态模式、迭代器模式、访问者模式、备忘录模式、命令模式、解释器模式、中介模式。

观察者模式

各种应用场景下观察者模式的不同实现方式

原理及应用场景剖析

  1. 观察者模式：也被称为发布订阅模式：Define a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically. 在对象之间定义一个一对多的依赖，当一个对象状态改变的时候，所有依赖的对象都会自动收到通知。
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());
  }
}
  1. 以上代码只是个模板，反映这个模式的大体思路。要根据实际业务调整。
  2. 为什么使用观察者模式？
  3. 举个例子：假设我们在开发一个 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;
  }
}
  1. 虽然注册接口做了注册和发放体验金违反单一职责，但是如果没有扩展和修改的需求，现在就还好，非要引入观察者模式反而是一种过度设计。
  2. 但是如果需求频繁变动，比如用户注册成功后换成发优惠券而不是体验金，那我们的register需要反复修改，违反OCP。而且如果操作越来越多（比如后面要加一个发放完优惠券后还要发个邮件给用户），那register变得异常复杂，影响其可读性。
  3. 这个时候可以考虑采用观察者模式：
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;
  }
}
  1. 当我们需要添加新的观察者，register不需要修改，只需要再添加一个实现了RegObserver的接口类，并且通过setRegObservers注册到UserController中即可。
  2. 设计模式要干的事情就是解耦。创建新模式是将创建和使用代码解耦，结构型模式是将不同的功能代码解耦，行为型是将不同的行为代码解耦。
  3. 具体到观察者模式，它是将观察者和被观察者代码解耦。以满足开闭原则、高内聚低耦合，控制代码的复杂性，提升代码的可扩展性。

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

  1. 之前的例子是同步阻塞，观察者和被观察者在一个线程内，被观察者一直阻塞直到所有观察者代码都执行完之后才执行后续代码。
  2. 如果注册接口是一个调用很频繁的接口，对性能敏感，希望接口调用时间尽可能短，那我们可以将同步阻塞的实现方式改为异步非阻塞的实现。
    1. 具体来讲就是当regster执行完成后我们启动一个新线程来执行观察者的handleRegSuccess函数。这样register不需要等他们都执行完再返回结果给客户端。
  3. 如何实现异步非阻塞的观察者模式呢？
    1. 简单的就是在每个handleRegSuccess中创建一个新线程。
    2. 更好的实现看下节基于Google Guava EvenBus的实现。
  4. 如何实现跨进程的观察者模式呢？
    1. 如果用户注册成功之后，我们需要发送用户信息给大数据征信系统，而大数据征信系统是一个独立的系统。
    2. 如果大数据征信系统提供了发送用户注册信息的RPC接口，我们可以沿用之前的思路。
    3. 但是我们还可以基于消息队列做这件事。但是引入新的系统带来的就是新的复杂。好处是消息队列的解耦更彻底。观察者和被观察者互相不感知。被观察者只管发送消息到消息队列，观察者只管从消息队列中读取消息来执行相应的逻辑。

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

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

  1. 之前说过两种简单的实现方式，一种是通过创建新线程；一种是在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;
  }
}
  1. 第一种方式频繁创建销毁线程比较耗时，并发线程数无法控制，过多线程可能会导致堆栈溢出。第二种实现方式，虽然解决了第一种方式问题，但是线程池、异步执行的逻辑耦合在了register函数中，增加了这部分业务代码的维护成本。
  2. 再极端一点，需要再同步和异步之间切换，那就要不停修改Controller代码。
  3. 而且如果不止一个地方用到异步非阻塞观察者模式，这样的代码也无法复用。

EventBus框架功能需求介绍

  1. 他提供了实现观察者模式的骨架代码。不仅支持异步非阻塞也支持同步阻塞模式。
  2. 利用框架实现上节的用户注册案例：
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, "...");
  }
}

  1. 总体思路一样，定义Observer，通过register注册observer，也都需要通过调用某个函数来给Observer发消息。

  2. 基于 EventBus，我们不需要定义 Observer 接口，任意类型的对象都可以注册到 EventBus 中，通过 @Subscribe 注解来标明类中哪个函数可以接收被观察者发送的消息。

  3. Guava EventBus的几个主要类和函数

    • EventBus, AsyncEventBus: Guava EventBus对外暴露的所有可调用接口，都封装在EventBus类中。其中EventBus实现了同步阻塞的观察者模式，AsyncEventBus继承自EventBus提供了异步非阻塞的观察者模式。EventBus eventBus = new EventBus(); // 同步阻塞模式 EventBus eventBus = new AsyncEventBus(Executors.newFixedThreadPool(8))；// 异步阻塞模式

    • register(): 注册观察者。他可以接受任何类型的观察者，而在经典观察者模式中，register必须接受实现了同一个Observer接口的类对象。public void register(Object object);

    • unregister(): 删除某个观察者,public void unregister(Object object);

    • post(): 给观察者发送消息,public void post(Object event);。 跟经典观察者模式不同的是，post并非发消息给所有观察者而是发给可匹配的观察者。所谓可匹配指的是，能接收的消息类型是发送消息（post 函数定义中的 event）类型的父类

      • 比如，AObserver 能接收的消息类型是 XMsg，BObserver 能接收的消息类型是 YMsg，CObserver 能接收的消息类型是 ZMsg。其中，XMsg 是 YMsg 的父类。当我们如下发送消息的时候，相应能接收到消息的可匹配观察者如下所示

        XMsg xMsg = new XMsg(); YMsg yMsg = new YMsg(); ZMsg zMsg = new ZMsg(); post(xMsg); => AObserver接收到消息 post(yMsg); => AObserver、BObserver接收到消息 post(zMsg); => CObserver接收到消息

    • @Subscribe: 标明某个函数能接受哪种类型的消息。

      • 当通过 register() 函数将 DObserver 类对象注册到 EventBus 的时候，EventBus 会根据 @Subscribe 注解找到 f1() 和 f2()，并且将两个函数能接收的消息类型记录下来（PMsg->f1，QMsg->f2）。当我们通过 post() 函数发送消息（比如 QMsg 消息）的时候，EventBus 会通过之前的记录（QMsg->f2），调用相应的函数（f2）。
public DObserver {
  //...省略其他属性和方法...
  
  @Subscribe
  public void f1(PMsg event) { //... }
  
  @Subscribe
  public void f2(QMsg event) { //... }
}

手把手实现一个EvenBus框架

  1. 两个关键函数register 和post的原理

    img

    img

  2. 关键的数据结构是一个Observer注册表。记录了消息类型和可接受消息函数的对应关系。当调用register函数注册观察者的时候，EventBus通过解析@Subscribe注解，生成Observer表，当调用post函数发送消息的时候，框架通过注册表找到可接受消息的函数通过Java反射动态的床架对象、执行函数。

  3. 对于同步阻塞模式，框架在一个线程内依次执行相应的函数；异步非阻塞模式，EventBus通过线程池执行相应函数。
  4. 小框架的代码实现包括 5 个类：EventBus、AsyncEventBus、Subscribe、ObserverAction、ObserverRegistry。
1. Subscribe

他是一个注解，用于标明观察者中哪个函数可以接收消息

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@Beta
public @interface Subscribe {}
2. ObserverAction

ObserverAction 类用来表示 @Subscribe 注解的方法，其中，target 表示观察者类，method 表示方法。它主要用在 ObserverRegistry 观察者注册表中。

public class ObserverAction {
  private Object target;
  private Method method;

  public ObserverAction(Object target, Method method) {
    this.target = Preconditions.checkNotNull(target);
    this.method = method;
    this.method.setAccessible(true);
  }

  public void execute(Object event) { // event是method方法的参数
    try {
      method.invoke(target, event);
    } catch (InvocationTargetException | IllegalAccessException e) {
      e.printStackTrace();
    }
  }
}
3. ObserverRegistry
public class ObserverRegistry {
  private ConcurrentMap<Class<?>, CopyOnWriteArraySet<ObserverAction>> registry = new ConcurrentHashMap<>();

  public void register(Object observer) {
    Map<Class<?>, Collection<ObserverAction>> observerActions = findAllObserverActions(observer);
    for (Map.Entry<Class<?>, Collection<ObserverAction>> entry : observerActions.entrySet()) {
      Class<?> eventType = entry.getKey();
      Collection<ObserverAction> eventActions = entry.getValue();
      CopyOnWriteArraySet<ObserverAction> registeredEventActions = registry.get(eventType);
      if (registeredEventActions == null) {
        registry.putIfAbsent(eventType, new CopyOnWriteArraySet<>());
        registeredEventActions = registry.get(eventType);
      }
      registeredEventActions.addAll(eventActions);
    }
  }

  public List<ObserverAction> getMatchedObserverActions(Object event) {
    List<ObserverAction> matchedObservers = new ArrayList<>();
    Class<?> postedEventType = event.getClass();
    for (Map.Entry<Class<?>, CopyOnWriteArraySet<ObserverAction>> entry : registry.entrySet()) {
      Class<?> eventType = entry.getKey();
      Collection<ObserverAction> eventActions = entry.getValue();
      if (postedEventType.isAssignableFrom(eventType)) {
        matchedObservers.addAll(eventActions);
      }
    }
    return matchedObservers;
  }

  private Map<Class<?>, Collection<ObserverAction>> findAllObserverActions(Object observer) {
    Map<Class<?>, Collection<ObserverAction>> observerActions = new HashMap<>();
    Class<?> clazz = observer.getClass();
    for (Method method : getAnnotatedMethods(clazz)) {
      Class<?>[] parameterTypes = method.getParameterTypes();
      Class<?> eventType = parameterTypes[0];
      if (!observerActions.containsKey(eventType)) {
        observerActions.put(eventType, new ArrayList<>());
      }
      observerActions.get(eventType).add(new ObserverAction(observer, method));
    }
    return observerActions;
  }

  private List<Method> getAnnotatedMethods(Class<?> clazz) {
    List<Method> annotatedMethods = new ArrayList<>();
    for (Method method : clazz.getDeclaredMethods()) {
      if (method.isAnnotationPresent(Subscribe.class)) {
        Class<?>[] parameterTypes = method.getParameterTypes();
        Preconditions.checkArgument(parameterTypes.length == 1,
                "Method %s has @Subscribe annotation but has %s parameters."
                        + "Subscriber methods must have exactly 1 parameter.",
                method, parameterTypes.length);
        annotatedMethods.add(method);
      }
    }
    return annotatedMethods;
  }
}

CopyOnWriteArraySet，顾名思义，在写入数据的时候，会创建一个新的 set，并且将原始数据 clone 到新的 set 中，在新的 set 中写入数据完成之后，再用新的 set 替换老的 set。这样就能保证在写入数据的时候，不影响数据的读取操作，以此来解决读写并发问题。除此之外，CopyOnWriteSet 还通过加锁的方式，避免了并发写冲突。

4. EventBus

EventBus 实现的是阻塞同步的观察者模式。看代码你可能会有些疑问，这明明就用到了线程池 Executor 啊。实际上，MoreExecutors.directExecutor() 是 Google Guava 提供的工具类，看似是多线程，实际上是单线程。之所以要这么实现，主要还是为了跟 AsyncEventBus 统一代码逻辑，做到代码复用。

public class EventBus {
  private Executor executor;
  private ObserverRegistry registry = new ObserverRegistry();

  public EventBus() {
    this(MoreExecutors.directExecutor());
  }

  protected EventBus(Executor executor) {
    this.executor = executor;
  }

  public void register(Object object) {
    registry.register(object);
  }

  public void post(Object event) {
    List<ObserverAction> observerActions = registry.getMatchedObserverActions(event);
    for (ObserverAction observerAction : observerActions) {
      executor.execute(new Runnable() {
        @Override
        public void run() {
          observerAction.execute(event);
        }
      });
    }
  }
}
5. AsyncEventBus
public class AsyncEventBus extends EventBus {
  public AsyncEventBus(Executor executor) {
    super(executor);
  }
}

Guava EventBus Source Code

Thoughts? Leave a comment