简单的观察者模式实现

import java.util.ArrayList;

import java.util.Collections;

import java.util.Iterator;

import java.util.List;

/**

* 观察者模式中用到了回调:

* A. 观察者将自己注册到被观察者的监听者列表，且观察者类自身提供了一个回调函数

* B. 被观察者(Observable或Subject)维护观察者列表，并且可以注册和解注册观察者

* C. 一旦被观察者状态发生改变，它可以调用notifyObservers()，这个方法将遍历观察者列表并逐个调用

观察者提供的回调函数

* @author will

*

*/

public class SimpleObserverPattern {

public static void main(String[] args) {

SimpleObserverPattern sop = new SimpleObserverPattern();

List observers = new ArrayList ();

IObserver observerA = sop.new Observer("ObserverA");

IObserver observerB = sop.new Observer("ObserverB");

observers.add(observerA);

observers.add(observerB);

IObservable observable = sop.new Observable(observers);

observable.registerObserver(sop.new Observer("ObserverC"));

observable.changeState();

observable.close();

}

// 被观察者，有的地方叫Subject

interface IObservable {

void registerObserver(IObserver observer);

void unregisterObserver(IObserver observer);

void notifyObservers();

String getState();

void changeState();

void close();

}

class Observable implements IObservable {

private static final String NEW = "New";

private static final String CHANGED = "Changed";

private static final String CLOSED = "Closed";

private String state;

private List observers;

public Observable() {

this(null);

}

public Observable(List observers) {

if(observers == null) {

observers = new ArrayList ();

}

this.observers = Collections.synchronizedList(observers);

this.state = NEW;

}

@Override

public void registerObserver(IObserver observer) {

observers.add(observer);

}

@Override

public void unregisterObserver(IObserver observer) {

observers.remove(observer);

}

@Override

public void notifyObservers() {

Iterator iter = observers.iterator();

while(iter.hasNext()) {

iter.next().update(this);

}

}

@Override

public String getState() {

return state;

}

@Override

public void changeState() {

this.state = CHANGED;

notifyObservers();

}

@Override

public void close() {

this.state = CLOSED;

notifyObservers();

}

}

interface IObserver {

void update(IObservable observalbe);

}

class Observer implements IObserver {

private String name;

public Observer(String name) {

this.name = name;

}

@Override

public void update(IObservable observalbe) {

System.out.println(

String.format("%s receive observalbe's change, current observalbe's state is %s",

name, observalbe.getState()));

}

}

}

上面的实现直接将被观察者对象作为回调函数参数，这样做很不优雅，在简单的场景可能奏效。

但事实上更多情况下，一个被观察者有很多种事件或者状态，而每个观察者可能感兴趣的事件或状态都不相同，或者为了信息隐藏的目的，不想让每个观察者都能访问到Observable内部的所有状态。

这样我继续演化代码为下面这个版本，注意我这里没有很细致地考虑并发问题。

import java.util.Collections;

import java.util.HashSet;

import java.util.Hashtable;

import java.util.Iterator;

import java.util.Set;

