PROSAGA-DDS/Mqtt/MODBUS-从微服务到工作流：Jet订单系统演变过程分享

<h2><a href="#jet的订单管理系统-oms-负责许多业务功能-" name="jet的订单管理系统-oms-负责许多业务功能-">Jet的订单管理系统（OMS）负责许多业务功能：</a></h2>订单初始化和验证 收费/信贷/资金管理 订单履行整合 订单历史 优惠（退款，退货等） OMS的上述功能已经基于微服务使用pub / sub、事件溯源、HTTP调用和一些其他技术进行了组合开发和运行。然而，随着Jet的发展和业务需求的扩大，系统和架构的复杂性也在增加，以至于我们运行了数十种服务。随着服务数量的增加，维护和改进系统变得更加困难。这导致更长的功能开发周期，因为几乎所有功能的逻辑都分布在多个服务上，这使得维护和调试很难。大多数基于微服务的架构都是这种情况，因为它们通常与输入/输出，SLA和个人服务职责紧密相关，所有这些都使其更难以适应不断变化的需求。对于Jet的系统，每个“业务流程”都是通过一组处理该流程特定操作的微服务进行编排的，即（创建订单，更新订单历史记录，发送交易电子邮件等）在我们基于微服务的架构中，每个服务都使用相同的样板实现： <code>
输入流incoming-streams
|> 解码decode (DomainEvent -> Input option)
|> 处理handle (Input -> Async<Output>)
|> 解释interpret (Output -> Async<unit>)
</code> 解码功能是消费使用输入流中的领域事，把它变成了输入类型，并送喂到处理handle功能; 在处理功能这块将执行各种检查或收集所需的任何其他数据，然后将其传递给执行副作用的解释函数。这个过程本质上很复杂，需要大量的样板才能有效地建模和构建。虽然我们使用聊上述这种decode -> handle -> interpret的管道作为构建服务的模板，但仍然要求每个微服务具有可扩展性，性能调整，幂等性，错误处理和重试，日志，指标，仪表板等纳入设计。随着系统的发展，构建/维护基于微服务的体系结构的复杂性最终会对整个系统和团队产生负面影响。这就是为什么在2017年1月，我们开始建模并创建一个新的系统/平台，能够代表我们目前在微服务中的所有复杂业务流程。设计/构建一套工作流系统，现在称为OMS 2.0.OMS 2.0系统是一个工作流系统，其设计和构建旨在帮助促进Jet订单处理系统的改进开发，可扩展性和可扩展性。工作流系统的核心设计基于我们在旧的基于微服务的架构中尝试提供的一组保证：幂等性 - 基于唯一标识符去除重复事件（触发器）的能力。 一致性 - 作为我们的状态管理层为多个不同的后备存储提供支持。这意味着我们还提供基于后备存储的可配置的一致性模型。我们的状态管理层的实现将总是模仿强一致性模型，因为我们需要能够读取我们自己的写入。 与我们之前基于微服务的架构相比，这些保证最终促成了开发系统时提供了许多额外的功能，这些功能是我们基于微服务的架构与新工作流系统之间的一些关键差异。工作流新系统的主要功能是：深度工作流 - 工作流表示为DAG（有向非循环图），但可以按照您希望的方式嵌套。 EventSourcing事件溯源 - 状态更改都是完全事件源于日志，而无需了解事件源语义。该系统的设计旨在鼓励用“事件”进行思考。注意：OMS系统始终将事件源作为关键组件，但是使用了工作流新系统以后，它直接进入系统默认方式。 简单实现 - 业务功能被工作流定义和相应的步骤驱动实现，这迫使系统实现模块化，并迫使开发人员在进入实现之前提前思考并思考业务流程。 可重用性 - 在设计时考虑了可重用性，但为开发人员留下了足够的灵活性，以便他们可以根据自己的需要设计流程。当需要将新工作流引入现有流时，开发人员可以从现有步骤创建新流，也可以在工作流级重用。例如，为了处理Jet内的数字SKU（Apple Care），我们添加了新的工作流程来处理保修，同时允许订单工作流程的其余部分保持不变。这使我们能够快速迭代并以极快的速度向系统提供新功能。 验证/回归 - 通过工具验证步骤行为可以快速运行回归测试。我们在Jet上充分利用了这一功能，我们的工作流程实现具有超过80％的代码覆盖率，而无需编写专用测试。只需单击按钮即可轻松生成新测试。 幂等 - 为工作流程提供幂等保证。通过从工作流DSL配置的唯一标识符的组合来实现幂等性。 系统可扩展性 - 系统本质上是弹性的。可以轻松扩展系统以提高业务流的吞吐量。在我们过去系统中，系统的规模受到核心服务之间通信的分区消费者信道的数量的限制。 工作流版本控制 - 系统在执行工作流程期间会维护工作流程的版本。也就是说，一旦实例启动，工作流就不会更改。这使我们可以将更改部署到工作流程中，而无需担心运行中的那些执行。由于工作流是业务流的表示，因此我们可以独立地迭代业务流。 低级别问题只需处理一次 - 可伸缩性，性能，幂等性，重试次数，错误处理等都在平台级而不是在每个微服务中处理。 度量标准/监控 - 系统级别度量标准易于控制，可轻松跟踪/跟踪和监控业务流程执行。 状态管理 - 这是所有状态变化的单一事实来源，称为Journal。Journal充当为系统提供动力的标准日志。它还充当调试工具，以完全了解指定工作流的历史记录。与我们历史上使用的事件源微服务相比，对状态变化的可追溯性是一个很大的不同。 对延迟工作流的支持 - 能够配置工作流定义以允许一次只执行一个工作流。在基于微服务的架构中实现这种能力非常具有挑战性。 手动审核 - 列入黑名单或有一些无根据失败的工作流程是为了人工审核而编写的，可以通过虚拟化进行审核和重新提交（工作流程系统的前端，后续帖子中有更多内容）<h2><a href="#架构概述" name="架构概述">架构概述</a></h2>新的工作流程系统很大程度上受到了Pat Helland的Life Beyond Distributed Transactions的启发，并被设计为一个双层架构：基础架构层 - 处理诸如可扩展性，幂等性和正确性，错误处理/重试，日志记录，度量等问题。目标是一次性解决这些问题，而不是针对每个服务或用例。 工作流层 - 与规模无关并处理实际的业务实现，这是工作流DSL和相应的步骤实现。 系统架构本质上是decode -> handle -> interpret我们在微服务中使用的管道的高度抽象版本。工作流系统也是采用此管道模板，并通过在每个操作之间绘制服务边界来进一步分离关注点：工作流触发器（解码） 工作流程执行器（句柄） 副作用执行者（解释）![](/user/files/NhN4XriUxPVpR7zWsBYGWdG1Umq3tCVBKw3aulIEWNY.png)由于我们使用上述体系结构定义了非常明确的服务边界，因此可以使工作流的执行高度并行化，每个服务的多个实例都可以独立扩展。<h2><a href="#工作流定义" name="工作流定义">工作流定义</a></h2>使用F＃DSL（域特定语言）定义工作流。但是，该系统并不严格限于F＃DSL，此DSL已被证明可与其他语言（如Javascript）一起使用。工作流DSL定义了一系列需要执行的步骤。工作流定义的示例：
<pre><code> [<Workflow>]
 let sampleWorkflow =
 workflow "SampleWorkflow"
 [Trigger.Stream ("kafka://jetkafka/mock-input", TaskIdType.FromPath(Path.JsonPath("$.orderId")), Path.JsonPath("$.orderId"))]
 mockWorkflowMetadata
 (step("CreateOrder", "CreateOrder") =>
 step("ReserveInventory", "ReserveInventory") => 
 step("ShipOrder", "ShipOrder") =>>
 [
 step("ChargeCustomer", "ChargeCustomer") =?>
 [
 cond("WriteChargeSuccess", "WriteChargeSuccess", Condition.Simple(Qualifier.State, "$.transactionSuccess", "true"))
 cond("WriteChargeFailure", "WriteChargeFailure", Condition.Simple(Qualifier.State, "$.transactionSuccess", "false"))
 ]
 step("UpdateOrderHistory", "UpdateOrderHistory")
 ]
 )
