首页 > 其他分享 >使用Spring Reactor优化推荐流程

使用Spring Reactor优化推荐流程

时间:2022-12-06 19:37:17浏览次数:68  
标签:异步 Reactor -- Spring 流程 获取 CompletableFuture 召回 画像

1. 背景

公司有一个推荐系统Rec,这个系统的主要功能是:

  1. 向外部系统提供推荐接口
  2. 根据请求获取推荐策略
  3. 根据推荐策略完成推荐的召回、过滤、打分、排序阶段

Rec作为微服务中的一环,本身不存储召回的物料信息,也不存储用户和物料的特征信息,它负责就是对各个服务的组合和流转

其流程如下:

流程图 (1).jpg

2. 问题

在开发Rec的过程中,发现流程中存在可以优化的地方,例如:

流程 问题
“合并节点”需要等待所有的召回结果完成之后merge到一个List,然后去获取详情信息 获取详情信息是一个需要分批的IO操作,既然需要分批,为什么不是每个召回完成就去获取详情,而要等待合并。即使不需要分批,获取详情也会随着召回结果的数量变多而耗时变长
“过滤节点”需要先准备过滤的数据,等所有数据准备好,再进行过滤 获取数据是IO操作,过滤是CPU密集操作,是不是可以获取完一部分数据就可以立即进行过滤

上述的问题是纯粹从性能方面去考虑,目前的流程从逻辑是更容易理解的

2.1 优化流程图

  1. 合并节点
flowchart LR 画像召回 --> 合并 类目召回 --> 合并 ItemCF召回 --> 合并 合并 --> 获取详情 flowchart LR 画像召回 --> 获取详情1 类目召回 --> 获取详情2 ItemCF召回 --> 获取详情3 获取详情1 --> 合并 获取详情2 --> 合并 获取详情3 --> 合并
  1. 过滤节点
flowchart LR 获取质量分数据 --> 完成 获取重复展示数据 --> 完成 获取黑名单数据 --> 完成 完成 --> 过滤 flowchart LR 获取质量分数据 --> 过滤1 --> 完成 获取重复展示数据 --> 过滤2 --> 完成 获取黑名单数据 --> 过滤3 --> 完成

3. 分析

我们分析上述问题,并提出了的初步的优化方式

我们还要考虑下面几个问题:

  1. 这样的优化是不是有效的,收益如何,从性能和代码理解上
  2. 如何具体实现

上面提出的两个问题本质上是属于流程统筹方面的问题,在这个流程中,我们还是可以找到一些别的优化的地方:

例如“路由节点获取准备数据”是不是可以不用等待,提交完异步任务之后直接继续,在使用的时候去直接拿,如果发现还没有拿到数据就等着,有数据就直接使用,而不是原来的等待拿到数据再继续下一步

flowchart LR 画像召回 --> id{获取画像数据} -->|已获取到| 继续 id{获取画像数据} -->|没获取到| 等待 等待 --> id{获取画像数据}

当存在多个获取任务时,会缩短执行的时间,举个例子:

原先的流程:都使用线程池都来完成,合并前花费10ms,合并后花费10ms,总耗时20ms

flowchart LR id["获取画像数据完成(10ms)"] -->合并 --> id2["画像召回(5ms)"] id1["获取热门类目数据完成(5ms)"]-->合并 --> id3["热门类目召回(10ms)"]

新的流程:都使用线程池都来完成,总耗时15ms

flowchart LR id["获取画像数据完成(10ms)"] --> id2["画像召回(5ms)"] -->合并 id1["获取热门类目数据完成(5ms)"] --> id3["热门类目召回(10ms)"]-->合并

上述的流程图展示出来和上面的提出的两个问题属于一类,但这个问题不能简单改变流程,因为:

  1. 从流程图中可以看出来它们离得太远,而移到一起这个逻辑会被打散,没有现在的直观
  2. 准备数据是在准备了很多数据,但把其中几个拿出来的效果不能确定

所以我们需要使用一些异步编程的手段,在流程不变的情况下,还能使其执行的更快,下面表示优化的时间是从哪来的:

流程图 (5).jpg

4. 调研

在调研的过程中,发现了很多相关的技术:

  1. Reactor,Java响应式编程
  2. CompletableFuture,Java异步编程
  1. Quasar,java协程

4.1 Future

Java中提供Future可以满足我们异步的需要吗?

如果只是一个异步任务,例如画像召回需要等待画像数据,我们可以在画像召回中使用future.get()

但在画像召回完成之后,我们要进行过滤,需要提前准备一些数据,例如已展示推荐数据,使用future.get()让画像召回变成了同步任务,在获取结果之前无法继续

项目目前就是使用了线程池+Future的方案,不过future.get()都在线程池submit之后

4.2 CompletableFuture

针对上述问题, JDK8设计出CompletableFuture。CompletableFuture提供了一种观察者模式类似的机制,可以让任务执行完成后通知监听的一方

利用CompletableFuture,我们可以这样写:

    CompletableFuture recallFuture = CompletableFuture
                                        .supplyAsync(获取画像数据任务)
                                        .thenApply(画像召回)
                                        .thenAccept(获取召回信息过滤数据)
    
    CompletableFuture filterDataFuture = CompletableFuture
                                        .supplyAsync(获取过滤数据任务)
                                        
                                        
    recallFuture.thenCombineAsync(filterDataFuture, (s, w) -> {过滤数据 });

4.3 Reactor

看起来CompletableFuture已经可以满足我们的需求了,为什么需要再了解Reactor呢?它们有什么差别?

官方参考手册通过对比Callback、CompletableFuture和Reactor,它们都可以实现异步功能,但CompletableFuture/Future有下面的缺点

