目录
- Rxjava是什么
- 如何使用
- Rxjava的知识点
- 使用场景
一、是什么?
RxJava是一个基于响应式编程(Reactive Programming)和观察者模式(Observer pattern)的Java库。它提供了一种异步、基于事件流的编程方式,旨在简化复杂的异步编程任务。
RxJava的使用场景非常广泛,尤其是在需要处理异步操作和事件流的场景下,常用于网络请求处理。通过Observable封装网络请求,并使用Observer来接收请求的结果或错误,可以轻松地将网络请求的异步性转化为基于事件的流处理。这样不仅可以避免回调地狱(Callback Hell),还可以更方便地处理多个并行或连续的网络请求。
二、如何使用?
(1)引入依赖
implementation 'io.reactivex.rxjava2:rxjava:2.2.2'
(2)定义 Observer 观察者:用于处理返回来的数据
Observer<String> observer = new Observer<String>() {@Overridepublic void onSubscribe(Disposable d) {// 当观察者订阅时的回调}@Overridepublic void onNext(String value) {// 当接收到新的事件时的回调}@Overridepublic void onError(Throwable e) {// 当发生错误时的回调}@Overridepublic void onComplete() {// 当事件流结束时的回调}
};
(3)定义 Observable 被观察者:用来发送数据
Observable<String> observable = Observable.just("Hello", "Rxjava");
(4)Observable 被观察者订阅 Observer 观察者:订阅
observable.subscribe(observer);
三、Rxjava的知识点
RxJava的组成要素包括Observable、Observer、Subscription、Operator和Scheduler。
Observable(被观察者)
Observable是一个可以发送消息的数据源,它可以同时发送若干消息。
Observable负责生成并发送数据给观察者,当所有消息发送完毕后,会通知观察者。
Observer(观察者)
Observer是一个接收Observable发送消息的对象。
它定义了处理Observable发射的消息的方法,包括处理正常数据(onNext)、处理错误(onError)和处理完成事件(onCompleted,但在RxJava 2.x中已改为onComplete,以符合Java命名习惯)。
Observer接收Observable发送的数据,并对这些数据执行相应的操作。同时,它也负责处理可能出现的错误和完成事件。
Subscription(订阅)
Subscription是Observer对Observable的绑定,它表示Observer正在接收Observable的数据项。
Subscription提供了取消订阅的功能。当订阅被取消后,Observer将不再接收Observable发送的消息。
Operator(操作符)
Operator是RxJava中用于处理和转换Observable发送的消息的中间处理器。
Operator提供了丰富的数据处理功能,如过滤、转换、合并、分组等。通过组合不同的Operator,可以实现复杂的异步数据处理逻辑。
Scheduler(调度器)
Scheduler是RxJava中用于控制事件序列在哪个线程上执行的工具。
Scheduler允许开发者指定Observable的发射线程和Observer的接收线程,从而更加灵活地控制程序的并发性和性能。RxJava提供了多种内置的Scheduler,如IO调度器(用于I/O操作)、计算调度器(用于CPU密集型计算)等。
四、使用场景:Rxjava搭配Retrofit使用
定义了一个接口,该接口包含了一个方法newFindOrder,该方法使用了@GET注解来指定HTTP请求的URL路径,并通过@QueryMap注解将一个Map作为查询参数传递给服务器。方法的返回类型是Observable,这意呀着Retrofit将执行HTTP GET请求,并将响应体自动转换为FindCoinOrderBean类型的对象,然后包装在一个Observable中返回。
@GET("/api/xxx")Observable<FindCoinOrderBean> newFindOrder(@QueryMap Map<String, Object> params);
一旦你有了服务的实例,你就可以调用newFindOrder方法来执行HTTP请求了。这个方法返回一个Observable,你可以使用RxJava(一个基于事件流的库)来处理这个Observable。
private void getRxjavaData() { String baseUrl = "https://xxx/v2/movie/"; Retrofit retrofit = new Retrofit.Builder() .baseUrl(baseUrl) .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .build(); CommonService movieService = retrofit.create(CommonService.class); movieService.newFindOrder(map).subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<FindCoinOrderBean>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { tv.setText(e.getMessage() + e.getLocalizedMessage()); Toast.makeText(MainActivity.this, "失败", Toast.LENGTH_SHORT).show(); } @Override public void onNext(FindCoinOrderBean findOrderBean) { Toast.makeText(MainActivity.this, "成功", Toast.LENGTH_SHORT).show(); } }); }
在上述的subscribe调用中,你创建了一个Observer的匿名子类实例,并将其传递给subscribe方法。当Observable(在这个例子中是Retrofit请求的结果)有数据要发送时,它会调用Observer的onNext方法,并传递数据作为参数。如果请求成功完成且没有更多的数据要发送,Observable会调用onComplete方法。如果发生错误,则会调用onError方法。