Rxjava2 操作符匯總

構造

create/generate 帚桩？
just/empty/error/never/range(Long)
from(Array/Callable/Future/Iterable/Publisher)
startWith/repeat/cache/defaultIfEmpty 增加默認值開頭/重復/緩存/返回默認值

加時

defer/timer/interval(Range) 延遲執(zhí)行
delay/timeout/timestamp 延時、超時、時間戳

異常

onError(ResumeNext/Return/ReturnItem) 錯誤處理
onExceptionResumeNext 放行error望薄、throwable
retry(Until/When)

并發(fā)與并行

parallel槽驶、serialize 并行
observeOn、subscribeOn 指定線程

獲取結果

blocking[First(T)/Last(T)/ForEach/Iterable(int)/Latest/MostRecent(T)/Single/Subscribe] 獲取結果
blockingNext ?
forEach(While) 直接消費結果
subscribe 直接消費結果

限流

distinct(UntilChanged)去重
filter 按條件過濾
ignoreElements 忽略所有
limit 限制個數(shù)
(take/skip)(Last/Until/While)
elementAt(OrError)
(first/last/single)(Element/OrError)
debounce 過濾蚯瞧，防止抖動嘿期，如果指定時間間接內沒有其他數(shù)據(jù)，則正常埋合。
sample 抽樣
throttle(First/Last/WithtimeOut) ?
window 窗口
onBackpressureBuffer/Drop/Latest 背壓

合并源

amb(Array) with 多個流取最快的一個
concat(Array)(Eager/DelayError) with 多個流順序合并
switchOnNext(DelayError) 不斷切換到最近開始的流
merge(Array)(DelayError) with 多個流程亂序合并
combineLastest(DelayError) 多個流流速不同時备徐，流速快的與流速慢的最新一個合并。 A1 B1 C2….
withLatestFrom
zip(Array/Iterable) with 多個流一一對應饥悴，合并處理坦喘，最先結束為截止點 A1 B2 C3 …
join ?

轉換

buffer(count,skip)、（timespan,timeskip, unit）分批
map/cast 轉換
flatMap/concatMap(Eager)(DelayError) 流轉換, 亂序西设、順序
switchMap（DelayError）瓣铣？
all/any/contains 條件判斷
isEmpty/sequenceEqual 兩個流一一比較是否相同，可以先后順序不同贷揽？
collect/reduce/scan 歸并
groupBy 分組
sorted 排序
toList/Map/Multimap/SortedList 集合

日志

doOnSubscribe/Request/Next/Complete/Error/Cancel
doOnTerminate = complete/error
doFinally = terminate/cancel
doOnLifecycle = subscribe/request/cancel
doOnEach = all
doAfterNext/AfterTerminate

新增操作

compose 組合多個操作
lift 創(chuàng)建新的操作

實例

maven依賴如下：

    <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-logging</artifactId>
        </dependency>
        <dependency>
            <groupId>io.reactivex.rxjava2</groupId>
            <artifactId>rxjava</artifactId>
        </dependency>
    </dependencies>

代碼示例如下：

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.util.StopWatch;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.IntStream;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;

import static java.util.stream.Collectors.joining;

@Slf4j
public class RxjavaTest {
  
  
  static void sleep(long delayed, TimeUnit unit){
    try {
      unit.sleep(delayed);
    } catch (InterruptedException e) {
    }
  }
  
  String findById(Long id){
    log.info("called findById {}", id);
    sleep(200, TimeUnit.MILLISECONDS);
    return id + "-name";
  }
  
  String findById(String id){
    log.info("called findById {}", id);
    sleep(200, TimeUnit.MILLISECONDS);
    return id + "-name";
  }
  
  String findByIdWithError(Long id){
    String result = findById(id);
    if(id%3==0){
      throw new RuntimeException("id.not.exist");
    }
    return result;
  }
  
  static void trace(Runnable runnable){
    StopWatch stopWatch = new StopWatch();
    stopWatch.start();
    try {
      runnable.run();
    }finally {
      stopWatch.stop();
      System.out.printf("trace = %s", stopWatch.prettyPrint());
    }
  }
  
