Rx Java 异步编程框架

    /**
     * Rx 测试
     *
     * @param args
     */
    public static void main(String[] args) {
        // Demo1
        Flowable.just("Hello world!").subscribe(System.out::println);
        // Demo2 数据消费
        Observable.just("Hello World!").subscribe(System.out::println);
        // Demo3 数据中间处理
        Observable.just("Hello World!")
                .map(p -> p + " Master!")
                .subscribe(System.out::println);
        // Demo4 输出流数据转换
        Observable.just("Hello World!")
                .map(p -> p.hashCode())
                .subscribe(System.out::println);
        Observable.just("Hello World!")
                .map(p -> Base64Utils.encodeToString(p.getBytes()))
                .map(p -> new String(Base64Utils.decodeFromString(p)))
                .subscribe(System.out::println);
    }

Hello world!
Hello World!
Hello World! Master!
-969099747
Hello World!

source.operator1()
    .operator2()
    .operator3()
    .subscribe(consumer);
source.flatMap(value -> source.operator1().operator2().operator3());

Flowable<Integer> flow = Flowable.range(1, 5)
.map(v -> v * v)
.filter(v -> v % 3 == 0);

flow.subscribe(System.out::println)

Observable.create(emitter -> {
     while (!emitter.isDisposed()) {
         long time = System.currentTimeMillis();
         emitter.onNext(time);
         if (time % 2 != 0) {
             emitter.onError(new IllegalStateException("Odd millisecond!"));
             break;
         }
     }
})
.subscribe(System.out::println, Throwable::printStackTrace);

import io.reactivex.rxjava3.schedulers.Schedulers;

Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
})
  .subscribeOn(Schedulers.io())
  .observeOn(Schedulers.single())
  .subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000); // <--- wait for the flow to finish

Flowable<String> source = Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
});

Flowable<String> runBackground = source.subscribeOn(Schedulers.io());

Flowable<String> showForeground = runBackground.observeOn(Schedulers.single());

showForeground.subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000);

Observable.just("java", "C++", "python")
        //.observeOn(Schedulers.io())
    .observeOn(Schedulers.from(Executors.newFixedThreadPool(5)))
    .subscribe(System.out::println);

Flowable.range(1, 10)
  .observeOn(Schedulers.computation())
  .map(v -> v * v)
  .blockingSubscribe(System.out::println);

Flowable.range(1, 10)
  .flatMap(v ->
      Flowable.just(v)
        .subscribeOn(Schedulers.computation())
        .map(w -> w * w)
  )
  .blockingSubscribe(System.out::println);

Flowable.range(1, 10)
  .parallel()
  .runOn(Schedulers.computation())
  .map(v -> v * v)
  .sequential()
  .blockingSubscribe(System.out::println);

Flowable<Inventory> inventorySource = warehouse.getInventoryAsync();

inventorySource
    .flatMap(inventoryItem -> erp.getDemandAsync(inventoryItem.getId())
            .map(demand -> "Item " + inventoryItem.getName() + " has demand " + demand))
    .subscribe(System.out::println);

service.apiCall()
.flatMap(value -> service.anotherApiCall(value))
.flatMap(next -> service.finalCall(next))

service.apiCall()
.flatMap(value ->
    service.anotherApiCall(value)
    .flatMap(next -> service.finalCallBoth(value, next))
)

Observable continued = sourceObservable.flatMapSingle(ignored -> someSingleSource)
continued.map(v -> v.toString())
  .subscribe(System.out::println, Throwable::printStackTrace);

sourceObservable
  .ignoreElements()           // returns Completable
  .andThen(someSingleSource)
  .map(v -> v.toString())

AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.just(count.get()))
  .subscribe(System.out::println);

AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.defer(() -> Single.just(count.get())))
  .subscribe(System.out::println);

// 或者

AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.fromCallable(() -> count.get()))
  .subscribe(System.out::println);

Flowable<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper)

Flowable<R> flatMapMaybe(Function<? super T, ? extends MaybeSource<? extends R>> mapper)

Flowable<T> concatWith(Publisher<? extends T> other);

Flowable<T> concatWith(SingleSource<? extends T> other);

someSource.concatWith(s -> Single.just(2))
.subscribe(System.out::println, Throwable::printStackTrace);

someSource.concatWith(Single.defer(() -> Single.just(2)))
.subscribe(System.out::println, Throwable::printStackTrace);

Flowable<T> merge(Publisher<? extends Publisher<? extends T>> sources);

Flowable<T> mergeArray(Publisher<? extends T>... sources);

Flowable<T> concat(Publisher<? extends Publisher<? extends T>> sources);

Flowable<T> concatDelayError(Publisher<? extends Publisher<? extends T>> sources);

Flowable<T> concatArrayEagerDelayError(Publisher<? extends T>... sources);

