极客时间《设计模式之美》笔记---观察者模式

怼烎@ 2022-09-07 06:12 174阅读 0赞

### 文章目录 ###

*   *  观察者模式（Observer Design Pattern）
     *  如何实现一个异步非阻塞的EventBus框架？
     *   *  EventBus框架功能需求介绍
         *  手把手实现一个EventBus框架
         *   *  1.Subscribe
             *  2.ObserverAction
             *  3.ObserverRegistry
             *  4.EventBus
             *  5.AsyncEventBus

## 观察者模式（Observer Design Pattern） ##

> 在对象之间定义一个一对多的依赖，当一个对象状态改变的时候，所有依赖的对象都会自动收到通知。

一般情况下，被依赖的对象叫作**被观察者（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 observers = new ArrayList();
    
      @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());
      }
    }

实际上，**设计模式要干的事情就是解耦。创建型模式是将创建和使用代码解耦，结构型模式是将不同功能代码解耦，行为型模式是将不同的行为代码解耦，具体到观察者模式，它是将观察者和被观察者代码解耦。借助设计模式，我们利用更好的代码结构，将一大坨代码拆分成职责更单一的小类，让其满足开闭原则、高内聚松耦合等特性，以此来控制和应对代码的复杂性，提高代码的可扩展性**。

> 场景：一个P2P投资理财系统，用户注册成功之后，我们会给用户发放投资体验金。

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 regObservers = new ArrayList<>();
    
      // 一次性设置好，之后也不可能动态的修改
      public void setRegObservers(List 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;
      }
    }

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

异步非阻塞实现：

// 第一种实现方式，其他类代码不变，就没有再重复罗列
    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 regObservers = new ArrayList<>();
      private Executor executor;
    
      public UserController(Executor executor) { 
        this.executor = executor;
      }
    
      public void setRegObservers(List 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;
      }
    }

### 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 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框架 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70 1]

#### 1.Subscribe ####

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 ####

ObserverRegistry类就是前面讲到的Observer注册表，是最复杂的一个类，框架中几乎所有的核心逻辑都在这个类中。

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

#### 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 observerActions = registry.getMatchedObserverActions(event);
        for (ObserverAction observerAction : observerActions) { 
          executor.execute(new Runnable() { 
            @Override
            public void run() { 
              observerAction.execute(event);
            }
          });
        }
      }
    }

#### 5.AsyncEventBus ####

有了EventBus，AsyncEventBus的实现就非常简单了。为了实现异步非阻塞的观察者模式，它就不能再继续使用MoreExecutors.directExecutor()了，而是需要在构造函数中，由调用者注入线程池。

public class AsyncEventBus extends EventBus { 
      public AsyncEventBus(Executor executor) { 
        super(executor);
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70]: /images/20220829/1686d0f928e54a26a2eea5a6eed987eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXNoZW5neGlfcm9vdA_size_16_color_FFFFFF_t_70 1]: /images/20220829/84ec833343d74d9e9d293c01d2c11c55.png