为了表述方便 本文使用缩写Rx代替ReactiveX
目录
第一行代码
安装依赖
cnpm i rxjs
cnpm i -g babel-cli
强烈要求使用nvm安装node环境 / 关于Babel的详细介绍参考Babel 入门教程
第一段代码
touch observable.js
var Rx = require('rxjs/Rx');
var observable = Rx.Observable.create(function (observer) {
observer.next(1);
setTimeout(() => {
observer.next(2);
observer.complete();
}, 1000);
});
babel-node observable.js
// 没有任何打印信息
第二段代码
vim observable.js
console.log('just before subscribe');
observable.subscribe({
next: x => console.log('got value ' + x),
error: err => console.error('something wrong occurred: ' + err),
complete: () => console.log('done'),
});
console.log('just after subscribe');
babel-node observable.js
just before subscribe
got value 1
just after subscribe
got value 2
done
互动问答: 从这段代码大家对Rx有哪些感受?
Rx是什么?
ReactiveX is a library for composing asynchronous and event-based programs by using observable sequences
多语言
RxJava / RxJS / Rx.NET / RxScala / RxClojure / RxSwift etc.
多平台
RxNetty / RxAndroid / RxCocoa
微软开源
Rx / VS Code / TypeScript etc.
更多微软开源参考Github Microsoft
三大对象(接口)
Observable
原型定义
create(onSubscription: function(observer: Observer): TeardownLogic) Observable
创建一个新的Observable 当观察者(Observer)订阅该(Observable)时 它会执行指定的函数
touch objects.js
var Rx = require('rxjs/Rx');
var observable = Rx.Observable.create(function subscribe(observer) {
var id = setInterval(() => {
observer.next('hi')
}, 1000);
});
babel-node objects.js
Observer
原型定义:
interface Observer<T> {
closed?: boolean;
next: (value: T) => void;
error: (err: any) => void;
complete: () => void;
}
vim objects.js
observable.subscribe({
next: x => console.log('got value ' + x),
error: err => console.error('something wrong occurred: ' + err),
complete: () => console.log('done'),
});
babel-node objects.js
Subscription
原型定义
subscribe(observerOrNext: Observer | Function, error: Function, complete: Function): ISubscription
调用Observable的执行并注册Observer的处理器以便于发出通知
vim objects.js
var subscription = observable.subscribe(x => console.log(x.toUpperCase()));
setInterval(() => {
subscription.unsubscribe();
}, 1000);
babel-node objects.js
三大特性
Flow
touch characters.html
<!DOCTYPE html>
<html lang="zh_CN">
<head>
<meta charset="UTF-8">
<title>rxjs-demo</title>
<script src="https://unpkg.com/rxjs@5.4.3/bundles/Rx.min.js"></script>
</head>
<body>
<button>点击</button>
<script>
(function () {
var count = 0;
var button = document.querySelector('button');
button.addEventListener('click', () => console.log(`Clicked ${++count} times`));
})();
</script>
</body>
</html>
vim characters.html
<body>
<button>点击</button>
<script>
(function () {
var button = document.querySelector('button');
Rx.Observable.fromEvent(button, 'click')
.scan(count => count + 1, 0)
.subscribe(count => console.log(`Clicked ${count} times`));
})();
</script>
</body>
rx-flow.gif
Operators
vim characters.html
<body>
<button>点击</button>
<script>
(function () {
var button = document.querySelector('button');
Rx.Observable.fromEvent(button, 'click')
.scan(count => count + 1, 0)
.subscribe(count => console.log(`Clicked ${count} times`));
})();
</script>
</body>
rx-scan.png
vim characters.html
<body>
<button>点击</button>
<script>
(function () {
var button = document.querySelector('button');
Rx.Observable.fromEvent(button, 'click')
.throttle(ev => Rx.Observable.interval(3000))
.scan(count => count + 1, 0)
.subscribe(count => console.log(`Clicked ${count} times`));
})();
</script>
</body>
rx-throttle.png
vim characters.html
<body>
<button>点击</button>
<script>
(function () {
var button = document.querySelector('button');
Rx.Observable.fromEvent(button, 'click')
.scan(count => count + 1, 0)
.map(x => x * 10)
.subscribe(x => console.log(x));
})();
</script>
</body>
rx-map.png
Scheduler
private void login(String phone, String password) {
Type type = new TypeToken<HttpResponse<User>>() {
}.getType();
HttpRequest apiRequest = new HttpRequest.Builder()
.setUrl(ApiEndPoint.ENDPOINT_LOGIN)
.setParam("phone", phone)
.setParam("password", password)
.setParam("auto_login", "1")
.build();
getDataManager().<HttpResponse<User>>post(type, apiRequest)
.subscribeOn(Schedulers.io())
.doOnSubscribe(new Consumer<Disposable>() {
@Override
public void accept(@NonNull Disposable disposable) throws Exception {
LoggerUtil.i("showLoading");
}
})
.map(new Function<HttpResponse<User>, User>() {
@Override
public User apply(@NonNull HttpResponse<User> response) throws Exception {
return response.getData();
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<User>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
addDisposable(disposable);
}
@Override
public void onNext(@NonNull User user) {
LoggerUtil.i(user.toString());
getDataManager().setAccessToken(user.getId().toString());
}
@Override
public void onError(@NonNull Throwable e) {
LoggerUtil.i("dismissLoading");
LoggerUtil.i("onError");
}
@Override
public void onComplete() {
LoggerUtil.i("dismissLoading");
getNavigator().goToMainActivity();
}
});
}
