高性能工作流引擎：DataBuilder与polaris


DataBuilder 框架是一个高级逻辑执行引擎，可用于执行多步骤工作流。该引擎目前为 Flipkart 的结账系统以及诊断和其他工作流程提供支持。您应该针对以下场景查看此框架：

  • 多步骤工作流程执行，其中每个步骤都依赖于先前步骤生成的数据
  • 执行可以跨越一个或多个请求范围
  • 您的系统使用可重复使用的组件，这些组件可以以不同的方式组合以生成不同的最终结果

上述的几个例子是：
  • 类似结帐的系统，用户可以在多个步骤中提供全部或仅部分数据，并根据订单可能完成或将控制权返回给用户。随着用户填写更多详细信息，系统会更接近目标，最终在用户提供所有详细信息后生成订单。
  • API 网关结合多个来源的数据以生成最终响应

以下是显着特征：
  • 为构建者提供基于注释的元数据处理
  • 数据流分析器和构建器根据提供的元数据和目标动态生成执行图
  • 支持在一个执行范围的上下文中有意义的循环和瞬态数据
  • 单线程和多线程数据流执行器
  • 极低的开销（单线程执行器为 20 usec）
  • 公开低级和高级 API，用于动态注册构建器和目标以及构建流程
  • 大量测试用例，涵盖框架的各个方面，可供参考

该框架可以以两种模式使用：
  • 请求范围内的流
  • 跨多个请求的流程

1、请求范围内的基本流程如下：

  • 识别和创建数据类
  • 创建构建器
  • 注册构建器元数据（消耗、产生）
  • 构建（并可选择保存）指定目标数据的数据流
  • 开始接受请求
    • 创建数据流实例
    • 接受数据增量中的输入数据
    • 使用增量执行数据流
    • 此时可能会发生以下两种情况之一：
      • 给出并生成所有所需数据并生成目标数据
      • 所有必需的数据不存在并且流程未完成
    • 在上述两种情况下，都会返回当前流程执行期间生成的所有数据的映射
    • 所有非瞬态数据都添加到数据流实例中存在的数据集中
    • 可以在后续的执行调用中提供更多数据来完成流程
    • 执行器使用数据增量和数据集中存在的数据来生成更多数据并到达生成指定目标数据的最终状态

2、跨多个请求执行流程
在这种情况下，执行更加复杂，如下所示：

  • 为用户活动上下文创建 DataFlowInstance。例如，这代表结帐会话。
    • 对于跨流更改的事务性，每个 DataFlowInstance 应包含其自己的 DataFlow 副本
  • 每次调用执行器时，都需要将 DataFlowInstance 与一组输入数据一起传入
  • 新的输入数据（也称为增量）被添加到考虑执行此执行的数据集中
  • 系统开始执行并尽可能地继续执行流程
  • 每执行一步，新生成的数据都会添加到活动数据集中
  • 如果构建器的输入不足以生成数据，则它可能会选择返回 null
  • 如果流程中未达到最终状态，系统会再次从头开始循环，使用 DataSet，现在会使用先前执行生成的数据进行扩充
  • 系统在以下情况下停止：
    • 流的目标数据已生成
    • 迭代中没有生成新数据
  • 增强数据集保存在实例中。不添加瞬态数据

示例：多步结账

//Step 1:登录和地址
response = executor.run(instance, userDetails);
//Step 2: 购物车详情和编辑
response = executor.run(instance, cart);
//Step 3: 支付
response = executor.run(instance, payment);
//Done


相关：
polaris是Golang 的高性能工作流程编排器，灵感来自DataBuilder

用例

  1. 您具有多步骤工作流执行，其中每个步骤都依赖于先前步骤生成的数据。
  2. 执行可以跨越一个请求范围或多个范围。
  3. 您的系统使用可重复使用的组件，这些组件可以以不同的方式组合以生成不同的最终结果。
  4. 您的工作流程可以暂停、恢复，甚至从头开始。

局限性

  1. 工作流版本控制实施起来很棘手：
    2. 除非您能够承受 100% 的停机时间，确保所有活动工作流程进入最终状态，否则部署新代码需要确保向后兼容性。
    3. 这意味着 - 您需要部署一个向后兼容较旧的非终端工作流程的代码版本，而较新的工作流程将在新代码上执行。
    4. 一旦旧的工作流程完成，就需要进行部署来清理过时的代码。
  2. 与 Cadence、Conductor 相比，该框架的抽象级别较低：[list=1]
  3. 如果有外部（可靠）服务为每个工作流 ID 提供回调，则可以使工作流忽略故障。
  4. 可以通过添加自定义代码以通过侦听器推送事件来设置检测。


用法

// dataStore = DataStore{} - use your database by implementing the IDataStore interface
polaris.InitRegistry(dataStore)
polaris.RegisterWorkflow(workflowKey, workflow)

executor := polaris.Executor{
    Before: func(builder reflect.Type, delta []IData) {
        fmt.Printf("Builder %s is about to be run with new data %v\n", builder, delta)
    }
    After: func(builder reflect.Type, produced IData) {
        fmt.Printf("Builder %s produced %s\n", builder, produced)
    }
}

response, err := executor.Run(workflowKey, workflowId, dataDelta)