php二进制流转图片技巧_开源微做事编排框架Netflix Conductor

文章目录 [+]

本文紧张先容netflix conductor的基本观点和紧张运行机制。

一简介

netflix conductor是基于JAVA措辞编写的开源流程引擎，用于架构基于微做事的流程。
它具备如下特性：

php二进制流转图片技巧_开源微做事编排框架Netflix Conductor

许可创建繁芜的业务流程，流程中每个独立的任务都是由一个微做事所实现。
基于JSON DSL 创建事情流，对任务的实行进行编排。
事情流在实行的过程中可见、可追溯。
供应停息、规复、重启等多种掌握模型。
供应一种大略的办法来最大限度重用微做事。
拥有扩展到百万流程并发运行的做事能力。
通过行列步队做事实现客户端与做事真个分离。
支持 HTTP 或其他RPC协议进行数据传送二基本观点

1 Task

（图片来自网络侵删）

Task是最小实行单元，承载了一段实行逻辑，如发送HTTP要求等。

System Task：被conductor做事实行，这些任务的实行与引擎在同一个JVM中。
Worker Task：被worker做事实行，实行与引擎隔离开，worker通过行列步队获取任务后，实行并更新结果状态到引擎。
Worker的实现是跨措辞的，其利用Http协议与Server通信。

conductor供应了多少内置SystemTask:

功能性Task：HTTP：发送http要求JSON_JQ_TRANSFORM：jq命令实行，一样平常用户json的转换，详细可见jq官方文档KAFKA_PUBLISH: 发布kafka流程掌握Task：SWITCH（原Decision）：条件判断分支，类似于代码中的switch caseFORK：启动并行分支，用于调度并行任务JOIN：汇总并行分支，用于汇总并行任务DO_WHILE：循环，类似于代码中的do whileWAIT：一贯在运行中，直到外部韶光触发更新节点状态，可用于等待外部操作SUB_WORKFLOW：子流程，实行其他的流程TERMINATE：结束流程，以指定输出提前结束流程，可以与SWITCH节点合营利用，类似代码中的提前return语句自定义Task：对付System Task，Conductor供应了WorkflowSystemTask 抽象类，可以自定义扩展实现。
对付Worker Task，可以实现conductor的client Worker接口实现实行逻辑。

2 Workflow

Workflow由一系列须要实行的Task组成，conductor采取json来描述Task的流转关系。
除基本的顺序流程外，借助内置的SWITCH、FORK、JOIN、DO_WIHLE、TERMINATE任务，还能实现分支、并行、循环、提前结束等流程掌握。

3 Input&Output

Task的输入是一种映射，其作为事情流实例化的一部分或某些其他Task的输出。
许可将来自事情流或其他Task的输入/输出作为随后实行的Task的输入。

Task有自己的输入和输出，输入输出都是jsonobject类型。
Task可以引用其他Task的输入输出，利用${taskxxx.output}的办法引用。
引用语法为json-path，除最根本的${taskxxx.output}的值解析办法外，还支持其他繁芜操作，如过滤等，详细见json-path语法。
启动Workflow时可以传入流程的输入数据，Task可以通过${workflow.input}的办法引用。

Task实现原子操作的处理以及流程掌握操作，Workflow定义描述Task的流转关系，Task引用Workflow或者其它Task的输入输出。
通过这些机制，conductor实现了JSON DSL对流程的描述。

三整体架构

紧张分为几个部分：

Orchestrator: 卖力流程的流转调度事情；Management/Execution Service: 供应流程、任务的管理更新等操作；TaskQueues: 任务行列步队，Orchestrator解析出来的待实行Task会放到行列步队中；Worker: 任务实行worker，从TaskQueues中获取任务，通过Execution Service更新任务状态与结果数据；Database: 元数据&运行时数据库，用于保存运行时的Workflow、Task等状态信息，以及流程任务定义的等原信息；Index: 索引数据库，用于存储实行历史；四运行模型

