设计模式——观察者模式

一、定义与简单实现

1、定义

观察者模式的定义包括两点：

• 定义对象之间的一对多的依赖（对象之间的关系）
• 当一个对象改变状态时，它的所有观察者都会接受通知并自动更新（对象之间的交流：行为型模式）

观察者模式有点像一个去掉解耦队列之后的发布/订阅模型（Kafka），观察者与被观察者是直接互相依赖的，

2、UML类图

要素：观察者（Observer），被观察者（Subject），通知(notify()方法)

需要弄清楚:怎样建立的依赖关系以及通知的实现。

• 定义一个被观察者Subject，多个观察者Observer，注册形成一对多的依赖
• Subject.notifyObserver()通知观察者

定义了对象之间的一对多依赖，这样一来，当一个对象改变状态时，它的所有依赖者都会受到通知并自动更新。

设计模式原则：

• 封装变化
• 多用组合，少用继承
• 针对接口编程，而不针对具体实现
• 为交互对象中的松耦合设计而努力

3、简单实现

/* 被观察者（主题） */
public interface SubjectInterface {
    void registerObserver(Observer observer);
    void deleteObserver(Observer observer);
    void notifyObservers(Object object);
}

/* 观察者（订阅者） */
public interface Observer {
    void update(SubjectInterface subject, Object object);
    SubjectInterface getSubject();
}


public class Subject implements SubjectInterface {

    /**
     * 这个地方容器类型很重要，
     * 用HashMap甚至可以控制，固定事件给某一个观察者通知
     */
    private ArrayList<Observer> observers = new ArrayList<>();

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

    @Override
    public void deleteObserver(Observer observer) {
        if (observers.contains(observer)){
            observers.remove(observer);
        }
    }

    @Override
    public void notifyObservers(Object object) {
        if(!observers.isEmpty()){
            for (Observer observer : observers){
                observer.update(this,object);
            }
        }
    }
}

public class ObServerAImp implements Observer {

    private SubjectInterface subject;

    public ObServerAImp(SubjectInterface subject){
        this.subject = subject;
        subject.registerObserver(this);
    }
    @Override
    public void update(SubjectInterface subject, Object object) {
        System.out.println("A:I get a message from " + subject + "  :" +object);
    }

    @Override
    public SubjectInterface getSubject(){
        return subject;
    }
}

public class ObServerBImp implements Observer {

    private SubjectInterface subject;

    public ObServerBImp(SubjectInterface subject){
        this.subject = subject;
        subject.registerObserver(this);
    }
    @Override
    public void update(SubjectInterface subject, Object object) {
        System.out.println("B:I get a message from " + subject + "  :" +object);
    }

    @Override
    public SubjectInterface getSubject(){
        return subject;
    }
}

public class Main {

    public static void main(String[] args) {
        //定义一个主题，被观察者
        SubjectInterface subject = new Subject();
        //订阅主题，被观察者：观察者=1：n
        Observer observerA = new ObServerAImp(subject);
        Observer observerB = new ObServerBImp(subject);
        //通知
        subject.notifyObservers("hello Observer!");
        //A取消订阅
        observerA.getSubject().deleteObserver(observerA);
        subject.notifyObservers("hello World!");
    }
}

二、框架中的观察者模式

1、JDK中java.util包下就有一个观察者模式实现接口，Observer（观察者），Observeral（被观察者）

 

 ① 定义了一个changed哨兵，可以控制notifyObservers()是否需要通知观察者（比如可以做个延时）。让观察者模式的实现更有弹性。

/* java.util.Observable#notifyObservers(java.lang.Object) */
    public void notifyObservers(Object arg) {
        Object[] arrLocal;
        synchronized (this) {
            if (!changed)
                //若changed == false，就算状态改变了也不会通知观察者
                return;
            arrLocal = obs.toArray();
            clearChanged();
        }
        for (int i = arrLocal.length-1; i>=0; i--)
            ((Observer)arrLocal[i]).update(this, arg);
    }

② Observable是一个类(Subject)而不是一个接口(SubjectInterface)，所以必须采用继承实现，可能导致问题：

• Java是单继承的
• Observable的setChanged()方法是protected修饰的，意味着只能使用继承，而不能使用组合。违反“多用组合，少用继承”

2、spring中的监听器

简单的UML类图，然后再看看流程逻辑吧

