使用Spring Reactor优化推荐流程

使用,spring,reactor,优化,推荐,流程 · 浏览次数 : 402

小编点评

**1. 优化流程** * 将合并节点合并成一个线程,以减少流程中的IO操作数量。 * 将获取详情节点的逻辑移到获取图像数据节点之前,以避免阻塞操作。 * 将获取质量分数据节点放到最后一个合并节点中,以合并所有结果。 **2. 优化技术** * 使用CompletableFuture的观察者模式来处理异步任务,以避免阻塞操作。 * 使用 reactor 的Flux进行推荐结果的处理,以实现并发和并行执行。 **3. 性能分析** * 使用性能分析工具,例如堆积分析和线程池使用分析,来优化代码性能。 * 通过调整线程池的大小和配置来优化性能。 **4. 其他** * 使用 reactor 的observables 和 streams 来处理异步事件流。 * 使用 reactor 的Flux 来处理并发任务,以提高性能。 * 使用 reactor 的buffer来处理分批任务,以提高性能。

正文

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的异步化

与使用Spring Reactor优化推荐流程相似的内容:

使用Spring Reactor优化推荐流程

1. 背景 公司有一个推荐系统Rec,这个系统的主要功能是: 向外部系统提供推荐接口 根据请求获取推荐策略 根据推荐策略完成推荐的召回、过滤、打分、排序阶段 Rec作为微服务中的一环,本身不存储召回的物料信息,也不存储用户和物料的特征信息,它负责就是对各个服务的组合和流转 其流程如下: 2. 问题

Spring Reactor基本介绍和案例

1. Reactor 对比 1.1 Reactor 线程模型 Reactor 线程模型就是通过 单个线程 使用 Java NIO 包中的 Selector 的 select()方法,进行监听。当获取到事件(如 accept、read 等)后,就会分配(dispatch)事件进行相应的事件处理(han

使用 Spring 实现控制反转和依赖注入

使用 Spring 实现控制反转和依赖注入 概述 在本文中,我们将介绍IoC(控制反转)和DI(依赖注入)的概念,以及如何在Spring框架中实现它们。 什么是控制反转? 控制反转是软件工程中的一个原则,它将对象或程序的某些部分的控制权转移给容器或框架。我们最常在面向对象编程的上下文中使用它。 与传

Spring Boot中的 6 种API请求参数读取方式

使用Spring Boot开发API的时候,读取请求参数是服务端编码中最基本的一项操作,Spring Boot中也提供了多种机制来满足不同的API设计要求。 接下来,就通过本文,为大家总结6种常用的请求参数读取方式。如果你发现自己知道的不到6种,那么赶紧来查漏补缺一下。如果你知道的不止6种,那么告诉

使用 Spring Cloud Loadbalancer 实现客户端负载均衡

使用 Spring Cloud Loadbalancer 实现客户端负载均衡 作者:Grey 原文地址: 博客园:使用 Spring Cloud Loadbalancer 实现客户端负载均衡 CSDN:使用 Spring Cloud Loadbalancer 实现客户端负载均衡 背景 在Spring

[EasyExcel] 导出合并单元格

前言 使用spring boot 对excel 进行操作在平时项目中要经常使用。常见通过jxl和poi 的方式进行操作。但他们都存在一个严重的问题就是非常的耗内存。这里介绍一种 Easy Excel 工具来对excel进行操作。 一、Easy Excel是什么? EasyExcel是阿里巴巴开源的一

我是如何使用Spring Retry减少1000 行代码

本文翻译自国外论坛 medium,原文地址:https://levelup.gitconnected.com/how-i-deleted-more-than-1000-lines-of-code-using-spring-retry-9118de29060 > 使用 Spring Retry 重构代

在线程中使用Spring的Bean的方法、不推荐把“线程”注入到Spring

一、不推荐把“线程”注入到spring 将线程注入到Spring容器中并不是一个常见的做法,而且通常也不推荐这样做,原因如下: 生命周期管理困难: Spring管理的Bean生命周期由Spring容器管理,而线程的生命周期由JVM管理。将线程注入到Spring容器中会导致线程的生命周期与Spring

Spring Data MongoDB 使用

本文为博主原创,转载请注明出处: Spring Data MongoDB 是 Spring Data 系列的一部分,它提供了与 MongoDB 的集成和操作支持,类似于 JPA 对关系型数据库的支持。通过使用 Spring Data MongoDB,您可以使用类似于 JPA 的 Repository

掌握Spring条件装配的秘密武器

本文详细解释了如何使用Spring的@Profile和@Conditional注解实现条件装配,通过具体的示例可以更好地理解这两个注解的用法和适用场景。