观察者模式
场景描述
先看一个关注微信公众号的业务，关注完微信公众号之后要进行如下操作：
1、记录文本日志
2、记录数据库日志
3、发送短信
4、送优惠券，积分等
5、其他各类活动等
传统解决方案:
在关注公众号逻辑等类内部增加相关代码,完成各种逻辑。
存在问题:
1、一旦某个业务逻辑发生改变,如关注公众号业务中增加其他业务逻辑,需要修改关注公众号核心文件、甚至关注流程。
2、日积月累后,文件冗长,导致后续维护困难。
存在问题原因主要是程序的"紧密耦合",使用观察模式可以将目前的业务逻辑优化成"松耦合",达到易维护、易修改的目的,同时也符合面向接口编程的思想。
什么是观察者模式
观察者模式(Observer Pattern)：定义对象间的一种一对多依赖关系，使得每当一个对象状态发生改变时，其相关依赖对象皆得到通知并被自动更新。观察者模式又叫做发布-订阅（Publish/Subscribe）模式。观察者模式是一种对象行为型模式。
各个字段含义如下：
- Subject: 目标，抽象被观察者。 抽象主题角色把所有观察者对象保存在一个集合里，每个主题都可以有任意数量的观察者，抽象主题提供一个接口，可以增加和删除观察者对象。
- ConcreteSubject: 具体目标，具体的被观察者。 该角色将有关状态存入具体观察者对象，在具体主题的内部状态发生改变时，给所有注册过的观察者发送通知。
- Observer: 观察者。 是观察者者的抽象类，它定义了一个更新接口，使得在得到主题更改通知时更新自己。
- ConcreteObserver: 具体观察者。实现抽象观察者定义的更新接口，以便在得到主题更改通知时更新自身的状态。
观察者模式的简单实现
观察者模式这种发布-订阅的形式我们可以拿微信公众号来举例，假设微信用户就是观察者，微信公众号是被观察者，有多个的微信用户关注了程序猿这个公众号，当这个公众号更新时就会通知这些订阅的微信用户。看看用代码如何实现：
抽象观察者（Observer）
//抽象观察者
public interface Observer {
//收到订阅消息执行的操作
void update(String message);
}
具体观察者（ConcrereObserver）
//记录日志
public class LogObserver implements Observer{
@Override
public void update(String message) {
System.out.println("日志订阅---"+message+"---开始记录日志");
}
}
//优惠券订阅
public class TicketObserver implements Observer{
@Override
public void update(String message) {
System.out.println("优惠券订阅---"+message+"---开始发送优惠券");
}
}
抽象被观察者（Subject)
//抽象被观察者
public interface Subject {
/**
- 增加订阅者
- @param observer
*/
void attach(Observer observer);
/**
- 删除订阅者
- @param observer
*/
void detach(Observer observer);
/**
- 通知订阅者
*/
void notify(String message);
}
具体被观察者（ConcreteSubject）
//订阅公众号
public class SubscriptionSubject implements Subject {
//储存订阅公众号的业务
private List<Observer> list= new ArrayList<Observer>();
@Override
public void attach(Observer observer) {
list.add(observer);
}
@Override
public void detach(Observer observer) {
list.remove(observer);
}
@Override
public void notify(String message) {
for (Observer observer : list) {
observer.update(message);
}
}
public void subscribe(){
System.out.println("执行订阅的具体操作");
notify("订阅成功了");
}
}
客户端调用
public class Client {
public static void main(String[] args) {
SubscriptionSubject subscriptionSubject = new SubscriptionSubject();
//添加日志订阅
Observer logObserver = new LogObserver();
subscriptionSubject.attach(logObserver);
//添加优惠券订阅
TicketObserver ticketObserver = new TicketObserver();
subscriptionSubject.attach(ticketObserver);
//执行订阅的操作
subscriptionSubject.subscribe();
}
}
执行结果
执行订阅的具体操作
日志订阅---订阅成功了---开始记录日志
优惠券订阅---订阅成功了---开始发送优惠券
使用观察者模式的场景和优缺点
- 优点
解除耦合，让耦合的双方都依赖于抽象，从而使得各自的变换都不会影响另一边的变换。
- 缺点
在应用观察者模式时需要考虑一下开发效率和运行效率的问题，程序中包括一个被观察者、多个观察者，开发、调试等内容会比较复杂，而且在Java中消息的通知一般是顺序执行，那么一个观察者卡顿，会影响整体的执行效率，在这种情况下，一般会采用异步实现。
java中的实现
JDK 提供了 一套 观察者模式的实现，在java.util包中， java.util.Observable类和java.util.Observer接口。Observable是被观察者，Observer是观察者。
推模型和拉模型
在观察者模式中，又分为推模型和拉模型两种方式。
- 推模型
主题对象向观察者推送主题的详细信息，不管观察者是否需要，推送的信息通常是主题对象的全部或部分数据。
- 拉模型
主题对象在通知观察者的时候，只传递少量信息。如果观察者需要更具体的信息，由观察者主动到主题对象中获取，相当于是观察者从主题对象中拉数据。一般这种模型的实现中，会把主题对象自身通过update()方法传递给观察者，这样在观察者需要获取数据的时候，就可以通过这个引用来获取了。
根据上面的描述，发现上面的例子就是典型的推模型，下面给出一个拉模型的实例。
拉模型的抽象主题类
拉模型的抽象主题类主要的改变是nodifyObservers()方法。在循环通知观察者的时候，也就是循环调用观察者的update()方法的时候，传入的参数不同了。
public abstract class Subject {
/**
- 用来保存注册的观察者对象
*/
private List<Observer> list = new ArrayList<Observer>();
/**
- 注册观察者对象
@param observer 观察者对象
*/
public void attach(Observer observer) {
list.add(observer);
System.out.println("Attached an observer");
}
/**
- 删除观察者对象
@param observer 观察者对象
*/
public void detach(Observer observer) {
list.remove(observer);
}
/**
- 通知所有注册的观察者对象
*/
public void nodify() {
for (Observer observer : list) {
observer.update(this);
}
}
}
拉模型的具体主题类
跟推模型相比，有一点变化，就是调用通知观察者的方法的时候，不需要传入参数了。
public class ConcreteSubject extends Subject {
private String state;
public String getState() {
return state;
}
public void change(String newState) {
state = newState;
System.out.println("主题状态为：" + state);
//状态发生改变，通知各个观察者
this.nodify();
}
}
拉模型的抽象观察者类
拉模型通常都是把主题对象当做参数传递。
public interface Observer {
/**
- 更新接口
@param subject 传入主题对象，方面获取相应的主题对象的状态
*/
void update(Subject subject);
}
拉模型的具体观察者类
public class ConcreteObserver implements Observer {
//观察者的状态
private String observerState;
@Override
public void update(Subject subject) {
/**
- 更新观察者的状态，使其与目标的状态保持一致
*/
observerState = ((ConcreteSubject)subject).getState();
System.out.println("观察者状态为："+observerState);
}
}
两种模式的比较
推模型是假定主题对象知道观察者需要的数据；而拉模型是主题对象不知道观察者具体需要什么数据，没有办法的情况下，干脆把自身传递给观察者，让观察者自己去按需要取值。
推模型可能会使得观察者对象难以复用，因为观察者的update()方法是按需要定义的参数，可能无法兼顾没有考虑到的使用情况。这就意味着出现新情况的时候，就可能提供新的 update()方法，或者是干脆重新实现观察者；而拉模型就不会造成这样的情况，因为拉模型下，update()方法的参数是主题对象本身，这基本上是主题对象能传递的最大数据集合了，基本上可以适应各种情况的需要。
RxJava
场景描述
假设有这样一个需求：界面上有一个自定义的视图 imageCollectorView （图片展示），它的作用是显示多张图片，并能使用 addImage(Bitmap) 方法来任意增加显示的图片。现在需要程序将一个给出的目录数组 File[] folders 中每个目录下的 png 图片都加载出来并显示在 imageCollectorView 中。由于读取图片的这一过程较为耗时，需要放在后台执行，而图片的显示则必须在 UI 线程执行。常见的实现方式，在后台开启一个线程，执行加载图片的操作。代码如下：
new Thread() {
@Override
public void run() {
super.run();
for (File folder : folders) {
File[] files = folder.listFiles();
for (File file : files) {
if (file.getName().endsWith(".png")) {
final Bitmap bitmap = getBitmapFromFile(file);
getActivity().runOnUiThread(new Runnable() {
@Override
public void run() {
imageCollectorView.addImage(bitmap);
}
});
}
}
}
}
}.start();
而如果使用 RxJava ，实现方式是这样的：
Observable.from(folders)
.flatMap(new Func1<File, Observable<File>>() {
@Override
public Observable<File> call(File file) {
return Observable.from(file.listFiles());
}
})
.filter(new Func1<File, Boolean>() {
@Override
public Boolean call(File file) {
return file.getName().endsWith(".png");
}
})
.map(new Func1<File, Bitmap>() {
@Override
public Bitmap call(File file) {
return getBitmapFromFile(file);
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Bitmap>() {
@Override
public void call(Bitmap bitmap) {
imageCollectorView.addImage(bitmap);
}
});
从中可以看出rxjava代码逻辑的简洁。rxjava代码简洁不是指代码量少，而是指代码逻辑的简洁。
什么是RxJava
原文 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。这就是 RxJava ，概括得非常精准。
其实， RxJava 的本质可以压缩为异步这一个词。说到根上，它就是一个实现异步操作的库。
RxJava的特点
异步操作很关键的一点是程序的简洁性，因为在调度过程比较复杂的情况下，异步代码经常会既难写也难被读懂。 Android 创造的 AsyncTask 和Handler ，其实都是为了让异步代码更加简洁。RxJava 的优势也是简洁，但它的简洁的与众不同之处在于，随着程序逻辑变得越来越复杂，它依然能够保持简洁。
RxJava的 观察者模式
RxJava 的异步实现，是通过一种扩展的观察者模式来实现的。
RxJava 的观察者模式大致如下图：
RxJava 有四个基本概念：
- Observable (被观察者) 产生事件
- Observer (观察者) 接收事件，并给出响应动作
- subscribe (订阅) 连接 被观察者 & 观察者
- 事件 被观察者 & 观察者 沟通的载体
Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知Observer。
与传统观察者模式不同， RxJava 的事件回调方法除了普通事件 onNext() （相当于 onClick() / onEvent()）之外，还定义了两个特殊的事件：onCompleted() 和 onError()。
- onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。
- onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。
RxJava的实现
基于以上的概念， RxJava 的基本实现主要有三点：
- 创建 观察者Observer
Observer 即观察者，它决定事件触发的时候将有怎样的行为。 RxJava 中的 Observer 接口的实现方式：
Observer<String> observer = new Observer<String>() {
@Override
public void onNext(String s) {
Log.d(tag, "Item: " + s);
}
@Override
public void onCompleted() {
Log.d(tag, "Completed!");
}
@Override
public void onError(Throwable e) {
Log.d(tag, "Error!");
}
};
- 创建 被观察者Observable
Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。 RxJava 使用 create() 方法来创建一个 Observable ，并为它定义事件触发规则：
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello");
subscriber.onNext("Hi");
subscriber.onNext("Aloha");
subscriber.onCompleted();
}
});
可以看到，这里传入了一个 OnSubscribe 对象作为参数。OnSubscribe 会被存储在返回的 被观察者Observable 对象中，它的作用相当于一个计划表，当被观察者Observable 被订阅的时候，OnSubscribe 的 call() 方法会自动被调用，事件序列就会依照设定依次触发（对于上面的代码，就是观察者Subscriber 将会被调用三次 onNext() 和一次 onCompleted()）。这样，由被观察者调用了观察者的回调方法，就实现了由被观察者向观察者的事件传递，即观察者模式。
create() 方法是 RxJava 最基本的创造事件序列的方法。基于这个方法， RxJava 还提供了一些方法用来快捷创建事件队列，例如：
- just(T...): 将传入的参数依次发送出来。
Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
- from(T[]) / from(Iterable<? extends T>) : 将传入的数组或 Iterable 拆分成具体对象后，依次发送出来。
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
- Subscribe (订阅)
创建了被观察者 Observable 和观察者 Observer 之后，再用 subscribe() 方法将它们联结起来，整条链子就可以工作了。代码形式很简单：
observable.subscribe(observer);
// 或者：
observable.subscribe(subscriber);
整个过程中对象间的关系如下图：
基于事件流的链式调用
上述的实现方式是为了说明Rxjava的原理 & 使用
在实际应用中，会将上述步骤&代码连在一起，从而更加简洁、更加优雅，即所谓的 RxJava基于事件流的链式调用
// RxJava的链式操作
Observable.create(new ObservableOnSubscribe<Integer>() {
// 1. 创建被观察者 & 生产事件
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
// 2. 通过通过订阅（subscribe）连接观察者和被观察者
// 3. 创建观察者 & 定义响应事件的行为
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）
@Override
public void onNext(Integer value) {
Log.d(TAG, "对Next事件"+ value +"作出响应" );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
这种 基于事件流的链式调用，使得RxJava：
- 逻辑简洁
- 实现优雅
- 使用简单
更重要的是，随着程序逻辑的复杂性提高，它依然能够保持简洁 & 优雅。所以，一般建议使用这种基于事件流的链式调用方式实现RxJava。
线程控制
在 RxJava 的默认规则中，事件的发出和消费都是在同一个线程的。也就是说，如果只用上面的方法，实现出来的只是一个同步的观察者模式。观察者模式本身的目的就是『后台处理，前台回调』的异步机制，因此异步对于 RxJava 是至关重要的。而要实现异步，则需要用到 RxJava 的另一个概念： Scheduler 。
在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。
在RxJava 中，Scheduler ——调度器，相当于线程控制器，RxJava 通过它来指定每一段代码应该运行在什么样的线程。
略...
感兴趣可参考 给 Android 开发者的 RxJava 详解
场景示例
-
需求场景
- 功能逻辑
- 代码实现
public class RxJavaRetry {
private static final String TAG = "RxJava";
// 可重试次数
private int maxConnectCount = 10;
// 当前已重试次数
private int currentRetryCount = 0;
// 重试等待时间
private int waitRetryTime = 0;
@Override
protected void onCreate(Bundle savedInstanceState) {
//省略
// 步骤3：采用Observable<...>形式 对 网络请求 进行封装
Observable<Translation> observable = request.getCall();
// 步骤4：发送网络请求 & 通过retryWhen（）进行重试
// 注：主要异常才会回调retryWhen（）进行重试
observable.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {
// 参数Observable<Throwable>中的泛型 = 上游操作符抛出的异常，可通过该条件来判断异常的类型
return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {
// 输出异常信息
Log.d(TAG, "发生异常 = "+ throwable.toString());
/**
- 需求1：根据异常类型选择是否重试
- 即，当发生的异常 = 网络异常 = IO异常 才选择重试
*/
if (throwable instanceof IOException){
Log.d(TAG, "属于IO异常，需重试" );
/**
- 需求2：限制重试次数
- 即，当已重试次数 < 设置的重试次数，才选择重试
*/
if (currentRetryCount < maxConnectCount){
// 记录重试次数
currentRetryCount++;
/**
- 需求2：实现重试
- 通过返回的Observable发送的事件 = Next事件，从而使得retryWhen（）重订阅，最终实现重试功能
需求3：延迟1段时间再重试
- 采用delay操作符 = 延迟一段时间发送，以实现重试间隔设置
需求4：遇到的异常越多，时间越长
- 在delay操作符的等待时间内设置 = 每重试1次，增多延迟重试时间1s
*/
// 设置等待时间
waitRetryTime = 1000 + currentRetryCount* 1000;
return Observable.just(1).delay(waitRetryTime, TimeUnit.MILLISECONDS);
}else{
// 若重试次数已 > 设置重试次数，则不重试
// 通过发送error来停止重试（可在观察者的onError（）中获取信息）
return Observable.error(new Throwable("重试次数已超过设置次数 = " +currentRetryCount + "，即 不再重试"));
}
}else{
// 若发生的异常不属于I/O异常，则不重试
// 通过返回的Observable发送的事件 = Error事件 实现（可在观察者的onError（）中获取信息）
return Observable.error(new Throwable("发生了非网络异常（非I/O异常）"));
}
}
});
}
}).subscribeOn(Schedulers.io()) // 切换到IO线程进行网络请求
.observeOn(AndroidSchedulers.mainThread()) // 切换回到主线程 处理请求结果
.subscribe(new Observer<Translation>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Translation result) {
// 接收服务器返回的数据
Log.d(TAG, "发送成功");
result.show();
}
@Override
public void onError(Throwable e) {
// 获取停止重试的信息
Log.d(TAG, e.toString());
}
@Override
public void onComplete() {
}
});
}
}
更多场景可以参考 使用RxJava的最佳开发场景
HYSTRIX中的观察者模式
Hystrix中的命令
Hystrix有两个请求命令 HystrixCommand、HystrixObservableCommand。
HystrixCommand用在依赖服务返回单个操作结果的时候。又两种执行方式
- execute():同步执行。从依赖的服务返回一个单一的结果对象，或是在发生错误的时候抛出异常。
- queue();异步执行。直接返回一个Future对象，其中包含了服务执行结束时要返回的单一结果对象。
HystrixObservableCommand 用在依赖服务返回多个操作结果的时候。它也实现了两种执行方式
- observe():返回Obervable对象，他代表了操作的多个结果，他是一个HotObservable
- toObservable():同样返回Observable对象，也代表了操作多个结果，但它返回的是一个Cold Observable。
在Hystrix的底层实现中大量使用了RxJava。上面提到的Observable对象就是RxJava的核心内容之一，可以把Observable对象理解为事件源或是被观察者，与其对应的是Subscriber对象，可以理解为订阅者或是观察者。
- Observable用来向订阅者Subscriber对象发布事件，Subscriber对象在接收到事件后对其进行处理，这里所指的事件通常就是对依赖服务的调用。
- 一个Observable可以发出多个事件，直到结束或是发生异常。
- Observable对象每发出一个事件，就会调用对应观察者Subscriber对象的onNext()方法。
- 每一个Observable的执行，最后一定会通过调用Subscriber.onCompleted()或是Subscriber.onError()来结束该事件的操作流。
调用实例
public class HelloWorldHystrixObservableCommand extends HystrixObservableCommand<String> {
private final String name;
protected HelloWorldHystrixObservableCommand(String name) {
super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
this.name = name;
}
@Override
protected Observable<String> construct() {
System.out.println("in construct! thread:" + Thread.currentThread().getName());
return (Observable<String>) Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> observer) {
try {
System.out.println("in call of construct! thread:" + Thread.currentThread().getName());
if (!observer.isUnsubscribed()) {
// 直接抛异常退出，不会往下执行
// observer.onError(getExecutionException());
observer.onNext("Hello1" + " thread:" + Thread.currentThread().getName());
observer.onNext("Hello2" + " thread:" + Thread.currentThread().getName());
observer.onNext(name + " thread:" + Thread.currentThread().getName());
System.out.println("complete before------" + " thread:" + Thread.currentThread().getName());
// 不会往下执行observer的任何方法
observer.onCompleted();
System.out.println("complete after------" + " thread:" + Thread.currentThread().getName());
// 不会执行到
observer.onCompleted();
// 不会执行到
observer.onNext("abc");
}
} catch (Exception e) {
observer.onError(e);
}
}
});
}
public static void main(String[] args) {
Observable<String> observable = new HelloWorldHystrixObservableCommand("test").observe();
observable.subscribe(new Subscriber<String>() {
public void onCompleted() {
System.out.println("completed");
}
public void onError(Throwable throwable) {
System.out.println("error-----------" + throwable);
}
public void onNext(String v) {
System.out.println("next------------" + v);
}
});
}
}
执行结果
in construct! thread:main
in call of construct! thread:main
complete before------ thread:main
complete after------ thread:main
next------------Hello1 thread:main
next------------Hello2 thread:main
next------------test thread:main
completed