1 Task状态转移

SCHEDULED：待调度，task放到行列步队中还没有被poll出来实行时的状态IN_PROGRESS：实行中，被poll出来实行但还没有完成时的状态COMPLETED：实行完成FAILED：实行失落败CANCELLED：被中止时为此状态，一样平常涌如今两种情形：手动中止流程时，正在运行中的task会被置为此状态；多个fork分支，当某个分支的task失落败时，其它分支中正在运行的task会被置为此状态；

2 任务行列步队

任务的实行（同步的系统任务除外）都会先添加到任务行列步队中，是范例的生产者消费者模式。

任务行列步队，是一个带有延迟、优先级功能的行列步队；每种类型的Task是一个单独的行列步队，此外，如果配置了domain、isolationGroup，还会拆分成多个行列步队实现实行隔离；decider service是生产者，其根据流程配置与当前实行情形，解析出可实行的task后，添加到行列步队；任务实行器(SystemTaskWorker、Worker)是消费者，其长轮询对应的行列步队，从行列步队中获取任务实行；

行列步队接口可插拔，conductor供应了Dynomite 、MySQL、PostgreSQL的实现。

3 核心功能实现机制

conductor调度的核心是decider service，其根据当前流程运行的状态，解析出将要实行的任务列表，将任务入队交给worker实行。

decide紧张流程简化如下，详细代码见WorkflowExecutor.java的decide方法：

个中，调度任务处理流程简化如下，详细代码见WorkflowExecutor.java的scheduleTask方法：

decide的触发机遇

最紧张的触发机遇：

新启动实行时，会触发decide操作系统任务实行完成时，会触发decide操作Workder任务通过ExecutionService更新任务状态时，会触发decide操作

流程掌握节点的实现机制

1）Task & TaskMapper

对付每一个Task来说，都有Task和TaskMapper两部分:

Task：任务的实行逻辑代码，它的浸染是Task的实行TaskMapper：任务的映射逻辑代码，它通过Task的定义配置、当前实例的实行状态等信息，返回实际须要实行的Task列表

对付一样平常的任务来说，TaskMapper返回的是便是Task本身，补充一些实行实例的状态信息。
但是对付掌握节点来说，会有不同的逻辑。

2）条件分支(SWITCH)的实现机制

SWITCH用于根据条件判断，实行不同的分支。

实际上，该节点的Task不做任何操作，TaskMapper根据分支条件，判断出要走的分之后，返回对应分支的第一个Task。

SwitchTaskMapper.java getMappedTasks方法关键代码：

// 待调度的Task list，终极返回结果List<Task> tasksToBeScheduled = new LinkedList<>();// evalResult是分支条件变量的值(case)// decisionCases是一个Map构造，key为分支的case值，value为对应分支的任务定义list（分支内的任务定义会有多个）// 根据分支变量的实际值，获取对应分支的任务定义listList<WorkflowTask> selectedTasks = taskToSchedule.getDecisionCases().get(evalResult);// default的逻辑：如果获取不到对应的分支或者分支为空，则用默认的分支if (selectedTasks == null || selectedTasks.isEmpty()) { selectedTasks = taskToSchedule.getDefaultCase();}if (selectedTasks != null && !selectedTasks.isEmpty()) { // 获取分支的第一个(下标0)task，返回给decider service去做调度（decider会把任务添加到行列步队里，交给worker去实行） WorkflowTask selectedTask = selectedTasks.get(0); // 调用了deciderService的getTasksToBeScheduled方法，此方法里又获取到TaskMapper调用了getMappedTasks。这里采取了递归调用的办法，解析嵌套的Task List<Task> caseTasks = taskMapperContext.getDeciderService() .getTasksToBeScheduled(workflowInstance, selectedTask, retryCount, taskMapperContext.getRetryTaskId()); tasksToBeScheduled.addAll(caseTasks); switchTask.getInputData().put("hasChildren", "true");}return tasksToBeScheduled;

