Learning Reactive-Extensions (RxJava & RxJS)

JimmyLv commented 8 years ago

项目中的经验非常有用！从问题描述开始，聚合性 portfolio 所 deps 的各种 APIs：

Sam Newman - Backends For Frontends 文中的 And Multiple Downstream Services (Microservices!) 章节：

Multiple services hold the pieces of information we want. The Wishlist service stores information about the list, and IDs of each item. The Catalog service stores the name and price of each item, and the Stock levels are stored in our inventory service. So in our BFF we'd expose a method for retrieving the full playlist, which would consist of at least 3 calls:

Making multiple downstream calls to construct a view of a wishlist

From an efficiency point of view, it would be much smarter to run as many calls in parallel as possible. Once the initial call to the Wishlist service completes, ideally we'd like to then run the calls to the other services at the same time to reduce the overall call time. This need to mix calls that we want to run in parallel vs those that run in sequence can quickly become painful to manage, especially for more complex scenarios.

This is one area where a reactive style of programming can help (such as that provided by RxJava or Finagle's futures system) as the composition of multiple calls becomes easier to manage.

References

http://reactivex.io/documentation/operators.html official documents https://www.javadoc.io/doc/io.reactivex/rxjava/1.1.1 rxjava 1.1.1 api docs http://reactivex.io/RxJava/javadoc/ rxjava 2 api docs http://gank.io/post/560e15be2dca930e00da1083 useful tutorial for beginners

JimmyLv commented 8 years ago

对比 MobX #75，放弃的原因：Mobx vs Reactive Stream Libraries (RxJS, Bacon, etc) · mobxjs/mobx Wiki

JimmyLv commented 7 years ago

RxJS 并不只是 JavaScript，而是整个 Reactive-Extensions 体系，还包括现在项目所用的 RxJava，其思想和操作符更为重要和值得学习，而反而是 MobX 只局限于 React 世界里，并且还不一定拼得过 Redux，以及未来的 Relay 和 GraphQL…

JimmyLv commented 7 years ago

@Yanghai 组内分享 RxJava 的 Session 很不错，AGENDA 如下：

Introduction
Core Conceptions
Event Transformation
More Operations

Introduction

RxJava is a Java implementation of ReactiveX: a library for composing asynchronous and event-based programs by using observable sequences.

Observable.from(folders)
    .flatMap((Func1) (folder) -> { Observable.from(file.listFiles()) })
    .filter((Func1) (file) -> { file.getName().endsWith(".png") })
    .map((Func1) (file) -> { getBitmapFromFile(file) })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe((Action1) (bitmap)->  { imageCollectorView.addImage(bitmap) });

Core Conceptions

RxJava contains the following core concepts:

Observable: event source Observer/Subscriber: event consumer Scheduler: thread controller Subscribe: the operation to trigger event sequence

Obervable.onSubscrible() triggers events

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();
    }
});

Observable observable = Observable.just("Hello", "Hi", “Aloha");

It’s equivalent to:

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);

Subscriber

一般不用关心，Subscriber 属于 RxJava 实现上的内部概念。

Subscriber<String> subscriber = new Subscriber<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!");
    }
};

Observer

Observer will be converted into Subscriber.

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!");
    }
};

Subscribe

String[] words = {"Hello", "Hi", “Aloha”};
Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, “Word: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

Observable.from(words).subscribe(subscriber);