Warning: org.reactivestreams.FlowAdapters$FlowPublisherFromReactive: can't find superclass or interface java.util.concurrent.Flow$Publisher
Warning: org.reactivestreams.FlowAdapters$FlowToReactiveProcessor: can't find superclass or interface java.util.concurrent.Flow$Processor
Warning: org.reactivestreams.FlowAdapters$FlowToReactiveSubscriber: can't find superclass or interface java.util.concurrent.Flow$Subscriber
Warning: org.reactivestreams.FlowAdapters$FlowToReactiveSubscription: can't find superclass or interface java.util.concurrent.Flow$Subscription
Warning: org.reactivestreams.FlowAdapters: can't find referenced class java.util.concurrent.Flow$Publisher

<dependency>
    <groupId>io.reactivex.rxjava3</groupId>
    <artifactId>rxjava</artifactId>
    <version>x.y.z</version>
</dependency>

implementation 'io.reactivex.rxjava3:rxjava:x.y.z'

public abstract class Flowable<@NonNull T> implements Publisher<T>

  /**
     * Flowable 基类测试
     * @param args
     * @throws InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
        Disposable d = Flowable.just("Hello world!")
                .delay(1, TimeUnit.SECONDS)
                .subscribeWith(new DisposableSubscriber<String>() {
                    @Override
                    public void onStart() {
                        System.out.println("Start!");
                        request(1);
                    }
                    @Override
                    public void onNext(String t) {
                        System.out.println(t);
                        request(1);
                    }
                    @Override
                    public void onError(Throwable t) {
                        t.printStackTrace();
                    }
                    @Override
                    public void onComplete() {
                        System.out.println("Done!");
                    }
                });
        Thread.sleep(5000);
        // the sequence can now be cancelled via dispose()
        d.dispose();
    }

> Task :rx-java-examples:rx-java-chapter-1:FlowableTest.main()
Start!
Hello world!
Done!

    /**
     * 工厂方法创建 Flowable
     * @param args
     */
    public static void main(String[] args) {
        Flowable<String> source = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {

                // signal an item
                emitter.onNext("Hello");

                // could be some blocking operation
                Thread.sleep(1000);

                // the consumer might have cancelled the flow
                if (emitter.isCancelled()) {
                    return;
                }

                emitter.onNext("World");

                Thread.sleep(1000);

                // the end-of-sequence has to be signaled, otherwise the
                // consumers may never finish
                emitter.onComplete();
            }        
        }, BackpressureStrategy.BUFFER); // 指定反压策略

        System.out.println("Subscribe!");

        source.subscribe(System.out::println);

        System.out.println("Done!");
    }

> Task :rx-java-examples:rx-java-chapter-1:FlowableTest2.main()
Subscribe!
Hello
World
Done!

Flowable.range(0, 10)
                .subscribe(new Subscriber<Integer>() {

                    Subscription sub;
     // 当订阅后，会首先调用这个方法，其实就相当于onStart()，
              // 传入的 Subscription s 参数可以用于请求数据或者取消订阅
                    @Override
                    public void onSubscribe(Subscription s) {
                        System.out.println("onSubscribe START");
                        sub = s;
                        s.request(1);
                        try {
                            TimeUnit.SECONDS.sleep(3);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("onSubscribe END");
                    }

                    @Override
                    public void onNext(Integer v) {
                        System.out.println("onNext: " + v);
                        sub.request(1);
                    }

                    @Override
                    public void onError(Throwable t) {
                        t.printStackTrace();
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("Done!");
                    }
                });

        Thread.sleep(5000);
        // the sequence can now be cancelled via dispose()
    }
// 输出
onSubscribe START
onSubscribe END
onNext: 0
onNext: 1
...
onNext: 8
onNext: 9
Done!

 public static void main(String[] args) {
        System.out.println("DEMO 1 ---------------");
        Observable.just(1, 2, 3)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        d.dispose();
                        System.out.println("Cancel !");
                    }
                    @Override
                    public void onNext(Integer item) {
                        System.out.println("Next: " + item);
                    }
                    @Override
                    public void onError(Throwable error) {
                        System.err.println("Error: " + error.getMessage());
                    }
                    @Override
                    public void onComplete() {
                        System.out.println("Sequence complete.");
                    }
                });
        System.out.println("DEMO 2 ---------------");
        Observable.just(1, 2, 3)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        System.out.println("onSubscribe ! + " + d.isDisposed());
                    }
                    @Override
                    public void onNext(Integer item) {
                        System.out.println("Next: " + item);
                    }
                    @Override
                    public void onError(Throwable error) {
                        System.err.println("Error: " + error.getMessage());
                    }
                    @Override
                    public void onComplete() {
                        System.out.println("Sequence complete.");
                    }
                });
    }