3）并行(FORK)的实现机制

FORK用于开启多个并行分支。

实际上，该节点的Task不做任何操作，TaskMapper返回所有并行分支的第一个Task。
ForkJoinTaskMapper.java getMappedTasks关键代码：

// 待调度的Task list，终极返回结果List<Task> tasksToBeScheduled = new LinkedList<>();// 配置中的所有fork分支List<List<WorkflowTask>> forkTasks = taskToSchedule.getForkTasks();for (List<WorkflowTask> wfts : forkTasks) { // 每个分支取第一个Task WorkflowTask wft = wfts.get(0); // 调用了deciderService的getTasksToBeScheduled方法，此方法里又获取到TaskMapper调用了getMappedTasks。这里采取了递归调用的办法，解析嵌套的Task List<Task> tasks2 = taskMapperContext.getDeciderService() .getTasksToBeScheduled(workflowInstance, wft, retryCount); tasksToBeScheduled.addAll(tasks2);}return tasksToBeScheduled;

总的来说，分支(SWITCH)、并行(FORK)节点本身没有实行逻辑，其通过TaskMapper返回到实际要实行的Task，然后交给Decider Service处理。

重试的实现机制

重试和其延迟韶光设置，都是借助任务行列步队的功能实现的。

重试：将任务重新添加到任务行列步队

重试的延迟韶光：添加到任务行列步队时设置延迟韶光，延迟韶光过后，任务才能在行列步队中被poll出来实行

五完全性保障机制

由于调度过程中可能会涌现因机器重启、网络非常、JVM崩溃等偶发情形，这些会导致的decide过程意外终止，流程实行不完全，展现出如流程一贯运行中（实际已经没有在调度），或者其它状态缺点等非常征象。

1 WorkflowReconciler

针对这种情形，conductor有一个WorkflowReconciler，会定期考试测验decide所有正在运行中的流程，修复流程实行的同等性。
此外，它还有一个浸染是校验流程超时时间。

2 decideQueue

那么WorkflowReconciler是如何获取到当前运行中的流程呢，答案是decideQueue。
decideQueue和任务行列步队相同，也是一个具有延迟功能的行列步队，其存放的是正在实行中的流程的实例id。
在任务开始实行时（包括新启动实行、重试实行、规复实行、重跑实行等），会将实例id push到decideQueue中；在实行结束（成功、失落败）时，会从decideQueue中删除实例id。

3 ExecutionLockService

WorkflowReconciler会定期考试测验decide所有正在运行中的流程用于超时判断、掩护流程同等性。
但是流程本身正常实行也会触发decide，如果同一个实行同时触发两个decide，可能会导致状态混乱，实行卡住等问题。

conductor采取了锁来办理这个问题，其供应了单机LocalOnlyLock（基于旗子暗记量实现）、redis分布式锁（基于redission实现）、zookeeper分布式锁三种实现。

decide方法中最开始会考试测验获取锁，如果获取失落败则直接返回。
通过锁来保障不会对同一个流程实例并发实行decide。

if (!executionLockService.acquireLock(workflowId)) { return false;}

由于锁是可配置的，可能会导致一个误区：单台机器的话不用配置锁。
实在单机也是须要配置锁的，由于WorkflowReconciler和流程正常实行会产生冲突，可能会导致偶发的流程状态混乱问题。

参考：Github: https://github.com/Netflix/conductor官方文档：https://netflix.github.io/conductor/WorkflowReconciler：https://github.com/Netflix/conductor/blob/main/core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowReconciler.javaWorkflowSystemTask：https://github.com/Netflix/conductor/blob/main/core/src/main/java/com/netflix/conductor/core/execution/tasks/WorkflowSystemTask.java?spm=ata.21736010.0.0.2b501a3cYnrSfT&file=WorkflowSystemTask.java

作者 | 夜阳

原文链接：https://developer.aliyun.com/article/818136?utm_content=g_1000311143

本文为阿里云原创内容，未经许可不得转载。