Future objects are a bit better than callbacks, but they still do not do well at composition, despite the improvements brought in Java 8 by CompletableFuture. Orchestrating multiple Future objects together is doable but not easy. Also, Future has other problems:

It is easy to end up with another blocking situation with Future objects by calling the get() method.

They do not support lazy computation.

They lack support for multiple values and advanced error handling.

  1. 调用future.get()就进入到了阻塞,这种情况很容易出现
  2. 不支持惰性计算,参考StackOverFlow中Oleh Dokuka的回答
  3. 不支持多个值处理和高级错误处理,查看官方参考手册中的例子可以看出

除此之外,官方参考手册介绍了一些其他特点,参考3.3

4.3.1 实现

如何使用Reactor实现上述功能呢?

         Mono<String> recallMono = Mono
                .fromCallable(() -> "获取画像数据")
                .flatMap((portraitData) -> Mono.fromCallable(() -> "画像召回"))
                .flatMap((recItemData) -> Mono.fromCallable(() -> "获取召回信息过滤数据"));

        Mono<String> filterDataMono = Mono
                .fromCallable(() -> "获取过滤数据任务");

        Mono.zip(recallMono, filterDataMono).filter((t)->true);

Java大部分的library都是同步的(HttpClient,JDBC),Mono可以和Future组合使用线程来实现异步任务,Java也存在一些异步库例如Netty,Redis Luttuce.

4.4 Quasar

至于为什么会提到Quasar,贝壳技术 | 响应式编程和协程在 Java 语言的应用中介绍了响应式编程和协程一起使用的场景,给出了原因:

  1. 响应式编程必须使用异步才能发挥其作用
  2. Java中异步的唯一解决方案就是线程
  3. 过多的线程会造成OOM,所以需要使用协程

我个人觉得响应式编程本质是从统筹学来优化程序的,最著名的例子就是烧水泡茶流程,我们只不过是通过合理编排让硬件资源最大化利用。

具体实现是将原本同步逻辑中的片段打散到不同的线程中去异步执行,原本同步阻塞的线程这时候可以给别的任务使用,应该会减少更多线程的使用

5. 实现

流程图 (2).jpg

  1. 使用线程池来处理Reactor中的异步任务
  2. 使用flatMap、map编排后续任务
  3. 使用Flux表示推荐结果流,通过不同召回不断把召回结果sink到流中
  4. 使用buffer来处理分批任务
  5. 使用zip或者flatMap来处理并发任务
  6. 使用distinct去重
  7. 使用block异步转同步获取推荐结果

6. 效果

待更新

参考

[1] 贝壳技术 | 响应式编程和协程在 Java 语言的应用
[2] 异步编程利器:CompletableFuture详解 |Java 开发实战
[3] Reactor Java文档
[4] 并发模型之Actor和CSP
[5] RxJava VS Reactor
[6] CompletableFuture原理与实践-外卖商家端API的异步化

标签:异步,Reactor,--,Spring,流程,获取,CompletableFuture,召回,画像
From: https://www.cnblogs.com/songjiyang/p/16960263.html

相关文章

  • 微服务开发平台 Spring Cloud Blade 部署实践
    本文介绍使用Rainbond快速部署SpringCloudBlade微服务平台。SpringCloudBlade是一个由商业级项目升级优化而来的微服务架构,采用SpringBoot2.7、SpringCloud2......
  • 多数据源配置 springboot+druid+mybatisplus使用注解整合
     1.pom.xml<?xmlversion="1.0"encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"......
  • SpringMVC
    回顾MVC什么是MVCMVC是模型(Model)、视图(View)、控制器(Controller)的简写,是一种软件设计规范。是将业务逻辑、数据、显示分离的方法来组织代码。MVC主要作用是......
  • Spring for GraphQL入门教程(二)
    5.带注释的控制者SpringforGraphQL提供了一个基于注释的编程模型,其中组件使用注释来声明具有灵活方法签名的处理程序方法获取特定GraphQL字段的数据。例如:​​@Contr......
  • Spring Session for Apache Geode 教程
    1.简介Spring会话提供了用于管理用户会话信息的API和实现。它还提供与以下各项的透明集成:HttpSession-使它能够被集群化(即复制)实现高可用性),而无需绑定到特定于应用程......
  • 客户生命周期流程设计
    需求:记录一个客户从注册到注销经历的各个生命周期节点,给与查询展示设计:1、生命周期节点分为以下几个流程:注册、授信、资料获取...(先开发部分流程,剩余用...表示)......
  • 天大福利!世界第一科技出版公司 Springer 免费开放 400 多本电子书!
    前几天,世界著名的科技期刊/图书出版公司施普林格(Springer)宣布:免费向公众开放400多本正版的电子书!!Springer即施普林格出版社,于1842年在德国柏林创立,20世纪60年代建立了......
  • 微服务开发平台 Spring Cloud Blade 部署实践
    本文介绍使用Rainbond快速部署SpringCloudBlade微服务平台。SpringCloudBlade是一个由商业级项目升级优化而来的微服务架构,采用SpringBoot2.7、SpringCloud20......
  • Spring for GraphQL入门
    1.概述SpringforGraphQL为基于GraphQLJava 构建的Spring应用程序提供支持。这是GraphQLJava团队和Spring工程。SpringforGraphQL是GraphQLJavaSpring项目......
  • SpringBoot文件分片上传,断点续传
    ​ 1 背景用户本地有一份txt或者csv文件,无论是从业务数据库导出、还是其他途径获取,当需要使用蚂蚁的大数据分析工具进行数据加工、挖掘和共创应用的时候,首先要将本地文......