Spring 响应式编程 随记 -- C2 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 组合两个并行流
更多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 可以通过更加通用和灵活的方式来异步组合数据流
标签：异步,订阅,Observable,Spring,编程,响应,线程,interface,query 来源： https://blog.csdn.net/ParrySMS/article/details/120761813