Spring 响应式编程 随记 -- C2 Spring 响应式编程基本概念 (三)

Spring 响应式编程 随记 -- C2 Spring 响应式编程基本概念 (三)【 好书分享:《Spring 响应式编程》-- 京东】

2.2.3 生成异步序列

RxJava 可以生产一个未来的事件,还可以基于时间间隔生产一个异步事件序列。

Observable.interval(1,TimeUnit.SECONDS)
    .subcriber(msg -> System.out.println("received: " + msg));

Thread.sleep(3000); 
// output: 
// received: 0
// received: 1
// received: 2

sleep 是为了防止主线程直接完成执行。因为生成事件和消费的过程在同一个单独的守护线程中。

当然,可以用更好的 Subscription 来控制观察者-订阅者协作的关系和操作。

interface Subscription{
    void unsubscribe();
    boolean isUnsubcribed();
}

unsubscribe 方法能让 Subscriber 通知 Observable 不需要再发送新的事件,也就是源取消订阅。
isUnsubcribed 方法则反过来,让 Observable 可以调用来检查 Subscriber 是否已经取消,用于处理是否需要发消息。

示例:订阅者一直消费数据直到收到 CountDownLatch 信号。

CountDownLatch externalSignal = ... ;

Subscription subscription = Observable.interval(100, TimeUnit.MILLISECONDS)
    .subscribe(System.out::println);

externalSignal.await();
subscription.unsubscribe();

2.2.4 流转换和弹珠图

弹珠图(marble diagram) 用于可视化的表述操作符对应的流转换。操作符可以用于调整流元素以及更改流结构。

  • map 一对一的转换
  • filter 条件过滤
  • count 返回输入流的元素数量,但只在原始流结束的时候才发出结果,所以无限流不可用。
  • zip 组合两个并行流

Spring 响应式编程 随记 -- C2 Spring 响应式编程基本概念 (三)
更多API参考: https://rxmarbles.com/

除了框架自带的API,也完全可以根据自己的需求去实现自己需要的操作符。

只需要实现 Observable.Transformer<T,R> 派生的类即可,执行调用的时候使用 .compose(yourTransformer)

2.2.5 RxJava 的先决条件和优势

基本概念:订阅者订阅可观察流,流反过来触发时间生成的异步过程。

通过一个内存搜索引擎示例来进行对比。这个服务返回一个URL集合,用户搜索相关内容。

  • 传统API做法:
public interface SearchEngine{
    List<URL> search(String query, int limit);
}

服务执行,收集limit个结果,然后放到list里返回一个列表。实际业务中可能存在这样的场景,用户可能只需要前面的几个结果,不想等待整个结果集的完全计算结果。

  • 游标迭代器法
public interface IterableSearchEngine{
    Iterable<URL> search(String query, int limit);
}

客户端和服务端正在通过 Iterable 接口互相通信,客户端的线程在主动等待新的数据时会产生阻塞,交互效率不高。

  • CompletableFuture 构建异步服务
public interface FutureSearchEngine{
    CompletableFuture<List<URL>> search(String query, int limit);
}

没有请求阻塞,异步调用,但是调用的服务端仍然是计算整个结果集返回,要么线程唤起收到全部,要么线程异步去处理其他。

  • RxJava
public interface RxSearchEngine{
    Observable<URL> search(String query);
}

异步处理并且对每个事件做出响应,并且客户端可以根据业务情况随时取消订阅,减少搜索服务处理过程中所需要完成的工作量。

这样子,即使客户端没有收到所有的结果,它也可以处理已经到达的部分。

所以 RxJava 可以通过更加通用和灵活的方式来异步组合数据流

上一篇:RxJava使用详解


下一篇:熔断器 Hystrix 的原理与使用