Netflix Conductor：流程的反转控制

本文首发于 Netflix ，经网站授权由 InfoQ 翻译为中文版分享给中文读者。

In December 2016, we open sourced Netflix Conductor .

2016 年 12 月， Netflix Conductor 开源了。

自此以后，增加了更多的特性，增加了工作流临近的用户界面，并使系统更稳定了。使用率、新的用例、特性需求都显著增加了，社区里也出现了许多有意思的事情。在本文中，我们将讨论本季度添加到 Conductor 中的几个主要特性。

Conductor 中的反转控制

Conductor 使工作流的复用以及作为子流程嵌入到其他流程中更为简单了。子流程是使过程复用和模块化的好方法。随着 Netflix Conductor 使用率的上升，我们发现了一些有趣的用例，里边的业务流程是基于其他流程中的事件和状态变化而启动的。

传统上，应用是这样实现类似用例的：使用公共 / 子系统并随应用状态变化发布事件，并订阅感兴趣的事件以发起行为。

我们寻求具有以下特征的解决方案：

• 工作流之间松散耦合
• 能基于其他流程中的状态变化启动流程
• 为通过 SQS/SNS 生产或消费事件的外部系统提供集成

现在说说工作流的反转控制。这个思想使流程的启动、流程中任务的完成等等绑定到流程行为上成为了可能。

为进一步说明这一点，让我们看两个在文件提取之后触发“质量控制流程”的解决方案。一旦文件提取完毕，就会触发由不同应用所有的多个流程。其中一个是文件验证流程（质量控制流程），它被触发去识别文件的类型（音频、视频等等），执行适当的检查，然后对这个文件开始编码。

如果使用 Conductor，可以用两种方式来实现这一点，这两种方案都能很好地完成，但有轻微的差异。

将其他流程作为子流程来启动

(点击放大图像)

假定，子流程是任务，他们的输出可以被质量控制流程之后的文件提取流程中的其他任务消费，换句话说，就是质量控制流程的输出。当业务流程间为紧依赖松耦合时，子流程是很有用的。

在事件上启动一个流程

(点击放大图像)

在以上方法中，文件提取流程产生一个“提取完成”事件。这个事件可以被一个或多个事件处理程序消费去执行行为，包括启动一个流程。

• 文件提取流程没有任何直接的流程依赖
• 流程基于事件触发，从而促成松耦合
• 质量控制任务流程的输出不影响文件提取流程
• 多个流程或行为可被每个事件执行

事件任务

我们引入一个新的任务类型，称为事件。在流程定义中可添加事件任务。执行时，它会按照规范产生一个“sink”事件。一个 sink 即为一个系统（比如 SQS、Conductor 或其他支持系统）事件。sink 遵循一个可插拔的架构，可以通过实现所需的接口以及在 Conductor server JVM 中的类路径中安装可用组件来添加 JMS、Kafka 之类的。

下面是 Conductor 的任务输入输出模型，对事件任务的输入是有计划的，作为载荷来发送。sink 可以是 Conductor，也可以是 SQS 之类的外部系统。这个支持多种类型 sink 的架构是基于插件的，像 JMS 或 Kafka 这样的新类型可以通过实现所需的接口以及在 Conductor server JVM 中的类路径中安装可用组件来予以添加。

下图是一个事件任务示例，它发布一个以名称标识的事件到 SQS 队列。

 
{
   "name": "example_event",
   "taskReferenceName": "event0",
   "input": {
       "filename": "${workflow.input.filename}",
       "location": "${anothertask.output.location}"
   },
   "type": "EVENT",
   "sink": "sqs:sqs_queue_name"
}

事件处理程序

事件处理程序即监听，特定事件到达时予以执行。处理程序监听多种事件来源，包括 Conductor 和 SQS/SNS，使用户可以通过 API 插入其他来源。对于每个事件类型，Conductor 支持多种事件处理程序。事件处理程序由 Conductor 使用事件 API 终端进行管理。对于单个事件来源来说，可以具有多个事件处理程序。

(点击放大图像)

事件条件

条件即载荷之上的一段 JavaScript 表达式，对于事件处理程序来说，其值必须判定为“真”才会执行相应行为。这个条件行为就像过滤器一样，在匹配条件的事件子集上有选择地采取行为。条件过滤器是可选的，未指定时该来源的所有事件都会被处理。

事件行为

每个事件处理程序都有一个或多个与其相关的行为。当与该事件相关的条件判定为“真”时，这些行为就会被执行。支持的事件是：

• 启动一个流程
• 完成一个任务：成功完成或以失败告终

举个例子

下面是事件处理程序示例，它监听了一个 SQS 队列，根据条件去启动相应流程

 
{
   "name": "conductor_event_test",
   "event": "sqs:example_sqs_queue_name",
   "condition": "$.file_type == ‘image’",
   "actions": [
       {
           "action": "start_workflow",
           "start_workflow": {
               "name": "workflow_name",
               "input": {
                   "file_name": "${filename}",
                   "file_type": "${file_type}"
               }
           }
       }
   ],
   "active": true
}

Conductor 界面

所有已注册的事件处理程序都可以在界面中予以检查。可以通过 REST API 或 Swagger API 界面来进行更新。

(点击放大图像)

JSON 数据转换

Conductor 现在支持基于 JSONPath 的数据转换。在输入配置中使用 JSON Path 就能让任务进行复杂的数据转换了，这样就减少了编写另行进行数据转换的一性次任务。

https://netflix.github.io/conductor/metadata/#wiring-inputs-and-outputs

推进计划

虽然忙了一个季度，但我们仍然没有完成。展望第二个季度，我们的关注点是让开发人员更容易测试流程，以及支持把任务执行时的上下文记录到日志中，以帮助理解和检修跨多个工作者和应用的流程。

如果你乐于接受构建分布式系统的挑战，并有兴趣构建 Netflix 工作室生态系统和大规模的内容管道，可以了解一下我们的职位空缺。

本文首发于 Netflix ，经网站授权由 InfoQ 翻译为中文版分享给中文读者。