  /**
   * 各種簡單創(chuàng)建接口
   * @throws Exception
   */
  @Test
  public void test_create() throws Exception{
    //列舉
    Observable.just(1,2).blockingSubscribe(System.out::println);
    //空
    Observable.empty().blockingSubscribe(System.out::println);
    //錯誤
    Observable.error(new RuntimeException("error")).onErrorReturnItem("default").blockingSubscribe(System.out::println);
    //范圍
    Observable.rangeLong(1, 10).blockingSubscribe(System.out::println);
    //列表
    Observable.fromIterable(Arrays.asList(1,2)).blockingSubscribe(System.out::println);
    //數(shù)組
    Observable.fromArray(new Integer[]{1,2}).blockingSubscribe(System.out::println);
    //重復
    Observable.just(1,2).repeat(3).blockingSubscribe(System.out::println);
  }
  
  
  /**
   * 將分頁接口包裝成流
   *
   * @throws Exception
   */
  @Test
  public void test_page_observable() throws Exception{
    Observable<String> observable = Observable.<String>create(emitter -> {
      try {
        int pageNo = 1;
  
        while (pageNo < 5) {
          int current = pageNo;
          
          log.info("start read pageNo={}", pageNo);
          
          IntStream.range(0, 10).mapToObj(i -> "sku" + String.valueOf((current-1)*10000+i)).forEach(emitter::onNext);
          
          sleep(10, TimeUnit.MILLISECONDS);
          
          log.info("end read pageNo={}", pageNo);
          
          ++pageNo;
        }
        emitter.onComplete();
      } catch (Exception e) {
        log.error("{}", e);
      } finally {
        emitter.onComplete();
      }
    });
    
    observable
        .blockingSubscribe(System.out::println);
    
    
    TimeUnit.SECONDS.sleep(1);
    log.info("將讀取放在單獨的線程中執(zhí)行:");
    observable
        .subscribeOn(Schedulers.io())
        .blockingSubscribe(System.out::println);
  
    
    TimeUnit.SECONDS.sleep(1);
    log.info("將讀取放在單獨的線程中, 并且分批執(zhí)行:");
    observable
        .subscribeOn(Schedulers.io())
        .buffer(10) //分批處理
        .blockingSubscribe(System.out::println);
  }
  
  /**
   08:50:15.564 [main] INFO top.zhacker.rxjava.RxjavaTest - called findById 1
   08:50:15.854 [main] INFO top.zhacker.rxjava.RxjavaTest - subscribe observable
   08:50:15.914 [main] INFO top.zhacker.rxjava.RxjavaTest - subscribe observable
   08:50:15.914 [main] INFO top.zhacker.rxjava.RxjavaTest - called findById 1
   08:50:16.115 [main] INFO top.zhacker.rxjava.RxjavaTest - subscribe observable
   08:50:16.116 [main] INFO top.zhacker.rxjava.RxjavaTest - called findById 1
   */
  @Test
  public void test_defer(){
    Observable<String> observableNotDeferred = Observable.just(findById(1L));
    log.info("subscribe observable");
    observableNotDeferred.blockingSubscribe();
  
    Observable<String> observableDeferred = Observable.defer(()-> Observable.just(findById(1L)));
    log.info("subscribe observable");
    observableDeferred.blockingSubscribe();
  
    Observable<String> observable = Observable.fromCallable(()-> findById(1L));
    log.info("subscribe observable");
    observable.blockingSubscribe();
  }
  
  
  /**
   * Timed[time=1530762785174, unit=MILLISECONDS, value=1]
   * Timed[time=1530762785175, unit=MILLISECONDS, value=2]
   * Timed[time=1530762785175, unit=MILLISECONDS, value=3]
   * Timed[time=1530762787179, unit=MILLISECONDS, value=5]
   * Timed[time=1530762788180, unit=MILLISECONDS, value=6]
   * Timed[time=1530762789179, unit=MILLISECONDS, value=7]
   * Timed[time=1530762790182, unit=MILLISECONDS, value=8]
   * Timed[time=1530762791180, unit=MILLISECONDS, value=9]
   */
  @Test
  public void test_delay(){
    Observable
        .just(1,2,3)
        .delay(2,TimeUnit.SECONDS) //延遲2s
        .timestamp() //加時間戳 Timed<T>
        .blockingSubscribe(System.out::println);
    Observable
        .intervalRange(5, 5, 2, 1, TimeUnit.SECONDS) //延遲2s棠笑，之后每個延遲1s
        .timestamp()
        .blockingSubscribe(System.out::println);
  }
  
  
  /**
   *
   * @throws Exception
   */
  @Test
  public void test_error_retry() throws Exception{
    
    Observable.fromCallable(()-> findByIdWithError(3L))
        .doOnError(e-> log.warn("", e))
        .retry(2, e-> e.getMessage().equals("id.not.exist"))
        .blockingSubscribe();
  }
  
