当前位置：首页 > news >正文

购物网站的设计思路/网店培训班

news 2025/8/4 0:36:44

购物网站的设计思路,网店培训班,建筑网课平台,wordpress文章页头添加广告代码介绍 RxJava，响应式编程，告别AsyncTask，与Retrofit组成了日常开发经典组合。之前，好读书不求甚解得态度，让我对其印象并不深刻，知识层面比较浅薄。为了加深印象，重新学习。添加依赖 impleme…

介绍

RxJava，响应式编程，告别AsyncTask，与Retrofit组成了日常开发经典组合。之前，好读书不求甚解得态度，让我对其印象并不深刻，知识层面比较浅薄。为了加深印象，重新学习。

添加依赖

implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'
implementation 'io.reactivex.rxjava2:rxjava:2.1.7'

举个栗子

1. 上游和下游（被观察者和观察者）

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(ObservableEmitter<Integer> e) throws Exception {e.onNext(111);e.onNext(222);e.onComplete();}
});Observer<Integer> observer = new Observer<Integer>() {@Overridepublic void onSubscribe(Disposable d) {Logger("onSubscribe");}@Overridepublic void onNext(Integer integer) {Logger("onNext integer = " + integer);}@Overridepublic void onError(Throwable e) {Logger("onError e = " + e.getMessage());}@Overridepublic void onComplete() {Logger("onComplete");}
};
observable.subscribe(observer);

2. 运行结果

这里写图片描述

3. 流程

创建Observable
创建Observer
Observerable订阅（subscribe）Observer

4. Observable

Observable是一个抽象类，实现ObservableSource接口

public abstract class Observable<T> implements ObservableSource<T>

而ObservableSource接口中只有一个订阅方法

public interface ObservableSource<T> {/*** Subscribes the given Observer to this ObservableSource instance.* @param observer the Observer, not null* @throws NullPointerException if {@code observer} is null*/void subscribe(@NonNull Observer<? super T> observer);
}

通过create方法创建Observable，稍微追下源码，帮助理解

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {ObjectHelper.requireNonNull(source, "source is null");return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));}

参数类型为ObservableOnSubscribe接口

public interface ObservableOnSubscribe<T> {/*** Called for each Observer that subscribes.* @param e the safe emitter instance, never null* @throws Exception on error*/void subscribe(@NonNull ObservableEmitter<T> e) throws Exception;
}

ObservableEmitter可以理解成发射器，看下它的源码。它继承Emitter。而Emitter中才是我们最基础最常用的功能。

public interface ObservableEmitter<T> extends Emitter<T> {void setDisposable(@Nullable Disposable d);void setCancellable(@Nullable Cancellable c);boolean isDisposed();ObservableEmitter<T> serialize();@Experimentalboolean tryOnError(@NonNull Throwable t);
}

上面这部分代码，删除了部分注释，可在源码中查阅。

public interface Emitter<T> {/*** Signal a normal value.* @param value the value to signal, not null*/void onNext(@NonNull T value);/*** Signal a Throwable exception.* @param error the Throwable to signal, not null*/void onError(@NonNull Throwable error);/*** Signal a completion.*/void onComplete();
}

从上面Emitter的代码可以看出，我们可以发射三种消息：
onNext, onComplete, onError。从语义上就很容易看出他们的作用。
针对三个事件，做一些说明：
1. 发射器可以发送无限个onNext, Observer也可以接收无限个onNext。
2. 发射器发送一个onComplete后, onComplete之后的事件将会继续发送, 而Observer收到第一个onComplete事件之后将不再继续接收事件。
3. 发射器发送了一个onError后, onError之后的事件将继续发送, 而Observer收到onError事件之后将不再继续接收事件.
4. 发射器可以不发送onComplete或onError。
5. 发射器不能同时发出onComplete和onError。
6. 不允许发出多个onError或者onComplete

5. Observer

相对于Observable而言，Observer就简单很多，我们只需要针对我们感兴趣的事件作出相应的处理即可。

public interface Observer<T> {void onSubscribe(@NonNull Disposable d);void onNext(@NonNull T t);void onError(@NonNull Throwable e);void onComplete();
}

这里有一个新内容Disposable，先看代码，“象征一个一次性资源”，从注释上看是这个意思。但是如何理解？

/*** Represents a disposable resource.*/
public interface Disposable {/*** Dispose the resource, the operation should be idempotent.*/void dispose();/*** Returns true if this resource has been disposed.* @return true if this resource has been disposed*/boolean isDisposed();
}

看个栗子：

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(ObservableEmitter<Integer> e) throws Exception {e.onNext(111);e.onNext(222);e.onNext(333);e.onComplete();e.onNext(444);}
});Observer<Integer> observer = new Observer<Integer>() {Disposable mDisposable;@Overridepublic void onSubscribe(Disposable d) {Logger("onSubscribe");mDisposable = d;}@Overridepublic void onNext(Integer integer) {Logger("onNext integer = " + integer);if(integer== 222 && !mDisposable.isDisposed()) {mDisposable.dispose();}}@Overridepublic void onError(Throwable e) {Logger("onError e = " + e.getMessage());}@Overridepublic void onComplete() {Logger("onComplete");}
};
observable.subscribe(observer);

运行结果：
这里写图片描述
可以看到发射器发送了222之后，剩下的事件Observer并没有处理，那么剩下的事件发射器有没有发送呢？在发射器发射事件之前添加log。

可以看到发射器的事件并没有受到Disposable影响，只是Observer没有接收到222之后的事件而已。
Disposable的作用应该是，截断发射器与Observer之间的事件通道。