文章目录
一. 观察者模式1.1 分析关系图1. 2 代码实现
二. RxJava2.1 简介2.2 使用步骤
三. Schedulers线程控制3.1 耗时操作切换到子线程3.2 更新ui切换回主线程
一. 观察者模式
定义：观察者(Observer)模式:是对象的行为模式，又叫做发布-订阅 (Publish/Subscribe)模式、模型-视图 (Model/View) 模式、源-监听 (Source/Listener)模式。观察者模式定义了一种一对多的依赖关系，让多个观察者对象同时监听某个对象，这个对象在状态上发生变化时，会通知所有观察者对象使它们能够自动更新自己。
如下UML图：
1.1 分析关系图
我们先来分析一下这个关系图
我们要先抽象一个Observer接口，它里面有个抽象方法update()，这个方法的作用就是通知观察者更新自己。ConcrereObserver是它的具体实现子类，在子类中会实现这个update()方法，然后在当前子类更新我们自己的状态。Subject就是我们要监听的那个对象，可以这么理解，attach()方法是绑定观察者，detach()移除观察者，notify()是把当前这个对象的变化通知给观察者，然后观察者根据这些变化，做对应的事情。ConcreteSubject是Subject的子类，子类中我们可以自定义一些特有的方法，比如我们加上一个子类独有的change()方法，更新自己的状态，只要子类发生变化就可以调用父类中的notify()方法，通知到Observer处理。观察者Observer，一般我们是在Subject这个对象中建立对应的依赖关系，关系是一对多的关系，一个Subject可以对应多个观察者。
对上面的内容小结一下：
Subject：这个类就是把观察者的引用进行依赖关联，比如把观察者放在集合或者数组里边，每一个对象都可以有任意数目的观察者，这个被观察的对象可以被称为被观察者Observable，被观察者可以增删观察者对象，比如我们上面说到的attach()和detach()。ConcreteSubject：这个是被观察者的子类，在具体的对象也就是当前这个子类对象的某些状态发生改变的时候，给绑定了该对象的观察者发送通知，通过调用notify()方法，它也算是一个被观察者，因为我们实现了父类抽象类Subject，具体的一些行为都是在子类中去实现了的。Observer：为所有的具体观察者提供一个统一抽象接口，具体观察者意思就是其对应的具体实现子类，在我们被观察者也就是Subject的子类发送来的通知时，更新自己。ConcrereObserver：就是观察者的子类，每个子类都会有自己对应的实现，对update()方法，也就是说每个子类收到通知之后更新自己的状态是不受其他子类的影响的。
1. 2 代码实现
/**
* 观察者抽象接口
*/
public interface Observer {
//将状态的更新通知给观察者具体的实现类
void update(String state);
}
/**
* 被观察者
*/
public abstract class Subject {
//存放观察者对象引用
private List
public void attach(Observer observer) {
observers.add(observer);
System.out.println("绑定观察者对象");
}
public void detach(Observer observer) {
observers.remove(observer);
System.out.println("解除绑定观察者");
}
/**
* 通知观察者更新状态
*/
public void notifyObserver(String state) {
for (Observer observer : observers) {
observer.update(state);
}
}
}
/**
* 观察者具体的实现子类1
*/
public class SubOneObserverImpl implements Observer {
@Override
public void update(String state) {
System.out.println("SubOneObserverImpl观察者接收到状态改变的消息：" + state);
}
}
/**
* 观察者具体的实现子类2
*/
public class SubTwoObserverImpl implements Observer {
@Override
public void update(String state) {
System.out.println("SubTwoObserverImpl观察者接收到状态改变的消息：" + state);
}
}
/**
* 具体实现的被观察者子类
*/
public class SubOneSubject extends Subject {
private String state = "subOneSubject";
//发通知给绑定的观察者对象
public void change(String state) {
notifyObserver(state);
}
}
public class Test {
public static void main(String[] args) {
Observer observer1 = new SubOneObserverImpl();
Observer observer2 = new SubTwoObserverImpl();
SubOneSubject subject = new SubOneSubject();
subject.attach(observer1);
subject.attach(observer2);
subject.change("SubOneSubject");
}
}
看看代码的运行结果：
二. RxJava
2.1 简介
Observable: 被观察者 (Subject)。Observer/Subscriber: 观察者。Subscribe:订阅。Observable 和 Observer 通过 subscribe0)方法实现订阅关系，可以理解为绑定关联起来。
2.2 使用步骤
使用前提：要先添加依赖库
implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
implementation 'io.reactivex.rxjava3:rxjava:3.0.0'
创建Observable(被观察者)
Observable
@Override
public void subscribe(@NonNull ObservableEmitter
emitter.onNext("hello");
emitter.onNext("world");
emitter.onComplete();
}
});
创建Observer
Observer
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e(TAG, "onSubscribe: " + d.isDisposed());
}
@Override
public void onNext(@NonNull String s) {
Log.e(TAG, "onNext: " + s);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError: " + e.getLocalizedMessage());
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
};
订阅
observable.subscribe(observer);
我们来看看一个Demo运行的效果
先写一个xml布局文件
xmlns:app="http://schemas.android.com/apk/res-auto" xmlns:tools="http://schemas.android.com/tools" android:layout_width="match_parent" android:layout_height="match_parent" android:orientation="vertical" tools:context=".MainActivity"> android:id="@+id/btn" android:layout_width="match_parent" android:layout_height="wrap_content" android:text="test" /> android:id="@+id/tv" android:layout_width="wrap_content" android:layout_height="wrap_content" android:text="Hello World!" android:textSize="40sp" />
测试代码
public class MainActivity extends AppCompatActivity {
private static final String TAG = MainActivity.class.getSimpleName();
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Button button = findViewById(R.id.btn);
TextView textView = findViewById(R.id.tv);
Observable
@Override
public void subscribe(@NonNull ObservableEmitter
emitter.onNext("hello");
emitter.onNext("world");
emitter.onComplete();
}
});
Observer
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e(TAG, "onSubscribe: " + d.isDisposed());
}
@Override
public void onNext(@NonNull String s) {
Log.e(TAG, "onNext: " + s);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError: " + e.getLocalizedMessage());
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
};
//设置一个按钮监听事件，点击之后就进行订阅。
button.setOnClickListener(v -> {
observable.subscribe(observer);
});
}
}
看看运行的结果如何
好，以上就是rxjava的基本入门写法。
三. Schedulers线程控制
Android是不能阻塞UI线程的，所以比较耗时的一些操作我们都应该放到一个不是UI线程的工作线程去处理。
常用函数：
Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。Schedulers.io(): I/O操作(读写文件、读写数据库、网络信息交互等)所使用的 Scheduler。和newThread()差不多， 但是io()的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下io()比newThread()更有效率。不要把计算工作放在io()中，可以避免创建不必要的线程。Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是CPU密集型计算，即不会被I/O等操作限制性能的操作，这个 Scheduler 使用的固定的线程池，大小为 CPU核数。不要把I/O操作放在computation()中，否则I/O 操作的等待时间会浪费 CPU。AndroidSchedulers.mainThread(): 它指定的操作将在Android 主线程运行。
我们用代码来实现一下网络请求线程的切换，看看是怎么样的
首先我们使用Retrofit框架进行对网络请求数据访问，Retrofit这个框架在本文不进行细节分析，大伙儿可以先去了解了解怎么使用，这里我就简单介绍一下用法在Demo中。
先添加对应的依赖
implementation 'com.squareup.retrofit2:retrofit:2.4.0'
implementation 'com.squareup.retrofit2:converter-gson:2.0.2'
定义一个接口，该接口的作用可以理解为，存放对应访问服务器的具体Url路径，已经传递对应的参数。
public interface TestApi {
//定义登录的方法，并通过@Filed注解配置请求体参数
@FormUrlEncoded//会对请求体的数据进行url编码
//请求登录所需要的两个参数，用户名和密码
@POST("login")
Call
}
在MainActivity中写测试代码
public class MainActivity extends AppCompatActivity {
private static final String TAG = MainActivity.class.getSimpleName();
private Retrofit retrofit;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Button button = findViewById(R.id.btn);
TextView textView = findViewById(R.id.tv);
retrofit = new Retrofit.Builder()
//我们使用鸿洋大神的wanAndroid的开放api进行网络请求测试
.baseUrl("https://www.wanandroid.com/user/")
.addConverterFactory(GsonConverterFactory.create())
.build();
TestApi testApi = retrofit.create(TestApi.class);
//这是我的账号，大伙儿先整一个账号来使用。
Call
//按钮点击监听事件
button.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
Observable.create(new ObservableOnSubscribe
@Override
public void subscribe(@NonNull ObservableEmitter
Response
try {
execute = login.execute();
} catch (Exception e) {
throw new RuntimeException(e);
}
emitter.onNext(execute.body());
emitter.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull ResponseBody responseBody) {
String responseData = null; // 使用保存的结果
try {
responseData = responseBody.string();
} catch (Exception e) {
throw new RuntimeException(e);
}
Log.e(TAG, "onNextqqfh: " + responseData);
textView.setText(responseData);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "An error occurred: " + e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
});
}
});
}
}
对应的xml文件
xmlns:app="http://schemas.android.com/apk/res-auto" xmlns:tools="http://schemas.android.com/tools" android:layout_width="match_parent" android:layout_height="match_parent" android:orientation="vertical" tools:context=".MainActivity"> android:id="@+id/btn" android:layout_width="match_parent" android:layout_height="wrap_content" android:text="test" /> android:id="@+id/tv" android:layout_width="wrap_content" android:layout_height="wrap_content" android:text="Hello World!" android:textSize="20sp" />
运行结果：
从截图的结果我们可以看出，请求网络之后，给我们返回的字符串我们显示在了UI上，说明我们这一次的网络请求是成功的。
==========================================
3.1 耗时操作切换到子线程
下面然后我们来看一个错误的示例，我们稍微改下代码：
button.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
Observable.create(new ObservableOnSubscribe
@Override
public void subscribe(@NonNull ObservableEmitter
Response
try {
execute = login.execute();
} catch (Exception e) {
throw new RuntimeException(e);
}
emitter.onNext(execute.body());
emitter.onComplete();
}
}).subscribe(new Observer
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull ResponseBody responseBody) {
String responseData = null; // 使用保存的结果
try {
responseData = responseBody.string();
} catch (Exception e) {
throw new RuntimeException(e);
}
Log.e(TAG, "onNextqqfh: " + responseData);
textView.setText(responseData);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "An error occurred: " + e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
});
}
});
我们看下结果很明显，我们的UI没有更新出对应的请求网络的数据信息，而且大家仔细观察日志就会发现，我们订阅完之后，执行的是onError方法，前面我们提到，onError和onComplete方法是互斥的，二者只会执行一个，我们这里输出了这条日志Log.e(TAG, "An error occurred: " + e.getMessage());，说明代码执行过程中发生了错误，抛出了异常，但是我们捕获了异常，所以程序没有崩溃，而且日志并没有执行onComplete()方法。
这是为什么呢？我们研究一下
异常信息也很明显，意思就是不能在主线程调用网络请求这种耗时间的操作。从这个异常我们也可以反推出在subscribe()中是在Android的主线程里面执行的代码，那我们应该加上这行代码。
//将网络请求切换至IO线程处理
.subscribeOn(Schedulers.io())
Schedulers.io()：会开启一个线程池，在子线程去调用我们的网络请求的代码。
加上这行代码后我们再运行一下看看。
button.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
Observable.create(new ObservableOnSubscribe
@Override
public void subscribe(@NonNull ObservableEmitter
Response
try {
execute = login.execute();
} catch (Exception e) {
throw new RuntimeException(e);
}
emitter.onNext(execute.body());
emitter.onComplete();
}
}).subscribeOn(Schedulers.io())
.subscribe(new Observer
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull ResponseBody responseBody) {
String responseData = null; // 使用保存的结果
try {
responseData = responseBody.string();
} catch (Exception e) {
throw new RuntimeException(e);
}
textView.setText(responseData);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "An error occurred: " + e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
});
}
});
看看改了之后结果：
我们发现还是抛出了异常信息，并且我们的UI确实也没有得到更新，这个异常意思是必须要在Android的主线程去更新UI，所以我们要加上这句代码即可更新UI。
3.2 更新ui切换回主线程
.observeOn(AndroidSchedulers.mainThread())
大家加上这行代码之后可以自己运行看看。
小结：
我们不加subscribeOn这句代码切换线程的时候，我们的代码其实都是在Android的主线程里面运行的。加上subscribeOn后，我们的代码都会被切换到子线程去执行，然后更新UI的时候可以只把观察者切换回主线程更新UI，就是这行代码observeOn，意思就是不能让观察者和被观察者处于同一个线程中。observeOn()指定观察者线程，subscribeOn()指定被观察者线程。
相关阅读
您阅读本篇文章共花了：
发表评论