① AbstractApplicationContext中定义了一个被观察者AbstractApplicationEventMulticaster，和多个观察者ApplicationListener

public abstract class AbstractApplicationContext extends DefaultResourceLoader implements ConfigurableApplicationContext {

    /*
     * 观察者容器
     */
    private final Set<ApplicationListener<?>> applicationListeners = new LinkedHashSet<>();
    
    /*
     * 被观察者  里面的defaultRetriever也是观察者容器。
     */

    //在refresh()方法中initApplicationEventMulticaster()时会初始化，
    //初始化之后，所有的观察者都会陆续放入到 applicationEventMulticaster.defaultRetriever.applicationListeners容器中
    private ApplicationEventMulticaster applicationEventMulticaster;

}

/*
 * 观察者
 */
public interface ApplicationListener<E extends ApplicationEvent> extends EventListener {

    void onApplicationEvent(E event);

}

/*
 * 被观察者
 */
public abstract class AbstractApplicationEventMulticaster
        implements ApplicationEventMulticaster, BeanClassLoaderAware, BeanFactoryAware {

        /*
         * 存放观察者的容器
         */
        private final ListenerRetriever defaultRetriever = new ListenerRetriever(false);

}

② 订阅形成依赖：

订阅形成依赖有两种实现

a. AbstractApplicationContext.refresh()方法中的registerListeners()

    /* 
     * 注册各种形式的观察者到被观察者的容器中，形成一对多的依赖
     */
    protected void registerListeners() {
        // Register statically specified listeners first.
        for (ApplicationListener<?> listener : getApplicationListeners()) {
            getApplicationEventMulticaster().addApplicationListener(listener);
        }
        String[] listenerBeanNames = getBeanNamesForType(ApplicationListener.class, true, false);
        for (String listenerBeanName : listenerBeanNames) {
            getApplicationEventMulticaster().addApplicationListenerBean(listenerBeanName);
        }
        Set<ApplicationEvent> earlyEventsToProcess = this.earlyApplicationEvents;
        this.earlyApplicationEvents = null;
        if (earlyEventsToProcess != null) {
            for (ApplicationEvent earlyEvent : earlyEventsToProcess) {
                getApplicationEventMulticaster().multicastEvent(earlyEvent);
            }
        }
    }

b. AbstractApplicationContext.refresh()中的方法registerBeanPostProcessors()执行时注册了一个ApplicationListenerDetector类型的Bean后置处理器，

Bean初始化最后阶段，调用ApplicationListenerDetector的后置方法postProcessAfterInitialization，将观察者注册到被观察者的容器中

/* org.springframework.context.support.ApplicationListenerDetector#postProcessAfterInitialization */
    public Object postProcessAfterInitialization(Object bean, String beanName) {
        if (bean instanceof ApplicationListener) {
            Boolean flag = this.singletonNames.get(beanName);
            if (Boolean.TRUE.equals(flag)) {
                 //单例的ApplicationListener类型（观察者）Bean会被放入到AbstractApplicationContext.applicationListener容器中
                this.applicationContext.addApplicationListener((ApplicationListener<?>) bean);
            }
            else if (Boolean.FALSE.equals(flag)) {
                // ...
                }
                // ...
            }
        }
        return bean;
    }

③ 通知：调用每一个ApplicationListener（两个容器中）的onApplicationEvent方法

AbstractApplicationContext.refresh()最后的finishRefresh()方法中publishEvent(new ContextRefreshedEvent(this));执行所有的监听器

/* org.springframework.context.support.AbstractApplicationContext#publishEvent(java.lang.Object) */
    protected void publishEvent(Object event, @Nullable ResolvableType eventType) {
        //...
        //Bean容器初始化完成，后通知ApplicationListener（观察者）执行
        getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);
        //...
    }

/* org.springframework.context.event.SimpleApplicationEventMulticaster#multicastEvent(org.springframework.context.ApplicationEvent, org.springframework.core.ResolvableType) */
    public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) {
        ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
        Executor executor = getTaskExecutor();
        //从两个容器中获取所有的ApplicationListener（观察者）
        //
        for (ApplicationListener<?> listener : getApplicationListeners(event, type)) {
            if (executor != null) {
                executor.execute(() -> invokeListener(listener, event));
            }
            else {
                //通知观察者，其实就是调用ApplicationListener.onApplicationEvent()，
                invokeListener(listener, event);
            }
        }
    }