</code></pre>我将在本系列的第二篇文章中讨论DSL如何工作的细节，现在，需要注意的重要部分是工作流必须包含三个方面：触发：我们应该如何触发这个工作流程，即Kafka消息，服务总线消息，EventStore Steam，REST等。 元数据：这是控制工作流的元参数，如重试，并发锁定，聚合等。 步骤：工作流程的关键行为是什么或步骤应该是什么？我们应该以什么顺序执行这些步骤，如果有条件步骤，我们可以并行执行多个步骤等。 Side-Effects副作用如何在这个DSL中代表？答案是它们是在步骤定义中实现的： ``` [] module CheckApiHealth = open OMS.Common open OMS.Workflow.Core.Types
<pre><code>[<WorkflowStep>]
let CheckApiHealth (state : State) (aggregate : OrderAggregate) (input : TestInput) =
 let endpoint = sprintf "http://fakeApi?omsservice-method=get"
 let request = obj()
 let healthCheckEvt = 
 { DomainEvent.empty with 
 name = "Health"
 data = request |> Json.serializeToBytes }
 let healthCheckSideEffect = 
 healthCheckEvt |> SideEffect.fromUri endpoint

StepEvaluation.Result(state, "", [healthCheckSideEffect])
</code></pre>
<pre><code>每个步骤都必须返回StepEvaluation.Result包含三个参数的类型：

状态：步骤之间传递的当前状态
输入/输出：步骤能够传递输出，该输出在输入时提供给前一步骤。
副作用：步骤需要执行的副作用列表。
这些参数用于协调工作流引擎中的步骤。详细介绍了这些过程的工作原理和方式是后期文章的主题。

可视化展现

工作流程系统还包括一个名为可视化Visualizer工具的工具，可让用户获得有关工作流程的所见即所得。通过在工作流日志中显示详细信息，Visualizer可以显示正在进行的工作流程以及历史工作流程的工作流程详细信息。下面显示的是Visualizer如何支持检查工作流的任何单次执行的示例。

面介绍了一些重要的亮点：

验证和回归测试：能够验证先前执行工作流程的一个或多个步骤。
手动审核：能够检查并重新提交失败的工作流程或副作用
自我记录：能够在工作流程执行的任何时刻检查每个工作流程和每个步骤的状态，输入和输出。
一些统计数据

以下是我们一年多来运行的生产实例的一些统计数据。大多数这些工作流程都在标准订单处理流程中使用：

Journal工作流历史纪录： 2千2百万

工作流实例完成： 9千3百万

唯一步骤：197

独立工作流定义：18

未来扩展

以下功能是作为工作流引擎的逻辑扩展进行讨论的内容。

支持Lambda（或类似）无服务器功能。这些函数将充当工作流的步骤实现，Orchestrator将利用这些实现。
适用于OMS 2.0的PAAS模型，其中核心平台已部署一次，但新工作流程可上载到现有运行部署中。使用.Net是支持这一点的限制因素，但是，最近添加的Javascript支持允许这样做。
支持.Net Core和Linux容器
备择方案

我们并不知道有工作流程编排和设计的替代方案，如：

Netflix Conductor
Amazon State Language + AWS Lambda
Workflow Engine
Workflow Foundation
BizTalk
我们选择自己构建OMS 2.0平台有几个原因，而不是试图改进上述开箱即用的东西（banq注：关键时上述工作流引擎都不是基于事件溯源构建的）：

能够维护单独的数据存储以保存工作流事件以从故障中恢复
状态跟踪和管理，能够在执行的任何时刻重放或可视化状态
用于流和流程工作流可视化的UI
与现有基础架构集成，并与现有技术堆栈集成（Microsoft Azure + F＃）
根据业务需求添加新功能或功能的可扩展性
可扩展性，在多个区域中跨多个VM扩展工作流程执行的能力。
一些替代方案具有一些此功能，但缺乏对可扩展性，状态跟踪/管理等内容的支持，这最终导致我们决定构建自己的系统。

## 结论

从基于分布式微服务的体系结构迁移到基于工作流的体系结构，对我们的开发，支持和设计开销产生了巨大影响。作为DSL设计和合理化复杂业务流程然后实施单一责任步骤的能力对我们创新和构建复杂系统的能力产生了深远的影响。工作流引擎提供的其他好处，如工具，故障排除，可扩展性。,## Jet的订单管理系统（OMS）负责许多业务功能：

订单初始化和验证
收费/信贷/资金管理
订单履行整合
订单历史
优惠（退款，退货等）
OMS的上述功能已经基于微服务使用pub / sub、事件溯源、HTTP调用和一些其他技术进行了组合开发和运行。然而，随着Jet的发展和业务需求的扩大，系统和架构的复杂性也在增加，以至于我们运行了数十种服务。随着服务数量的增加，维护和改进系统变得更加困难。这导致更长的功能开发周期，因为几乎所有功能的逻辑都分布在多个服务上，这使得维护和调试很难。

大多数基于微服务的架构都是这种情况，因为它们通常与输入/输出，SLA和个人服务职责紧密相关，所有这些都使其更难以适应不断变化的需求。对于Jet的系统，每个“业务流程”都是通过一组处理该流程特定操作的微服务进行编排的，即（创建订单，更新订单历史记录，发送交易电子邮件等）

在我们基于微服务的架构中，每个服务都使用相同的样板实现：
</code></pre>输入流incoming-streams |> 解码decode (DomainEvent -> Input option) |> 处理handle (Input -> Async) |> 解释interpret (Output -> Async) ``` 解码功能是消费使用输入流中的领域事，把它变成了输入类型，并送喂到处理handle功能; 在处理功能这块将执行各种检查或收集所需的任何其他数据，然后将其传递给执行副作用的解释函数。这个过程本质上很复杂，需要大量的样板才能有效地建模和构建。虽然我们使用聊上述这种decode -> handle -> interpret的管道作为构建服务的模板，但仍然要求每个微服务具有可扩展性，性能调整，幂等性，错误处理和重试，日志，指标，仪表板等纳入设计。随着系统的发展，构建/维护基于微服务的体系结构的复杂性最终会对整个系统和团队产生负面影响。这就是为什么在2017年1月，我们开始建模并创建一个新的系统/平台，能够代表我们目前在微服务中的所有复杂业务流程。设计/构建一套工作流系统，现在称为OMS 2.0.OMS 2.0系统是一个工作流系统，其设计和构建旨在帮助促进Jet订单处理系统的改进开发，可扩展性和可扩展性。工作流系统的核心设计基于我们在旧的基于微服务的架构中尝试提供的一组保证：幂等性 - 基于唯一标识符去除重复事件（触发器）的能力。 一致性 - 作为我们的状态管理层为多个不同的后备存储提供支持。这意味着我们还提供基于后备存储的可配置的一致性模型。我们的状态管理层的实现将总是模仿强一致性模型，因为我们需要能够读取我们自己的写入。 与我们之前基于微服务的架构相比，这些保证最终促成了开发系统时提供了许多额外的功能，这些功能是我们基于微服务的架构与新工作流系统之间的一些关键差异。工作流新系统的主要功能是：深度工作流 - 工作流表示为DAG（有向非循环图），但可以按照您希望的方式嵌套。 EventSourcing事件溯源 - 状态更改都是完全事件源于日志，而无需了解事件源语义。该系统的设计旨在鼓励用“事件”进行思考。注意：OMS系统始终将事件源作为关键组件，但是使用了工作流新系统以后，它直接进入系统默认方式。 简单实现 - 业务功能被工作流定义和相应的步骤驱动实现，这迫使系统实现模块化，并迫使开发人员在进入实现之前提前思考并思考业务流程。 可重用性 - 在设计时考虑了可重用性，但为开发人员留下了足够的灵活性，以便他们可以根据自己的需要设计流程。当需要将新工作流引入现有流时，开发人员可以从现有步骤创建新流，也可以在工作流级重用。例如，为了处理Jet内的数字SKU（Apple Care），我们添加了新的工作流程来处理保修，同时允许订单工作流程的其余部分保持不变。这使我们能够快速迭代并以极快的速度向系统提供新功能。 验证/回归 - 通过工具验证步骤行为可以快速运行回归测试。我们在Jet上充分利用了这一功能，我们的工作流程实现具有超过80％的代码覆盖率，而无需编写专用测试。只需单击按钮即可轻松生成新测试。 幂等 - 为工作流程提供幂等保证。通过从工作流DSL配置的唯一标识符的组合来实现幂等性。 系统可扩展性 - 系统本质上是弹性的。可以轻松扩展系统以提高业务流的吞吐量。在我们过去系统中，系统的规模受到核心服务之间通信的分区消费者信道的数量的限制。 工作流版本控制 - 系统在执行工作流程期间会维护工作流程的版本。也就是说，一旦实例启动，工作流就不会更改。这使我们可以将更改部署到工作流程中，而无需担心运行中的那些执行。由于工作流是业务流的表示，因此我们可以独立地迭代业务流。 低级别问题只需处理一次 - 可伸缩性，性能，幂等性，重试次数，错误处理等都在平台级而不是在每个微服务中处理。 度量标准/监控 - 系统级别度量标准易于控制，可轻松跟踪/跟踪和监控业务流程执行。 状态管理 - 这是所有状态变化的单一事实来源，称为Journal。Journal充当为系统提供动力的标准日志。它还充当调试工具，以完全了解指定工作流的历史记录。与我们历史上使用的事件源微服务相比，对状态变化的可追溯性是一个很大的不同。 对延迟工作流的支持 - 能够配置工作流定义以允许一次只执行一个工作流。在基于微服务的架构中实现这种能力非常具有挑战性。 手动审核 - 列入黑名单或有一些无根据失败的工作流程是为了人工审核而编写的，可以通过虚拟化进行审核和重新提交（工作流程系统的前端，后续帖子中有更多内容）<h2><a href="#架构概述" name="架构概述">架构概述</a></h2>新的工作流程系统很大程度上受到了Pat Helland的Life Beyond Distributed Transactions的启发，并被设计为一个双层架构：基础架构层 - 处理诸如可扩展性，幂等性和正确性，错误处理/重试，日志记录，度量等问题。目标是一次性解决这些问题，而不是针对每个服务或用例。 工作流层 - 与规模无关并处理实际的业务实现，这是工作流DSL和相应的步骤实现。 系统架构本质上是decode -> handle -> interpret我们在微服务中使用的管道的高度抽象版本。工作流系统也是采用此管道模板，并通过在每个操作之间绘制服务边界来进一步分离关注点：工作流触发器（解码） 工作流程执行器（句柄） 副作用执行者（解释）![](/user/files/NhN4XriUxPVpR7zWsBYGWdG1Umq3tCVBKw3aulIEWNY.png)由于我们使用上述体系结构定义了非常明确的服务边界，因此可以使工作流的执行高度并行化，每个服务的多个实例都可以独立扩展。<h2><a href="#工作流定义" name="工作流定义">工作流定义</a></h2>使用F＃DSL（域特定语言）定义工作流。但是，该系统并不严格限于F＃DSL，此DSL已被证明可与其他语言（如Javascript）一起使用。工作流DSL定义了一系列需要执行的步骤。工作流定义的示例：
<pre><code> [<Workflow>]
 let sampleWorkflow =
 workflow "SampleWorkflow"
 [Trigger.Stream ("kafka://jetkafka/mock-input", TaskIdType.FromPath(Path.JsonPath("$.orderId")), Path.JsonPath("$.orderId"))]
 mockWorkflowMetadata
 (step("CreateOrder", "CreateOrder") =>
 step("ReserveInventory", "ReserveInventory") => 
 step("ShipOrder", "ShipOrder") =>>
 [
 step("ChargeCustomer", "ChargeCustomer") =?>
 [
 cond("WriteChargeSuccess", "WriteChargeSuccess", Condition.Simple(Qualifier.State, "$.transactionSuccess", "true"))
 cond("WriteChargeFailure", "WriteChargeFailure", Condition.Simple(Qualifier.State, "$.transactionSuccess", "false"))
 ]
 step("UpdateOrderHistory", "UpdateOrderHistory")
 ]
 )