public class MultiEventObserverPattern {

public static void main(String[] args) {

MultiEventObserverPattern meop = new MultiEventObserverPattern();

IObservable observable = meop.new Observable();

IObserver observerA = meop.new Observer("ObserverA");

IObserver observerB = meop.new Observer("ObserverB");

// 注册感兴趣的事件

observable.registerObserver(observable.getEventA(), observerA);

observable.registerObserver(observable.getEventB(), observerB);

// 改变被观察者状态

observable.changeStateA();

observable.changeStateB();

}

interface IEvent {

void eventChange();

String getState();

}

class EventA implements IEvent {

private static final String INITIALIZED = "Initialized";

private static final String PENDING = "Pending";

private String state;

public EventA() {

this.state = INITIALIZED;

}

@Override

public void eventChange() {

System.out.println("EventA change");

this.state = PENDING;

}

@Override

public String toString() {

return "EventA";

}

@Override

public String getState() {

return state;

}

}

class EventB implements IEvent {

private static final String NEW = "New";

private static final String IDLE = "Idle";

private String state;

public EventB() {

this.state = NEW;

}

@Override

public void eventChange() {

System.out.println("EventB change");

this.state = IDLE;

}

@Override

public String toString() {

return "EventB";

}

@Override

public String getState() {

return state;

}

}

// 被观察者(Observable)，有的地方叫Subject

interface IObservable {

void registerObserver(IEvent event, IObserver observer);

void unregisterObserver(IEvent event, IObserver observer);

// 通知观察者某个事件发生了

void notifyObservers(IEvent event);

void changeStateA();

void changeStateB();

IEvent getEventA();

IEvent getEventB();

}

class Observable implements IObservable {

private IEvent eventA;

private IEvent eventB;

private Hashtable> eventObserverMapping;

public Observable() {

this(null);

}

// 这里如果evenObserverMapping传入的某些Set是未被同步修饰的，那么也没办法

public Observable(Hashtable> eventObserverMapping) {

if(eventObserverMapping == null) {

eventObserverMapping = new Hashtable> ();

}

this.eventObserverMapping = new Hashtable> ();

this.eventA = new EventA();

this.eventB = new EventB();

}

@Override

public void registerObserver(IEvent event, IObserver observer) {

Set observers = eventObserverMapping.get(event);

if(observers == null) {

observers = Collections.synchronizedSet(new HashSet ());

observers.add(observer);

eventObserverMapping.put(event, observers);

}

else {

observers.add(observer);

}

}

@Override

public void unregisterObserver(IEvent event, IObserver observer) {

Set observers = eventObserverMapping.get(event);

if(observers != null) {

observers.remove(observer);

}

}

@Override

public void notifyObservers(IEvent event) {

Set observers = eventObserverMapping.get(event);

if(observers != null && observers.size() > 0) {

Iterator iter = observers.iterator();

while(iter.hasNext()) {

iter.next().update(event);

}

}

}

@Override

public void changeStateA() {

// 改变状态A会触发事件A

eventA.eventChange();

notifyObservers(eventA);

}

@Override

public void changeStateB() {

// 改变状态B会触发事件B

eventB.eventChange();

notifyObservers(eventB);

}

@Override

public IEvent getEventA() {

return eventA;

}

@Override

public IEvent getEventB() {

return eventB;

}

}

interface IObserver {

void update(IEvent event);

}

class Observer implements IObserver {

private String name;

public Observer(String name) {

this.name = name;

}

@Override

public void update(IEvent event) {

System.out.println(

String.format("%s receive %s's change, current observalbe's state is %s",

name, event, event.getState()));

}

}

}

似乎看起来挺完美了，但还是不够完美。因为事件被硬编码为被观察者类的属性。这样事件类型在编译时期就被定死了，如果要增加新的事件类型就不得不修改IObservable接口和Observable类，这大大削减了灵活性。

相当于被观察者耦合于这些具体的事件，那么我们如何来打破这个限制呢？

答案是引入一个新的组件，让那个组件来管理事件、观察者、被观察者之间的关系，事件发生时也由那个组件来调用观察者的回调函数。这也是一种解耦吧，有点类似Spring的IOC容器。

至于具体实现，我觉得Guava EventBus做得已经蛮好了，可以参考我前面提到的链接。

PS：本帖不是为Guava EventBus做广告，只是自己的思路一步步推进，逐渐地就和Guava EventBus的设计思路吻合了。

下面继续看看JDK标准类实现观察者模式的例子，然后分析下它的源码实现，要看的只有一个Observable类和一个Observer接口。

JDK标准类实现观察者模式

import java.util.Observable;

import java.util.Observer;

/**

* 使用java.util包中的标准类实现观察者模式

* @author will

*

*/

public class JDKObserverDemo {

public static void main(String[] args) {

JDKObserverDemo jod = new JDKObserverDemo();

// 被观察者

MyObservable myObservable = jod.new MyObservable("hello");

// 观察者

Observer myObserver = jod.new MyObserver();

// 注册

myObservable.addObserver(myObserver);

// 改变被观察者状态，触发观察者回调函数

myObservable.setValue("will");

}

class MyObservable extends Observable {

private String watchedValue;   // 被观察的值

public MyObservable(String watchedValue) {

this.watchedValue = watchedValue;

}

public void setValue(String newValue) {

if(!watchedValue.equals(newValue)) {

watchedValue = newValue;

setChanged();

notifyObservers(newValue);

}

}

@Override

public String toString() {

return "MyObservable";

}

}

class MyObserver implements Observer {

@Override

public void update(Observable o, Object arg) {

System.out.println(o + "'s state changed, argument is: " + arg);

}

}

}

