试手 RxJava 2.x 及对线程的初步理解

1Observable.just("Hello World").subscribe(System.out::println);

1Observable.fromIterable(Arrays.asList("one", "two", "three"))
2    .subscribe(s -> System.out.println("Hello " + s));

 1Observable.<String>create(emitter -> {
 2    emitter.onNext("Monday");
 3    emitter.onNext("Tuesday");
 4    // emitter.onError(new RuntimeException("Something wrong")); #1
 5    emitter.onNext("Wednesday");
 6    emitter.onComplete();
 7}).subscribe(
 8    day -> System.out.println("Day: " + day),  //onNext
 9    t -> System.err.println(t.getMessage()),   //onError
10    () -> System.out.println("Done")           //onComplete
11);
12
13System.out.println("Program exit");

 1Observable.<String>create(emitter -> {
 2    log("Producer");
 3    emitter.onNext("Monday");
 4    emitter.onComplete();
 5}).subscribe(
 6    day -> {
 7        log("Consumer-" + day);
 8    }
 9);
10
11log("Program exit");

1private static void log(String msg) {
2    System.out.println(Thread.currentThread() + ": " + msg);
3}

 1Observable.<String>create(emitter -> {
 2    log("Producer");
 3    emitter.onNext("Monday");
 4    emitter.onComplete();
 5}).observeOn(Schedulers.newThread())  //只有消息消费用这个线程(池), 生产者仍然是主线程
 6    .subscribe(
 7        day -> {
 8            Thread.sleep(200);
 9            log("Consumer-" + day);
10        }
11    );
12
13log("Program exit");
14Thread.sleep(1000);

 1Observable.<String>create(emitter -> {
 2    log("Producer");
 3    emitter.onNext("Monday");
 4    emitter.onComplete();
 5}).subscribeOn(Schedulers.newThread())   //消息生产与消费都会有这个线程(池)
 6    .subscribe(
 7        day -> {
 8            Thread.sleep(200);
 9            log("Consumer-" + day);
10        }
11    );
12
13log("Program exit");
14Thread.sleep(1000);

 1Observable.<String>create(emitter -> {
 2    log("Producer");
 3    emitter.onNext("Monday");
 4    emitter.onComplete();
 5}).subscribeOn(Schedulers.newThread()) //消息生产者用这个线程(池)
 6    .observeOn(Schedulers.io())        //消息消费者用这个线程(池)
 7    .subscribe(
 8        day -> {
 9            Thread.sleep(200);
10            log("Consumer-" + day);
11        }
12    );
13
14log("Program exit");
15Thread.sleep(1000);

1Observable.just(request1, request2)
2    .observeOn(Schedulers.io())
3    .map(Request::get)
4    .subscribe(System.out::println);

1ExecutorService threadPool = Executors.newFixedThreadPool(10);
2Stream.of(request1, request2)
3    .map(request -> CompletableFuture.supplyAsync(request::get, threadPool))
4    .forEach(completableFuture ->
5    log(completableFuture.join())
6);

1PublishProcessor<String> restProcessor = PublishProcessor.create();
2Observable.fromPublisher(restProcessor).subscribe(System.out::println);
3 
4restProcessor.offer("Monday");
5restProcessor.offer("Tuesday");
6restProcessor.offer("Wednesday");
7restProcessor.onComplete();