</code></pre>我将在本系列的第二篇文章中讨论DSL如何工作的细节，现在，需要注意的重要部分是工作流必须包含三个方面：触发：我们应该如何触发这个工作流程，即Kafka消息，服务总线消息，EventStore Steam，REST等。 元数据：这是控制工作流的元参数，如重试，并发锁定，聚合等。 步骤：工作流程的关键行为是什么或步骤应该是什么？我们应该以什么顺序执行这些步骤，如果有条件步骤，我们可以并行执行多个步骤等。 Side-Effects副作用如何在这个DSL中代表？答案是它们是在步骤定义中实现的： ``` [] module CheckApiHealth = open OMS.Common open OMS.Workflow.Core.Types
<pre><code>[<WorkflowStep>]
let CheckApiHealth (state : State) (aggregate : OrderAggregate) (input : TestInput) =
 let endpoint = sprintf "http://fakeApi?omsservice-method=get"
 let request = obj()
 let healthCheckEvt = 
 { DomainEvent.empty with 
 name = "Health"
 data = request |> Json.serializeToBytes }
 let healthCheckSideEffect = 
 healthCheckEvt |> SideEffect.fromUri endpoint

StepEvaluation.Result(state, "", [healthCheckSideEffect])
</code></pre>``` 每个步骤都必须返回StepEvaluation.Result包含三个参数的类型：状态：步骤之间传递的当前状态 输入/输出：步骤能够传递输出，该输出在输入时提供给前一步骤。 副作用：步骤需要执行的副作用列表。 这些参数用于协调工作流引擎中的步骤。详细介绍了这些过程的工作原理和方式是后期文章的主题。可视化展现工作流程系统还包括一个名为可视化Visualizer工具的工具，可让用户获得有关工作流程的所见即所得。通过在工作流日志中显示详细信息，Visualizer可以显示正在进行的工作流程以及历史工作流程的工作流程详细信息。下面显示的是Visualizer如何支持检查工作流的任何单次执行的示例。面介绍了一些重要的亮点：验证和回归测试：能够验证先前执行工作流程的一个或多个步骤。 手动审核：能够检查并重新提交失败的工作流程或副作用 自我记录：能够在工作流程执行的任何时刻检查每个工作流程和每个步骤的状态，输入和输出。 一些统计数据以下是我们一年多来运行的生产实例的一些统计数据。大多数这些工作流程都在标准订单处理流程中使用：Journal工作流历史纪录： 2千2百万工作流实例完成： 9千3百万唯一步骤：197独立工作流定义：18未来扩展以下功能是作为工作流引擎的逻辑扩展进行讨论的内容。支持Lambda（或类似）无服务器功能。这些函数将充当工作流的步骤实现，Orchestrator将利用这些实现。 适用于OMS 2.0的PAAS模型，其中核心平台已部署一次，但新工作流程可上载到现有运行部署中。使用.Net是支持这一点的限制因素，但是，最近添加的Javascript支持允许这样做。 支持.Net Core和Linux容器 备择方案我们并不知道有工作流程编排和设计的替代方案，如：Netflix Conductor Amazon State Language + AWS Lambda Workflow Engine Workflow Foundation BizTalk 我们选择自己构建OMS 2.0平台有几个原因，而不是试图改进上述开箱即用的东西（banq注：关键时上述工作流引擎都不是基于事件溯源构建的）：能够维护单独的数据存储以保存工作流事件以从故障中恢复 状态跟踪和管理，能够在执行的任何时刻重放或可视化状态 用于流和流程工作流可视化的UI 与现有基础架构集成，并与现有技术堆栈集成（Microsoft Azure + F＃） 根据业务需求添加新功能或功能的可扩展性 可扩展性，在多个区域中跨多个VM扩展工作流程执行的能力。 一些替代方案具有一些此功能，但缺乏对可扩展性，状态跟踪/管理等内容的支持，这最终导致我们决定构建自己的系统。<h2><a href="#结论" name="结论">结论</a></h2>从基于分布式微服务的体系结构迁移到基于工作流的体系结构，对我们的开发，支持和设计开销产生了巨大影响。作为DSL设计和合理化复杂业务流程然后实施单一责任步骤的能力对我们创新和构建复杂系统的能力产生了深远的影响。工作流引擎提供的其他好处，如工具，故障排除，可扩展性。

从微服务到工作流：Jet订单系统演变过程分享(微服务)