看了下JDK标准库中的Observer和Observable实现很简单，不想多说了。

下面是Quartz中的监听器实现。

QuartzScheduler被监听者

import java.util.ArrayList;

import java.util.Date;

import java.util.List;

/**

* Quartz核心类，相当于Observable(被观察者)

* @author will

*

*/

public class QuartzScheduler {

private ArrayList internalSchedulerListeners = new ArrayList(10);

// private ArrayList interanlJobListeners = new ArrayList();   // 一个Observable可以包含多组监听器

public Date scheduleJob(Trigger trigger) {

if(trigger == null) {

return null;

}

System.out.println("Schedule job, trigger: " + trigger);

notifySchedulerListenersScheduled(trigger);

return new Date();

}

public void unScheduleJob(Trigger trigger) {

if(trigger == null) {

return;

}

System.out.println("Unschedule job, trigger: " + trigger);

notifyShedulerListenerUnScheduled(trigger);

}

// 注册SchedulerListener

public void addInternalSchedulerListener(SchedulerListener schedulerListener) {

synchronized (internalSchedulerListeners) {

internalSchedulerListeners.add(schedulerListener);

}

}

// 移除SchedulerListener

public boolean removeInternalSchedulerListener(SchedulerListener schedulerListener) {

synchronized (internalSchedulerListeners) {

return internalSchedulerListeners.remove(schedulerListener);

}

}

public List getInternalSchedulerListeners() {

synchronized (internalSchedulerListeners) {

return java.util.Collections.unmodifiableList(new ArrayList(internalSchedulerListeners));

}

}

public void notifySchedulerListenersScheduled(Trigger trigger) {

for(SchedulerListener listener: getInternalSchedulerListeners()) {

listener.jobScheduled(trigger);

}

}

public void notifyShedulerListenerUnScheduled(Trigger trigger) {

for(SchedulerListener listener: getInternalSchedulerListeners()) {

listener.jobUnScheduled(trigger);

}

}

}

SchedulerListener

// 监听接口，回调函数，Client注册监听时需要提供回调函数实现

public interface SchedulerListener {

void jobScheduled(Trigger trigger);

void jobUnScheduled(Trigger trigger);

}

Trigger

// Trigger

public class Trigger {

private String triggerKey;

private String triggerName;

public Trigger(String triggerKey, String triggerName) {

this.triggerKey = triggerKey;

this.triggerName = triggerName;

}

public String getTriggerKey() {

return triggerKey;

}

public void setTriggerKey(String triggerKey) {

this.triggerKey = triggerKey;

}

public String getTriggerName() {

return triggerName;

}

public void setTriggerName(String triggerName) {

this.triggerName = triggerName;

}

public String toString() {

return String.format("{triggerKey: %s, triggerName: %s}", triggerKey, triggerName);

}

}

Test

public class Test {

public static void main(String[] args) {

QuartzScheduler qs = new QuartzScheduler();

SchedulerListener listenerA = new SchedulerListener() {

@Override

public void jobUnScheduled(Trigger trigger) {

System.out.println("listenerA job unscheduled: " + trigger.getTriggerName());

}

@Override

public void jobScheduled(Trigger trigger) {

System.out.println("listenerA job scheduled: " + trigger.getTriggerName());

}

};

SchedulerListener listenerB = new SchedulerListener() {

@Override

public void jobUnScheduled(Trigger trigger) {

System.out.println("listenerB job unscheduled: " + trigger.getTriggerName());

}

@Override

public void jobScheduled(Trigger trigger) {

System.out.println("listenerB job scheduled: " + trigger.getTriggerName());

}

};

// 注册Scheduler Listener

qs.addInternalSchedulerListener(listenerA);

qs.addInternalSchedulerListener(listenerB);

Trigger triggerA = new Trigger("Key1", "triggerA");

Trigger triggerB = new Trigger("Key2", "triggerB");

qs.scheduleJob(triggerA);

qs.scheduleJob(triggerB);

qs.unScheduleJob(triggerA);

}

}