响应式

JetLinks使用Project Reactor响应式 - 图1 (opens new window)作为响应式编程框架,从网络层(webflux,vert.x)到持久层(r2dbc,elastic)全部 封装为非阻塞,响应式调用.

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

注意

响应式传统编程最大的区别是:

响应式中的方法调用是在构造一个流以及处理流中数据的逻辑,当中产生了数据(发布,订阅),才会执行构造好的逻辑.

传统编程则是直接执行逻辑获取结果.

优点

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

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

缺点

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

此问题可以通过优化代码结构来解决,比如: 避免在响应式操作符中直接业务逻辑, 正确的做法是将业务逻辑抽离为独立的函数(方法),然后使用响应式来进行组合.

注意

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

选择合适的操作符

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

常用操作符:

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

完整文档请查看官方文档响应式 - 图2 (opens new window)

代码格式化

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

  1. //错误
  2. return paramMono.map(param->param.getString("id")).flatMap(this::findById);
  3. //建议
  4. return paramMono
  5. .map(param->param.getString("id"))
  6. .flatMap(this::findById);

lamdba

避免在一个lambda中编写大量的逻辑代码,推荐参考领域模型,将具体当逻辑放到对应到实体或者领域对象中.例如:

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

null处理

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

  1. //存在缺陷
  2. return this.findById(id)
  3. .map(UserEntity::getDescription); //getDescription可能返回null,为null时会抛出空指针,

reactor 3.4后可以使用以下方式来处理可能存在null的map操作

  1. return this.findById(id)
  2. .mapNotNull(UserEntity::getDescription);

非阻塞与阻塞

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

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

上下文

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

  1. //从上下文中获取
  2. @GetMapping
  3. public Mono<UserInfo> getCurrentUser(){
  4. return Mono.subscriberContext()
  5. .map(ctx->userService.findById(ctx.getOrEmpty("userId").orElseThrow(IllegalArgumentException::new));
  6. }
  7. //定义过滤器设置数据到上下文中
  8. class MyFilter implements WebFilter{
  9. public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain){
  10. return chain.filter(exchange)
  11. .subscriberContext(Context.of("userId",...))
  12. }
  13. }

注意

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

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

FAQ: 我写的操作看上去是正确的,但是没有执行.

有3种可能: 上游流为空,多个流未组合在一起,在不支持响应式的地方使用了响应式

上游流为空:

例:

  1. public Mono<Response> handleRequest(Request request){
  2. return this
  3. .findOldData(request)
  4. .flatMap(old -> {
  5. //这里为什么不执行?
  6. return ....
  7. })
  8. }

说明

findOldData返回的流为空时,下游的flatMap等需要操作流中元素的操作符是不会执行的. 可以通过switchIfEmpty操作符来处理空流的情况. 例如:

  1. return this
  2. .findOldData(request)
  3. //处理没获取到数据的情况
  4. .switchIfEmpty(Mono.error(()->new NotFoundException("error.data_not_found")))
  5. .flatMap(old -> {
  6. return ....
  7. })

如果flatMapswitchIfEmpty中的逻辑都没执行,那可能是下面一种情况.

多个流未组合在一起

例:

  1. public Result handleRequest(Request request){
  2. //处理请求
  3. service.handleRequest(request);
  4. return ok;
  5. }

注意

  1. 只要方法返回值是Mono或者Flux,都不能单独行动.
  2. 只要方法中调用了任何响应式操作.那这个方法也应该是响应式.(返回Mono或者Flux)

因此正确的写法是:

  1. public Mono<Result> handleRequest(Request request){
  2. return service
  3. //处理请求
  4. .handleRequest(request)
  5. //记录日志
  6. .then(saveLog(request))
  7. //返回结果
  8. .thenReturn(ok);
  9. }

在不支持响应式的地方使用响应式

  1. public Mono<Result> handleRequest(Request request){
  2. return service
  3. //处理请求
  4. .handleRequest(request)
  5. //记录日志 此为错误的用法
  6. .doOnNext(response-> saveLog(request,response) )
  7. //返回结果
  8. .thenReturn(ok);
  9. }

说明

doOnNext方法的语义以及参数Consumer<T>可知,此方法是不支持响应式的(Consumer<T>只有参数没有返回值),因此不能在此方法中使用响应式操作.

正确的写法:

  1. return service
  2. //处理请求
  3. .handleRequest(request)
  4. //记录日志 此为错误的用法
  5. .flatMap(response-> saveLog(request,response) )
  6. //返回结果
  7. .thenReturn(ok);

FAQ: 我想获取流中的元素怎么办

不要试图从流中获取数据出来,而是先思考需要对流中元素做什么, 需要对流中的数据进行操作时,都应该使用操作符来处理,根据Flux或者Mono提供的操作符API进行组合操作.比如:

传统:

  1. public List<Book> getAllBooks(){
  2. List<BookEntity> bookEntities = repository.findAll();
  3. List<Book> books = new ArrayList(bookEntities.size());
  4. for(BookEntity entity : bookEntities){
  5. Book book = entity.copyTo(new Book());
  6. books.add(book);
  7. }
  8. return books;
  9. }

响应式:

  1. public Flux<Book> getAllBooks(){
  2. return repository
  3. .findAll()
  4. .map(entity-> entity.copyTo(new Book()))
  5. }

FAQ: 我需要在非响应式方法中使用响应式怎么办

如果需要阻塞获取结果,可以使用flux.block(timeout).

如果需要异步获取结果,可以使用flux.subscribe(data->{ },error->{ })

如:

  1. public void handleRequest(Request request){
  2. //logService.saveLog(request).block()
  3. logService
  4. .saveLog(request)
  5. .subscribe(
  6. result->{
  7. log.debug("保存成功 {}",request)
  8. },
  9. error->{
  10. log.warn("保存失败 {}",request,error);
  11. }
  12. )
  13. }

相关资料

  1. reactive-streams响应式 - 图3 (opens new window)
  2. project-reactor响应式 - 图4 (opens new window)
  3. 使用 Reactor 进行反应式编程响应式 - 图5 (opens new window)