conductor 系统任务

动态任务：

参数：

dynamicTaskNameParam：来自任务输入的参数的名称，其值用于调度任务。 例如 如果参数的值为ABC，则调度的下一个任务类型为“ABC”。

Example

{ "name": "user_task", "taskReferenceName": "t1", "inputParameters": { "files": "${workflow.input.files}", "taskToExecute": "${workflow.input.user_supplied_task}" }, "type": "DYNAMIC", "dynamicTaskNameParam": "taskToExecute" } 
如果以user_supplied_task的值(假设值为user_task_2)来启动工作流，那么Conductor将在调度此动态任务时调度user_task_2。

决策任务：

决策任务与编程语言中的case ... switch语句相似。 该任务需要3个参数：
caseValueParam： 其值将用作开关的任务输入中的参数名称。
decisionCases： 映射其中key是caseValueParam的可能值，值是要执行的任务列表。
defaultCase：在决策案例中找不到匹配值时执行的任务列表（默认条件）

Example

{ "name": "decide_task", "taskReferenceName": "decide1", "inputParameters": { "case_value_param": "${workflow.input.movieType}" }, "type": "DECISION", "caseValueParam": "case_value_param", "decisionCases": { "Show": [ { "name": "setup_episodes", "taskReferenceName": "se1", "inputParameters": { "movieId": "${workflow.input.movieId}" }, "type": "SIMPLE" }, { "name": "generate_episode_artwork", "taskReferenceName": "ga", "inputParameters": { "movieId": "${workflow.input.movieId}" }, "type": "SIMPLE" } ], "Movie": [ { "name": "setup_movie", "taskReferenceName": "sm", "inputParameters": { "movieId": "${workflow.input.movieId}" }, "type": "SIMPLE" }, { "name": "generate_movie_artwork", "taskReferenceName": "gma", "inputParameters": { "movieId": "${workflow.input.movieId}" }, "type": "SIMPLE" } ] } }

Fork任务：

fork用于调度并行的任务。
参数：
forkTasks：任务列表的列表。 每个子列表被安排并行执行。 然而，子列表中的任务是以串行方式安排的。

Example

{ "forkTasks": [ [ { "name": "task11", "taskReferenceName": "t11" }, { "name": "task12", "taskReferenceName": "t12" } ], [ { "name": "task21", "taskReferenceName": "t21" }, { "name": "task22", "taskReferenceName": "t22" } ] ] } 
当执行时，task11和task21被调度为同时执行。


动态fork任务：
动态fork与FORK_JOIN任务相同。 除了在运行时使用任务的输入提供要分配的任务的列表。 当fork的任务数量不固定并且根据输入而变化时很有用。
dynamicForkTasksParam：包含并行执行的工作流任务配置列表的参数名称
dynamicForkTasksInputParamName： 参数的名字，此参数的值是一个映射，映射的key是fork任务的别名，value是fork任务的input

Example

{ "dynamicTasks": "${taskA.output.dynamicTasksJSON}", "dynamicTasksInput": "${taskA.output.dynamicTasksInputJSON}", "type": "FORK_JOIN_DYNAMIC", "dynamicForkTasksParam": "dynamicTasks", "dynamicForkTasksInputParamName": "dynamicTasksInput" } 

假设taskA的输出为：

{ "dynamicTasksInputJSON": { "forkedTask1": { "width": 100, "height": 100, "params": { "recipe": "jpg" } }, "forkedTask2": { "width": 200, "height": 200, "params": { "recipe": "jpg" } } }, "dynamicTasksJSON": [ { "name": "encode_task", "taskReferenceName": "forkedTask1", "type": "SIMPLE" }, { "name": "encode_task", "taskReferenceName": "forkedTask2", "type": "SIMPLE" } ] }
执行时，动态fork任务将调度两个并行任务，类型为“encode_task”，引用名称为“forkedTask1”和“forkedTask2”，由_ dynamicTasksInputJSON_指定

join任务：

join任务用于等待fork任务产生的一个或多个任务的完成。
参数：
joinOn：任务参考名称列表，其中JOIN将等待完成。