// 输出
DEMO 1 ---------------
Cancel !
DEMO 2 ---------------
onSubscribe ! + false
Next: 1
Next: 2
Next: 3
Sequence complete

  // 数组
  System.out.println("DEMO 3 ---------------");
        Integer[] items = {0, 1, 2, 3, 4, 5};
        Observable myObservable = Observable.fromArray(items);
        myObservable.subscribe(next -> System.out.println(next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
  // 异常处理
        System.out.println("DEMO 4 ---------------");
        Integer[] items2 = {0, 1, null, 3, 4, 5};
        myObservable = Observable.fromArray(items2);
        myObservable.subscribe(next -> System.out.println(next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
  // 迭代器
  System.out.println("DEMO 5 ---------------");
        Integer[] items3 = {0, 1, 2, 3, 4, 5};
        myObservable = Observable.fromIterable(Arrays.asList(items3));
        myObservable.subscribe(next -> System.out.println(next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
  // Future
  System.out.println("DEMO 6 ---------------");
        FutureTask<Long> task = new FutureTask<>(() -> System.currentTimeMillis());
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.submit(task);
        try {
            System.out.println("get:" + task.get());
        } catch (Exception e) {
            e.printStackTrace();
        }
        //executorService.shutdown(); 移至最后执行
        myObservable = Observable.fromFuture(task);
        myObservable.subscribe(next -> System.out.println("RECIEVE: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
  // Future 超时
  System.out.println("DEMO 7 ---------------");
        FutureTask<Long> task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return System.currentTimeMillis();
        });
        myObservable = Observable.fromFuture(task2, 1, TimeUnit.SECONDS);//等待1s
        executorService.submit(task2);
        myObservable.subscribe(next -> System.out.println("RECIEVE WITHIN 1s: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 8 ---------------");
        task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return System.currentTimeMillis();
        });
        myObservable = Observable.fromFuture(task2, 4, TimeUnit.SECONDS);//等待4s
        executorService.submit(task2);
        myObservable.subscribe(next -> System.out.println("RECIEVE WITHIN 4s: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 9 ---------------");
        task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return System.currentTimeMillis();
        });
        myObservable = Observable.fromFuture(task2, 1, TimeUnit.SECONDS);//等待1s
        executorService.submit(task2);
        myObservable.subscribe(next -> System.out.println("RECIEVE WITHIN 1s: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        System.out.println("DEMO 10 ---------------");
        task2 = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(3);
            return System.currentTimeMillis();
        });
        myObservable = Observable.fromFuture(task2, 4, TimeUnit.SECONDS);//等待4s
        executorService.submit(task2);
        myObservable.subscribe(next -> System.out.println("RECIEVE WITHIN 4s: " + next),
                error -> System.out.println(error),
                () -> System.out.println("DONE!")
        );
        executorService.shutdown();
// 输出
DEMO 3 ---------------
0
1
2
3
4
5
DONE!
DEMO 4 ---------------
0
1
java.lang.NullPointerException: The element at index 2 is null
DEMO 5 ---------------
0
1
2
3
4
5
DONE!
DEMO 6 ---------------
get:1608780273323
RECIEVE: 1608780273323
DONE!
DEMO 7 ---------------
java.util.concurrent.TimeoutException
DEMO 8 ---------------
RECIEVE WITHIN 4s: 1608781164556
DONE!
DEMO 9 ---------------
java.util.concurrent.TimeoutException
DEMO 10 ---------------
RECIEVE WITHIN 4s: 1608781168578
DONE!

 System.out.println("DEMO 1 ---------------");
        Integer[] items3 = {0, 1, 2};
        Observable.fromIterable(Arrays.asList(items3))
                //.repeat() // 一直重复发送
                .repeat(2) // 重复2次
                .subscribe(next -> System.out.println(next),
                        error -> System.out.println(error),
                        () -> System.out.println("DONE!")
                );
        System.out.println("DEMO 2 ---------------");
        Observable<Integer> observable = Observable.create(sub -> {
            System.out.println("START DEMO 2");
            sub.onNext(0);
            sub.onNext(1);
            sub.onComplete();
        });
        observable
                //.repeat() // 一直重复发送
                .map(p -> p * 2)
                .repeat(2) // 重复2次
                .map(p -> "modified: " + p)
                .repeat(2) // 重复2次
                .subscribe(next -> System.out.println(next),
                        error -> System.out.println(error),
                        () -> System.out.println("DONE!")
                );

// 输出
DEMO 1 ---------------
0
1
2
0
1
2
DONE!
DEMO 2 ---------------
START DEMO 2
modified: 0
modified: 2
START DEMO 2
modified: 0
modified: 2
START DEMO 2
modified: 0
modified: 2
START DEMO 2
modified: 0
modified: 2
DONE!

 .repeat(2) // 重复2次
 .repeat(2) // 重复2次
// 相当于：
 .repeat(2*2=4) // 重复4次