观察者模式是设计模式中的 “超级模式”,其应用随处可见,我们以微信公众号为例。
微信公众号有服务号、订阅号和企业号之分。当我们在公众号上发布一篇博文推送时,订阅的用户都能够在我发布推送之后及时接收到推送,即可方便地在手机端进行阅读。
观察者模式(Observer Pattern):定义对象之间的一种一对多依赖关系,使得每当一个对象状态发生改变时,其相关依赖对象皆得到通知并被自动更新。观察者模式是一种对象行为型模式。
观察者模式的别名包括发布-订阅(Publish/Subscribe)模式、模型-视图(Model/View)模式、源-监听器(Source/Listener)模式或从属者(Dependents)模式。
观察者模式包含观察目标和观察者两类对象,**一个目标可以有任意数目的与之相依赖的观察者,**一旦观察目标的状态发生改变,所有的观察者都将得到通知。
一般是多对一依赖,即一个被观察者,和多个观察者
一旦大忽悠更新了公众号,所有订阅其公众号的粉丝都会接收到更新推送
Subject(目标):目标又称为主题,它是指被观察的对象。在目标中定义了一个观察者集合,一个观察目标可以接受任意数量的观察者来观察,它提供一系列方法来增加和删除观察者对象,同时它定义了通知方法notify()。目标类可以是接口,也可以是抽象类或具体类。
ConcreteSubject(具体目标):具体目标是目标类的子类,通常它包含有经常发生改变的数据,当它的状态发生改变时,向它的各个观察者发出通知;同时它还实现了在目标类中定义的抽象业务逻辑方法(如果有的话)。如果无须扩展目标类,则具体目标类可以省略。
Observer(观察者):观察者将对观察目标的改变做出反应,观察者一般定义为接口,该接口声明了更新数据的方法update(),因此又称为抽象观察者。
ConcreteObserver(具体观察者):在具体观察者中维护一个指向具体目标对象的引用,它存储具体观察者的有关状态,这些状态需要和具体目标的状态保持一致;它实现了在抽象观察者Observer中定义的update()方法。通常在实现时,可以调用具体目标类的attach()方法将自己添加到目标类的集合中或通过detach()方法将自己从目标类的集合中删除。
首先需要一个订阅者接口(观察者),该接口有一个 receive
方法,用于接收公众号推送通知
//订阅者---观察者
public interface Subscriber
{
//接收发布者发布消息的方法
public void receive();
}
然后是一个微信客户端(具体观察者),实现了 receive 方法
//处理微信订阅的业务逻辑
public class WeChatSub implements Subscriber
{
//当前订阅者的名字
private String subName;
WeChatSub(String subName)
{
this.subName=subName;
}
@Override
public void receive(String publisher, String passageName) {
//接收到推送消息时的具体业务逻辑操作
System.out.println(String.format("用户[%s] , 接收到[%s]的订阅号推送," +
"推送文章为:%s ",subName,publisher,passageName));
}
}
发布者类(目标,被观察对象),该类维护了一个订阅者列表,实现了订阅、取消订阅、通知所有订阅者等功能
//发布者--被观察的对象Subject
public class Publisher
{
//存放所有需要通知的观察者
static private List<Subscriber> subscribers=new ArrayList<>();
//记录是否发布的状态,默认false
static private Boolean pubStatus=false;
protected void subscribe(Subscriber subscriber) {
this.subscribers.add(subscriber);
}
protected void unsubscribe(Subscriber subscriber) {
if (this.subscribers.contains(subscriber)) {
this.subscribers.remove(subscriber);
}
}
protected void notifySubscribers(String publisher, String articleName) {
if (this.pubStatus == false) {
return;
}
for (Subscriber subscriber : this.subscribers) {
subscriber.receive(publisher, articleName);
}
this.clearPubStatus();
}
protected void setPubStatus() {
this.pubStatus = true;
}
protected void clearPubStatus() {
this.pubStatus = false;
}
}
微信公众号类(具体目标),该类提供了 publishArticles
方法,用于发布推送,当文章发布完毕时调用父类的通知所有订阅者方法
public class WeChatPublisher extends Publisher
{
private String name;
public WeChatPublisher(String name) {
this.name = name;
}
public void publishArticles(String articleName, String content) {
System.out.println(String.format("\n<%s>微信公众号 发布了一篇推送,文章名称为 <%s>,内容为 <%s> ", this.name, articleName, content));
setPubStatus();
notifySubscribers(this.name, articleName);
}
}
客户端测试
//客户端
public class Client
{
public static void main(String[] args) {
//创建一个发布者
WeChatPublisher dhy=new WeChatPublisher("大忽悠");
//创建订阅者
Subscriber sub1=new WeChatSub("小朋友");
Subscriber sub2=new WeChatSub("小忽悠");
Subscriber sub3=new WeChatSub("大朋友");
//订阅大忽悠公众号
dhy.subscribe(sub1);
dhy.subscribe(sub2);
dhy.subscribe(sub3);
//大忽悠发布推送
dhy.publishArticles("设计模式","观察者模式");
}
}
观察者模式在Java语言中的地位非常重要。在JDK的 java.util
包中,提供了 Observable
类以及 Observer
接口,它们构成了JDK对观察者模式的支持。
其中的 Observer
接口为观察者,只有一个 update
方法,当观察目标发生变化时被调用,其代码如下:
public interface Observer {
void update(Observable o, Object arg);
}
Observable
类则为目标类,相比我们的示例中的 Publisher
类多了并发和NPE方面的考虑
public class Observable {
private boolean changed = false;
private Vector<Observer> obs = new Vector();
public Observable() {
}
// 用于注册新的观察者对象到向量中
public synchronized void addObserver(Observer var1) {
if (var1 == null) {
throw new NullPointerException();
} else {
if (!this.obs.contains(var1)) {
this.obs.addElement(var1);
}
}
}
// 用于删除向量中的某一个观察者对象
public synchronized void deleteObserver(Observer var1) {
this.obs.removeElement(var1);
}
public void notifyObservers() {
this.notifyObservers((Object)null);
}
// 通知方法,用于在方法内部循环调用向量中每一个观察者的update()方法
public void notifyObservers(Object var1) {
Object[] var2;
synchronized(this) {
if (!this.changed) {
return;
}
var2 = this.obs.toArray();
this.clearChanged();
}
for(int var3 = var2.length - 1; var3 >= 0; --var3) {
((Observer)var2[var3]).update(this, var1);
}
}
// 用于清空向量,即删除向量中所有观察者对象
public synchronized void deleteObservers() {
this.obs.removeAllElements();
}
// 该方法被调用后会设置一个boolean类型的内部标记变量changed的值为true,表示观察目标对象的状态发生了变化
protected synchronized void setChanged() {
this.changed = true;
}
// 用于将changed变量的值设为false,表示对象状态不再发生改变或者已经通知了所有的观察者对象,调用了它们的update()方法
protected synchronized void clearChanged() {
this.changed = false;
}
// 返回对象状态是否改变
public synchronized boolean hasChanged() {
return this.changed;
}
// 返回向量中观察者的数量
public synchronized int countObservers() {
return this.obs.size();
}
}
我们可以使用 Observable
类以及 Observer
接口来重新实现微信公众号示例。
增加一个通知类 WechatNotice
,用于推送通知的传递
@Data
@AllArgsConstructor
public class WechatNotice {
private String publisher;
private String articleName;
}
然后改写 WeChatClient
和 WeChatAccounts
,分别实现JDK的 Observer
接口和继承 Observable
类
public class WeChatClient implements Observer {
private String username;
public WeChatClient(String username) {
this.username = username;
}
@Override
public void update(Observable o, Object arg) {
//WeChatAccounts weChatAccounts = (WeChatAccounts) o;
WechatNotice notice = (WechatNotice) arg;
System.out.println(String.format("用户<%s> 接收到 <%s>微信公众号 的推送,文章标题为 <%s>", username, notice.getPublisher(), notice.getArticleName()));
}
}
public class WeChatAccounts extends Observable {
private String name;
public WeChatAccounts(String name) {
this.name = name;
}
public void publishArticles(String articleName, String content) {
System.out.println(String.format("\n<%s>微信公众号 发布了一篇推送,文章名称为 <%s>,内容为 <%s> ", this.name, articleName, content));
setChanged();
notifyObservers(new WechatNotice(this.name, articleName));
}
}
测试,与示例中的测试代码的区别在于调用的方法不同
public class Test {
public static void main(String[] args) {
WeChatAccounts accounts = new WeChatAccounts("大忽悠");
WeChatClient user1 = new WeChatClient("张三");
WeChatClient user2 = new WeChatClient("李四");
WeChatClient user3 = new WeChatClient("王五");
accounts.addObserver(user1);
accounts.addObserver(user2);
accounts.addObserver(user3);
accounts.publishArticles("设计模式 | 观察者模式及典型应用", "观察者模式的内容...");
accounts.deleteObserver(user1);
accounts.publishArticles("设计模式 | 单例模式及典型应用", "单例模式的内容....");
}
}
Guava 中的 EventBus
封装了友好的 “生产/消费模型”,通过非常简单的方式,实现了观察者模式中的监听注册,事件分发。
使用了 Guava
EventBus
之后,如果需要订阅消息,不需要实现任何接口,只需在监听方法上加上 @Subscribe
注解即可,EventBus
提供了 register
和 unregister
方法用于注册与取消注册事件,当 EventBus
调用 post
方法时将把事件分发给注册的对象
使用 Guava 重新实现示例
@Data
@AllArgsConstructor
public class WechatNotice {
private String publisher;
private String articleName;
}
public class WeChatClient {
private String username;
public WeChatClient(String username) {
this.username = username;
}
@Subscribe
public void listen(WechatNotice notice) {
System.out.println(String.format("用户<%s> 接收到 <%s>微信公众号 的推送,文章标题为 <%s>", username, notice.getPublisher(), notice.getArticleName()));
}
}
public class WeChatAccounts {
private String name;
private EventBus eventBus;
public WeChatAccounts(String name) {
this.name = name;
this.eventBus = new EventBus();
}
public void publishArticles(String articleName, String content) {
System.out.println(String.format("\n<%s>微信公众号 发布了一篇推送,文章名称为 <%s>,内容为 <%s> ", this.name, articleName, content));
//post方法将会将被派发的消息,传递给所有的订阅者,并调用订阅者的监听方法
this.eventBus.post(new WechatNotice(this.name, articleName));
}
public void register(WeChatClient weChatClient) {
this.eventBus.register(weChatClient);
}
public void unregister(WeChatClient weChatClient) {
this.eventBus.unregister(weChatClient);
}
}
测试
public class Test {
public static void main(String[] args) {
WeChatAccounts accounts = new WeChatAccounts("小旋锋");
WeChatClient user1 = new WeChatClient("张三");
WeChatClient user2 = new WeChatClient("李四");
WeChatClient user3 = new WeChatClient("王五");
accounts.register(user1);
accounts.register(user2);
accounts.register(user3);
accounts.publishArticles("设计模式 | 观察者模式及典型应用", "观察者模式的内容...");
accounts.unregister(user1);
accounts.publishArticles("设计模式 | 单例模式及典型应用", "单例模式的内容....");
}
}
spring
的事件机制是从java的事件机制拓展而来,ApplicationContext
中事件处理是由 ApplicationEvent
类和 ApplicationListener
接口来提供的。如果一个Bean
实现了 ApplicationListener
接口,并且已经发布到容器中去,每次 ApplicationContext
发布一个 ApplicationEvent
事件,这个Bean就会接到通知
使用 spring 事件机制重新实现示例
@Data
public class WechatNotice extends ApplicationEvent {
private String publisher;
private String articleName;
public WechatNotice(Object source, String publisher, String articleName) {
super(source);
this.publisher = publisher;
this.articleName = articleName;
}
}
public class WeChatClient implements ApplicationListener {
private String username;
public WeChatClient(String username) {
this.username = username;
}
@Override
public void onApplicationEvent(ApplicationEvent event) {
if (event instanceof WechatNotice) {
WechatNotice notice = (WechatNotice) event;
System.out.println(String.format("用户<%s> 接收到 <%s>微信公众号 的推送,文章标题为 <%s>", username, notice.getPublisher(), notice.getArticleName()));
}
}
public void setUsername(String username) {
this.username = username;
}
}
public class WeChatAccounts implements ApplicationContextAware {
private ApplicationContext ctx;
private String name;
public WeChatAccounts(String name) {
this.name = name;
}
public void setName(String name) {
this.name = name;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.ctx = applicationContext;
}
public void publishArticles(String articleName, String content) {
System.out.println(String.format("\n<%s>微信公众号 发布了一篇推送,文章名称为 <%s>,内容为 <%s> ", this.name, articleName, content));
ctx.publishEvent(new WechatNotice(this.name, this.name, articleName));
}
}
在 resources 目录下创建 spring.xml
文件,填入下面的内容
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="WeChatAccounts" class="com.observer.sprintevent.WeChatAccounts" scope="prototype">
<constructor-arg name="name" value=""></constructor-arg>
</bean>
<bean id="WeChatClient1" class="com.observer.sprintevent.WeChatClient">
<constructor-arg name="username" value="张三"></constructor-arg>
</bean>
<bean id="WeChatClient2" class="com.observer.sprintevent.WeChatClient">
<constructor-arg name="username" value="李四"></constructor-arg>
</bean>
<bean id="WeChatClient3" class="com.observer.sprintevent.WeChatClient">
<constructor-arg name="username" value="王五"></constructor-arg>
</bean>
</beans>
测试
public class Test {
public static void main(String[] args) {
ApplicationContext context = new ClassPathXmlApplicationContext("spring.xml");
WeChatAccounts accounts = (WeChatAccounts) context.getBean("WeChatAccounts");
accounts.setName("大忽悠");
accounts.setApplicationContext(context);
accounts.publishArticles("设计模式 | 观察者模式及典型应用", "观察者模式的内容...");
}
}
在此示例中 ApplicationContext
对象的实际类型为 ClassPathXmlApplicationContext
,其中的与 publishEvent
方法相关的主要代码如下:
private ApplicationEventMulticaster applicationEventMulticaster;
public void publishEvent(ApplicationEvent event) {
this.getApplicationEventMulticaster().multicastEvent(event);
if (this.parent != null) {
this.parent.publishEvent(event);
}
}
ApplicationEventMulticaster getApplicationEventMulticaster() throws IllegalStateException {
return this.applicationEventMulticaster;
}
protected void initApplicationEventMulticaster() {
ConfigurableListableBeanFactory beanFactory = this.getBeanFactory();
if (beanFactory.containsLocalBean("applicationEventMulticaster")) {
this.applicationEventMulticaster = (ApplicationEventMulticaster)beanFactory.getBean("applicationEventMulticaster", ApplicationEventMulticaster.class);
} else {
this.applicationEventMulticaster = new SimpleApplicationEventMulticaster(beanFactory);
beanFactory.registerSingleton("applicationEventMulticaster", this.applicationEventMulticaster);
}
}
其中的 SimpleApplicationEventMulticaster
如下,multicastEvent
方法主要是通过遍历 ApplicationListener
(注册由 AbstractApplicationEventMulticaster
实现),使用线程池框架 Executor
来并发执行 ApplicationListener
的 onApplicationEvent
方法,与示例本质上是一致的
public class SimpleApplicationEventMulticaster extends AbstractApplicationEventMulticaster {
private Executor taskExecutor;
public void multicastEvent(final ApplicationEvent event) {
Iterator var2 = this.getApplicationListeners(event).iterator();
while(var2.hasNext()) {
final ApplicationListener listener = (ApplicationListener)var2.next();
Executor executor = this.getTaskExecutor();
if (executor != null) {
executor.execute(new Runnable() {
public void run() {
listener.onApplicationEvent(event);
}
});
} else {
listener.onApplicationEvent(event);
}
}
}
}