响应式

响应式可以理解为观察者模式,通过订阅和发布数据流中的数据对数据进行处理. reactor提供了强大的API,简化了对数据各种处理方式的复杂度,如果你已经大量使用了java8 stream api,使用reactor将很容易上手.

注意

响应式与传统编程最大的区别是: 响应式中的方法调用是在构造一个流以及处理流中数据的逻辑,当流中产生了数据(发布,订阅),才会执行构造好的逻辑. 传统编程则是直接执行逻辑获取结果.

非阻塞,集成netty等框架可实现更高的网络并发处理能力. API丰富,实现很多复杂的功能只需要几行代码,例如:

  1. 前端展示实时数据处理进度.
  2. 请求撤销,可获取到连接断开事件.
  3. 定时(interval),延迟(delay),超时(timout),以及细粒度的流量控制(limitRate).

缺点

调试不易,异常栈难跟踪,对开发人员有更高的要求.

响应式只是一个编程模型,并不能直接提高系统的并发处理能力. 通常与netty(reactor-netty)等框架配合,从上(网络)到下(持久化)全套实现非阻塞,响应式才有意义.

说明

系统中大量使用到了reactor,其核心类只有2个Flux(0-n个数据的流),Mono(0-1个数据的流). 摒弃传统编程的思想,熟悉Flux,MonoAPI,就可以很好的使用响应式编程了.

常用API:

  1. : 转换流中的元素: flux.map(UserEntity::getId)
  2. flatMap: 转换流中的元素为新的流: flux.flatMap(this::findById)
  3. flatMapMany: 转换Mono中的元素为Flux(1转多): mono.flatMapMany(this::findChildren)
  4. concat: 将多个流连接在一起组成一个流(按顺序订阅) : Flux.concat(header,body)
  5. merge: 将多个流合并在一起,同时订阅流: Flux.merge(save(info),saveDetail(detail))
  6. zip: 压缩多个流中的元素: Mono.zip(getData(id),getDetail(id),UserInfo::of)
  7. then: 流完成后执行.
  8. doOnNext: 流中产生数据时执行.
  9. doOnCancel: 流被取消时执行.
  10. onErrorContinue: 流发生错误时,继续处理数据而不是终止整个流.
  11. defaultIfEmpty: 当流为空时,使用默认值.
  12. switchIfEmpty: 当流为空时,切换为另外一个流.
  13. as: 将流作为参数,转为另外一个结果:flux.as(this::save)

完整文档请查看官方文档 (opens new window)

注意

使用reactor时,应该注意代码尽量以.换行并做好相应到缩进.例如:

lamdba

  2. //错误
  3. return devicePropertyMono
  4.         .map(prop->{
  5.             Map<String,Object> map = new HashMap<>();
  6.             map.put("property",prop.getProperty());
  7.             ....
  8.             return map;
  9.         })
  12. //建议
  13. //在DeviceProperty中编写toMap方法实现上面lambda中到逻辑.
  14. return devicePropertyMono
  15.         .map(DeviceProperty::toMap)
  16.         .flatMap(this::doSomeThing)

数据流中到元素不允许为null,因此在进行数据转换到时候要注意null处理.例如:

非阻塞与阻塞

默认情况下,reactor的调度器由数据的生产者(Publisher)决定,在WebFlux中则是netty的工作线程, 为了防止工作线程被阻塞导致服务崩溃,在一个请求的流中,禁止执行存在阻塞(如执行JDBC)可能的操作的,如果无法避免阻塞操作,应该指定调度器如:

  1.   .publishOn(Schedulers.elastic()) //指定调度器去执行下面的操作
  2.   .map(param-> jdbcService.select(param))

在响应式中,大部分情况是禁止使用ThreadLocal的(可能造成内存泄漏).因此基于ThreadLocal的功能都无法使用,reactor中引入了上下文,在一个流中,可共享此上下文 ,通过上下文进行变量共享以例如:事务,权限等功能.例如:

注意

在开发中应该将多个流组合为一个流,而不是分别处理.例如:

  1. //错误
  2. return flux.doOnNext(data->this.save(data).subscribe());
  4. //正确
  5. return flux.flatMap(this::save);
  7. //错误,没有将流组合在一起
  8. request.flatMap(this::save);
  9. Mono<Void> result = this.notifySaveSuccess();
  10. return result;
  12. //正确
  13. return request
  14.     .flatMap(this::save)
  15.     .then(this.notifySaveSuccess());

相关资料

  2. project-reactor (opens new window)
  4. simviso视频教程 (opens new window)