Example

{ "joinOn": ["taskRef1", "taskRef3"] } 

Join Task Output

fork任务的输出将是一个JSON对象，其中key是任务别名，值是fork任务的输出。

子工作流任务：

Sub Workflow任务允许在另一个工作流中嵌套工作流。

参数：

subWorkflowParam：任务参考名称列表，其中JOIN将等待完成。

Example

{ "name": "sub_workflow_task", "taskReferenceName": "sub1", "inputParameters": { "requestId": "${workflow.input.requestId}", "file": "${encode.output.location}" }, "type": "SUB_WORKFLOW", "subWorkflowParam": { "name": "deployment_workflow", "version": 1 } } 
执行时，将使用两个输入参数requestId和file执行deployment_workflow。 完成生成的工作流程后，任务将被标记为已完成。 如果子工作流程终止或失败，则将任务标记为失败，并重新配置。


等待任务：

等待任务被实现为保持IN_PROGRESS状态的门，除非由外部触发器标记为COMPLETED或FAILED。 要使用等待任务，请将任务类型设置为WAIT

参数：无
外部触发等待任务：
任务资源端点可用于将任务的状态更新为终止状态。
Contrib模块提供SQS集成，其中外部系统可以将消息放置在服务器侦听的预配置队列中。 消息到达时，它们被标记为COMPLETED或FAILED。

SQS队列
可以使用以下API检索服务器用于更新任务状态的SQS队列：

GET /queue

当更新任务的状态时，消息需要符合以下规范：

     消息必须是有效的JSON字符串。
     消息JSON应包含一个名为externalId的密钥，其值为包含以下密钥的JSONized字符串：
         workflowId：工作流的ID
         taskRefName：应该更新的任务引用名称。
     每个队列代表一个特定的任务状态，相应地标记任务。 例如 消息进入COMPLETED队列将任务状态标记为COMPLETED。
     使用消息更新任务的输出。








示例SQS有效负载：

{ "some_key": "valuex", "externalId": "{ 
     \"taskRefName\":\"TASK_REFERENCE_NAME\",\"workflowId\":\"WORKFLOW_ID\"}" }


HTTP任务：


HTTP任务用于通过HTTP调用另一个微服务器。
该任务期望一个名为http_request的输入参数作为任务输入的一部分，具有以下详细信息：
uri：服务的URI。 使用vipAddress时可以是部分或包含服务器地址。
method： HTTP method. One of the GET, PUT, POST, DELETE, OPTIONS, HEAD
accept： Accept header as required by server.
contentType：Content Type - 支持的类型是 text/plain, text/html and, application/json
headers：要与请求一起发送的其他http标头的map。
body：Request body
vipAddress：使用基于发现的服务URL时。

HTTP 任务输出：
response： 包含response的JSON主体（如果存在）
headers： Response Headers
statusCode： Integer status code

Example

Task Input payload using vipAddress

{ "http_request": { "vipAddress": "examplevip-prod", "uri": "/", "method": "GET", "accept": "text/plain" } } 

Task Input using an absolute URL

{ "http_request": { "uri": "http://example.com/", "method": "GET", "accept": "text/plain" } } 
如果请求无法完成或远程服务器返回不成功的状态代码，任务将被标记为FAILED。


Event任务：

事件任务提供将事件（消息）发布到conductor或外部事件系统（如SQS）的能力。 事件任务对于为工作流和任务创建基于事件的依赖非常有用。
参数：
sink：生产的事件的合格名称。 例如 conductor或sqs：sqs_queue_name

Example

{ "sink": 'sqs:example_sqs_queue_name' } 

当使用conductor作为接收器生成事件时，事件名称遵循以下结构：conductor：<workflow_name>：<task_reference_name>
对于SQS，使用队列的名称，而不是URI。 conductor根据名称查找URI。

Supported Sinks

• Conductor
• SQS

事件任务输入：

给予事件任务的输入作为有效载荷提供给发布的消息。 例如 如果将消息放入SQS队列（sink is sqs）中，则消息有效内容将作为任务的输入。

时间任务输出：

event_produced生成的事件的名称。