  /**
   *
   * @throws Exception
   */
  @Test
  public void test_error_returnItem() throws Exception{
    
    Observable.fromCallable(()-> findByIdWithError(3L))
        .doOnError(e-> log.warn("", e))
        .onErrorReturnItem("default result")
        .blockingSubscribe(System.out::println);
  }
  
  /**
   *
   * @throws Exception
   */
  @Test
  public void test_error_resumeNext() throws Exception{
    
    Observable.fromCallable(()-> findByIdWithError(3L))
        .doOnError(e-> log.warn("", e))
        .onErrorResumeNext(Observable.just("default result"))
        .blockingSubscribe(System.out::println);
  }
  
  
  
  /**
   *
   * @throws Exception
   */
  @Test
  public void test_timeout_retry() throws Exception{
    Observable.fromCallable(()-> findById(1L))
        .timeout(100, TimeUnit.MILLISECONDS)
        .doOnError(e-> log.warn("", e))
        .retry(2, e-> e instanceof TimeoutException)
        .blockingSubscribe();
  }
  
  
  
  
  
  /**
   * start handle id= 1
   * start handle id= 2
   * start handle id= 3
   * start handle id= 4
   * start handle id= 5
   * result for :5
   * result for :1
   * result for :2
   * result for :3
   * result for :4
   * trace = StopWatch '': running time (millis) = 591
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00591  100%
   * @throws Exception
   */
  @Test
  public void test_flatMap_without_Errors() throws Exception{
  
    trace(()->{
      Observable.rangeLong(1,5)
          .doOnNext(id-> System.out.printf("start handle id= %s\n", id))
          .flatMap(id-> Observable.fromCallable(()->{
            return findById(id);
          }).subscribeOn(Schedulers.io()), 5)
          .doOnNext(result-> System.out.printf("%s\n", result))
          .blockingSubscribe();
    });
  }
  
  
  /**
   *
   * start handle id= 1
   * start handle id= 2
   * start handle id= 3
   * start handle id= 4
   * start handle id= 5
   * result for :2
   * result for :4
   * result for :1
   * result for :5
   * trace = StopWatch '': running time (millis) = 457
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00457  100%
   *
   * java.lang.RuntimeException: id.not.exist
   *
   * @throws Exception
   */
  @Test
  public void test_flatMap_throw_Errors_direct() throws Exception{
    
    trace(()->{
      Observable.rangeLong(1,5)
          .doOnNext(id-> System.out.printf("start handle id= %s\n", id))
          .flatMap(id-> Observable.fromCallable(()->{
            return findByIdWithError(id);
          }).subscribeOn(Schedulers.io()), 5)
          .doOnNext(result-> System.out.printf("%s\n", result))
          .blockingSubscribe();
    });
  }
  
  
  /**
   *start handle id= 1
   * start handle id= 2
   * start handle id= 3
   * start handle id= 4
   * start handle id= 5
   * result for :2
   * result for :1
   * result for :4
   * result for :5
   * trace = StopWatch '': running time (millis) = 450
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00450  100%
   *
   * java.lang.RuntimeException: id.not.exist
   */
  @Test
  public void test_flatMap_delayErrors(){
    trace(()->{
      Observable.rangeLong(1,5)
          .doOnNext(id-> System.out.printf("start handle id= %s\n", id))
          .flatMap(id-> Observable.fromCallable(()->{
            return findByIdWithError(id);
          }).subscribeOn(Schedulers.io()), true, 5)
          .doOnNext(result-> System.out.printf("%s\n", result))
          .blockingSubscribe();
    });
  }
  
  
  /**
   * start handle id= 1
   * start handle id= 2
   * start handle id= 3
   * start handle id= 4
   * start handle id= 5
   * result for :1
   * result for :2
   * trace = StopWatch '': running time (millis) = 350
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00350  100%
   *
   * java.lang.RuntimeException: id.not.exist
   * @throws Exception
   */
  @Test
  public void test_concatMap_direct_throw_Errors() throws Exception{
    
    trace(()->{
      Observable.rangeLong(1,5)
          .doOnNext(id-> System.out.printf("start handle id= %s\n", id))
          .concatMapEager(id-> Observable.<String>fromCallable(()->{
            return findByIdWithError(id);
          }).subscribeOn(Schedulers.io()))
          .doOnNext(result-> System.out.printf("%s\n", result))
          .blockingSubscribe();
    });
  }
  
  
  /**
   * start handle id= 1
   * start handle id= 2
   * start handle id= 3
   * start handle id= 4
   * start handle id= 5
   * result for :1
   * result for :2
   * result for :4
   * result for :5
   * trace = StopWatch '': running time (millis) = 410
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00410  100%
   *
   * java.lang.RuntimeException: id.not.exist
   * @throws Exception
   */
  @Test
  public void test_concatMap_delayErrors() throws Exception{
    
    trace(()->{
      Observable.rangeLong(1,5)
          .doOnNext(id-> System.out.printf("start handle id= %s\n", id))
          .concatMapEagerDelayError(id-> Observable.fromCallable(()->{
            return findByIdWithError(id);
          }).subscribeOn(Schedulers.io()), true)
          .doOnNext(result-> System.out.printf("%s\n", result))
          .blockingSubscribe();
    });
  }
  
  
  /**
   21
   22
   23
   trace = StopWatch '': running time (millis) = 120
   -----------------------------------------
   ms     %     Task name
   -----------------------------------------
   00120  100%
   * @throws Exception
   */
  @Test
  public void test_amb() throws Exception{
  
    trace(()->{
      Observable.ambArray(
          Observable.intervalRange(1, 3, 50,100,TimeUnit.MILLISECONDS),
          Observable.intervalRange(11, 3, 40,100,TimeUnit.MILLISECONDS),
          Observable.range(21, 3).delay(30,TimeUnit.MILLISECONDS)
      ).blockingSubscribe(System.out::println);
    });
    
  }
  
  
  /**
   * 21
   * 22
   * 23
   * 11
   * 1
   * 12
   * 2
   * 13
   * 3
   * trace = StopWatch '': running time (millis) = 389
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00389  100%
   * @throws Exception
   */
  @Test
  public void test_merge() throws Exception{
    
    trace(()->{
      Observable.merge(
          Observable.intervalRange(1, 3, 50,100,TimeUnit.MILLISECONDS),
          Observable.intervalRange(11, 3, 40,100,TimeUnit.MILLISECONDS),
          Observable.range(21, 3).delay(30,TimeUnit.MILLISECONDS)
      ).blockingSubscribe(System.out::println);
    });
  }
  
  
  /**
   * 1
   * 2
   * 3
   * 11
   * 12
   * 13
   * 21
   * 22
   * 23
   * trace = StopWatch '': running time (millis) = 670
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00670  100%
   * @throws Exception
   */
  @Test
  public void test_concat() throws Exception{
    trace(()->{
      Observable.concat(
          Observable.intervalRange(1, 3, 50,100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()),
          Observable.intervalRange(11, 3, 40,100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()),
          Observable.range(21, 3).delay(30,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io())
      ).blockingSubscribe(System.out::println);
    });
  }
  
  
  /**
   * 1
   * 2
   * 3
   * 11
   * 12
   * 13
   * 21
   * 22
   * 23
   * trace = StopWatch '': running time (millis) = 407
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00407  100%
   * @throws Exception
   */
  @Test
  public void test_concatEager() throws Exception{
    trace(()->{
      Observable.concatEager(Arrays.asList(
          Observable.intervalRange(1, 3, 50,100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()),
          Observable.intervalRange(11, 3, 40,100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()),
          Observable.range(21, 3).delay(300,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io())),
          3, 3
      ).blockingSubscribe(System.out::println);
    });
  }
  
  
  /**
   * 1,11,21
   * 2,12,22
   * 3,13,23
   * trace = StopWatch '': running time (millis) = 337
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00337  100%
   * @throws Exception
   */
  @Test
  public void test_zip() throws Exception{
    trace(()->{
      Observable.zipIterable(Arrays.asList(
          Observable.intervalRange(1, 4, 50,100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()),
          Observable.intervalRange(11, 3, 40,100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()),
          Observable.range(21, 4).delay(100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io())),
          a-> Arrays.stream(a).map(String::valueOf).collect(joining(",")),true,10
      ).blockingSubscribe(System.out::println);
    });
  }
  
  
  /**
   * 1,11,21
   * 1,11,22
   * 1,11,23
   * 1,11,24
   * 1,12,24
   * 2,12,24
   * 2,13,24
   * trace = StopWatch '': running time (millis) = 361
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 00361  100%
   * @throws Exception
   */
  @Test
  public void test_combineLatest() throws Exception{
    trace(()->{
      Observable.combineLatest(Arrays.asList(
          Observable.intervalRange(1, 2, 50,100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()),
          Observable.intervalRange(11, 3, 40,100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()),
          Observable.range(21, 4).delay(100,TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io())),
          a-> Arrays.stream(a).map(String::valueOf).collect(joining(",")),10
      ).blockingSubscribe(System.out::println);
    });
  }
  
  
  /**
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 23256  054%  without-buffer
   * 19987  046%  with-buffer
   *
   * @throws Exception
   */
  @Test
  public void test_without_backPressure() throws Exception{
    
    StopWatch stopWatch = new StopWatch();
 
    Observable<String> observable = Observable.<String>create(emitter -> {
      try {
        
        int pageNo = 1;
        
        while (pageNo <= 5) {
          
          final int tmp = pageNo;
          
          log.info("start read pageNo={}", pageNo);
          
          IntStream.range(0,10).mapToObj(i-> String.valueOf((tmp-1)*10000+i)).forEach(sku->{
  
            sleep(10, TimeUnit.MILLISECONDS);
        
            emitter.onNext(sku);
          });
          
          log.info("end read pageNo={}", pageNo);
          
          ++ pageNo;
        }
        
        emitter.onComplete();
      } catch (Exception e) {
        log.error("{}", e);
      } finally {
        emitter.onComplete();
      }
    }).subscribeOn(Schedulers.io());
  
    
    stopWatch.start("without-buffer");
    
    observable
    .doOnNext(list-> log.info("process {}", list))
    .flatMap(a-> Observable
        .fromCallable(()->findById(a))
        .subscribeOn(Schedulers.io()),
        false, 3)
    .blockingSubscribe();
    
    stopWatch.stop();
  
    
    stopWatch.start("with-buffer");
    observable
        .buffer(3)
        .flatMapIterable(list-> Observable.fromIterable(list)
            .flatMap(a-> Observable
                .fromCallable(()-> findById(a))
                .subscribeOn(Schedulers.io()),
                false, list.size())
            .toList()
            .blockingGet())
        .blockingSubscribe();
    stopWatch.stop();
    
    log.info("{}", stopWatch.prettyPrint());
  
  }
  
  
  /**
   * -----------------------------------------
   * ms     %     Task name
   * -----------------------------------------
   * 04152  055%  manual subscribe
   * 03362  045%  with buffer
   * @throws Exception
   */
  @Test
  public void test_flowable_backpressure() throws Exception{
    
    Flowable<String> flowable = Flowable.<String>create(emitter -> {
      try {
        
        int pageNo = 1;
        
        while (pageNo < 5) {
          
          final int tmp = pageNo;
          
          log.info("start pageNo={}", pageNo);
          
          Flowable.range(1,10).map(i-> String.valueOf(tmp*10000+i)).blockingSubscribe(sku-> {
            while(emitter.requested()==0){
              if(emitter.isCancelled()){
                break;
              }
            }
            sleep(10, TimeUnit.MILLISECONDS);
            emitter.onNext(sku);
          });
          
          log.info("end pageNo={}", pageNo);
          
          ++ pageNo;
        }
        
        emitter.onComplete();
      } catch (Exception e) {
        log.error("{}", e);
      } finally {
        emitter.onComplete();
      }
    }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io());
  
  
    StopWatch stopWatch = new StopWatch();
    
    stopWatch.start("manual subscribe");
    Subscriber<String> subscriber = new Subscriber<String>() {
      
      Subscription mSubscription;
      
      @Override
      public void onSubscribe(Subscription s) {
        log.info("onSubscribe");
        mSubscription = s;
        s.request(1);
      }
      
      @Override
      public void onNext(String sku) {
        log.info("onNext: {}", sku);
        mSubscription.request(1);
        sleep(100, TimeUnit.MILLISECONDS);
      }
      
      @Override
      public void onError(Throwable t) {
        log.info("onError: ", t);
        mSubscription.cancel();
      }
      
      @Override
      public void onComplete() {
        log.info("onComplete");
        mSubscription.cancel();
      }
    };
    
    flowable.blockingSubscribe(subscriber);
    stopWatch.stop();
    
    stopWatch.start("with buffer");
    flowable
        .buffer(3)
        .doOnNext(list-> log.info("process batch {}", list))
        .map(list-> Flowable.fromIterable(list)
            .flatMap(a-> Flowable
                .fromCallable(()->findById(a))
                .subscribeOn(Schedulers.io()),
                false,
                list.size())
            .toList()
            .blockingGet())
        .flatMap(Flowable::fromIterable)
        .blockingSubscribe();
    stopWatch.stop();
    
    log.info("{}", stopWatch.prettyPrint());
  
  }
  
}

rxjava2 flowable/observable操作符匯總

rxjava2 flowable/observable操作符匯總

Rxjava2 操作符匯總

構造

加時

異常

并發(fā)與并行

獲取結果

限流

合并源

轉換

日志

新增操作

實例