流程 (Flow)

🎬 Automate with flows · 39초 · 2026.08 기준 화면

消息结构 · 各触发器载荷示例 · 节点目录
触发器 · 过滤 · 转换
动作 — 集成/存储 · 资产事件发布 · 域 CRUD
Edge · 外部集成 · 流程控制
节点选项详细参考 · 编写脚本节点 · JS 执行环境规范 · 内存安全编写模式 · 移动推送通知渠道 · 外部认证令牌自动刷新

示例·模式

示例流程 (17种) · 运维模式 (Recipes 10种) · 应用示例

运维

常规工作流 · 导入/导出 · 错误处理 · 运维诊断 · 权限

生产运维

常见问题 (FAQ) · 运维最佳实践 · 安全·敏感信息处理
流程指标与报警 · 消息处理语义与背压 · 从执行历史中能看到什么
集群·高可用(HA)动作 · 端到端追踪 · 图形模式目录 · 流程测试最佳实践
性能极限与调优 · 审计·历史追踪 · 新建流程部署清单 · 部署策略 (Canary/Blue-Green/A·B) · 紧急应对流程
节点快速设置参考 · 流程 REST API · Webhook 触发器 · OPC/PLC 工业集成模式
外部系统集成 Cookbook · 数据转换 Cookbook · 可复用脚本集合

故障排除

分步调试指南 · 常见问题

其他

术语词典 · 版本说明 · 相关画面

概述

流程是以可视化图形方式定义与外部系统(MES/ERP/SCADA 等)之间的数据集成、域对象(标签·资产·作业指示·作业者等)自动生成·更新的自动化菜单。操作员无需编码即可直接设计·部署·运维自动化场景。

在画布上以拖放方式放置约 100 余种节点,并通过连线连接,构成数据处理流水线。

路径: 左侧菜单 > Automation > 流程

学习路线图 — 按角色推荐入门顺序

本手册是超过 4,000 行的综合指南。建议从符合自身角色和目的的章节开始阅读,更为高效。

👶 初次使用者(1小时内创建第一个流程)

核心概念 — 5分钟
初次开始 — 5分钟
端到端教程 — 30分钟 (跟随步骤1~10)
画面构成 + 编辑画面 — 10分钟
跟随示例流程1·2·3 — 10分钟

→ 完成首个流程部署。之后所需节点可在节点目录中查找。

🧑‍🏭 现场操作员(编写自动化场景)

各触发器载荷示例 — 理解实际数据结构
节点选项详细参考 — 熟悉常用节点选项
数据转换 Cookbook — 复制常见转换模式使用
图形模式目录 — 选择连线模式
示例流程 (17种) — 参考各场景成品

🛠 系统管理员(运维·调优·故障应对)

流程指标与报警 — 应关注哪些指标
从执行历史中能看到什么 — 解读诊断日志
集群·HA 动作 — 理解多节点环境
端到端追踪 — 追踪问题消息
性能极限与调优 + 紧急应对流程

🔌 开发者·集成工程师(外部系统集成)

外部系统集成 Cookbook — Slack/Teams/Jira/SAP 示例
流程 REST API — 以程序操作流程
Webhook 触发器 — 从外部触发流程
OPC/PLC 工业集成模式 — 工业现场场景
JS 执行环境规范 + 可复用脚本集合

🔐 安全负责人(审计·认证)

安全·敏感信息处理 — 凭据保管
外部认证令牌自动刷新 — OAuth2 令牌运维
权限 — 各角色可执行的操作
审计·历史追踪 — 保留变更/执行历史

📚 快速参考(已熟悉的用户)

查找的信息	章节
节点 ID 与一行说明	节点目录
节点选项默认值	节点选项详细参考
消息 JSON 示例	各触发器载荷示例
可直接使用的脚本	可复用脚本集合 · 数据转换 Cookbook
API 调用 curl	流程 REST API
执行历史解读	从执行历史中能看到什么
故障排除指南	分步调试指南 · 常见问题
快速选项表	节点快速设置参考

所有章节链接均为同一文档内的锚点。使用 Ctrl+F 进行关键词搜索也很有效。

核心概念

术语	说明
流程 (Flow)	由节点和连线(关系)集合构成的一个自动化工作流。是具有入口点的有向图,通过部署/解除开关控制是否激活
节点 (Node)	接收消息进行处理后传递给下一个节点的单位。分为7个类别(触发器·过滤·转换·动作·外部集成·流程控制·Edge)
关系 (Relation)	从节点输出到下一节点的连线标签。节点直接指定 `SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` 等标签进行分支。所有标签均标准化为大写
消息 (Message)	在流程内部流动的载荷。包含 `type`(分类)、`originator`(主体实体)、`data`(本体)、`metadata`(上下文)
触发器 (Trigger)	作为流程起点的节点。分为域事件(标签点位·报警·资产事件等)、外部接入(webhook·MQTT·外部数据库)、时间(计划)三种

初次开始

通过以下4个步骤,可以最快熟悉流程功能。

在列表画面点击 새 플로우 按钮创建空流程(仅输入名称和描述)。
在编辑画面中依次将左侧调色板的触发器节点(例:flow_on_tag_alarm)→过滤→动作(例:flow_send_email)拖放放置,并在节点之间以连线连接。
点击各节点,在右侧检查器中输入选项后,点击右上角保存 → 测试运行,检验一次动作。
打开顶部部署开关后,每当触发器事件进入时流程会自动执行,可在实时调试面板和执行历史画面确认结果。

自动化场景的连线模式请参考本文档的示例流程和应用示例。从头到尾创建实际流程的分步教程请参考端到端教程。

端到端教程

从头到尾创建一个在运行环境中实际可用的自动化流程的分步教程。场景:当电机温度超过 80°C 时,自动发布紧急维护作业指示并通过邮件通知负责人。

步骤 1 — 创建新流程

在左侧菜单点击 Automation > 流程 → 打开列表画面
点击右上角新建流程按钮
输入以下信息后点击确认

项目	输入值
名称	`모터 과열 자동 정비 발행`
描述	`[자동화] 모터 자산의 온도 80°C 초과 시 긴급 정비 작업지시 + 이메일. 담당: ops@example.com`

流程创建后,会自动打开空画布的编辑画面。

步骤 2 — 放置触发器节点

从资产遥测事件接收电机温度数据。

展开左侧调色板的触发器 (Trigger) 类别
将 flow_on_tag_point 节点拖放到画布上
点击节点 → 在右侧检查器中输入以下选项

选项	值
显示名	`태그 포인트 인입`
`tag_id_pattern`	`MOTOR-*.TEMP`

通过 tag_id_pattern 只让电机温度标签通过,可大幅减少后续处理量。其他标签将被标记为 SKIPPED,不计入计数器。

步骤 3 — 阈值过滤

添加脚本过滤器,只让超过80°C的消息通过。

从过滤 (Filter) 类别拖出 flow_script_filter
从触发器节点的输出端口连线到新过滤节点的输入端口
在检查器中输入以下内容

选项	值
显示名	`임계값 필터 (80°C 초과)`
`language`	`JS`
`script`	`data.value > 80`

步骤 4 — 域变更 (标签 → 资产)

由于报警·作业指示以资产单位发布更为自然,需将消息的 originator 从标签转换为上层资产。

从转换 (Transform) 类别拖出 flow_change_originator
从过滤节点的 TRUE 输出连线
检查器:

选项	值
显示名	`Tag → Asset 변경`
`entity_type`	`Asset`
`id_field`	`metadata.asset_id`

metadata.asset_id 会在标签点位接入时自动填充。若消息中没有,可通过脚本提取标签ID的前缀部分(例:MOTOR-001.TEMP → MOTOR-001)。

步骤 5 — 发布作业指示

自动创建紧急维护作业指示。

从动作 (Action) — 域 CRUD 类别拖出 flow_create_work_order
从转换节点的 SUCCESS 输出连线
检查器:

选项	值
显示名	`긴급 정비 작업지시 생성`
`asset_id_field`	`originator.id`
`order_name_field`	(静态值) `긴급 점검 — 모터 과열`
`master_id_field`	(可选) `data.master_id` (若无则自动生成)
`description_field`	(静态值) `자동 발행: 임계 온도 초과로 긴급 점검 필요`
`default_priority`	`HIGH`

步骤 6 — 邮件通知(成功分支)

作业指示发布成功时通知负责人。

从外部集成 (External) 类别拖出 flow_send_email
从作业指示节点的 SUCCESS 输出连线
检查器:

选项	值
显示名	`정비 담당자 이메일`
`to`	`ops@example.com`
`subject_template`	`[과열 정비] ${originator.id} 작업지시 ${data.work_order_id}`
`body_template`	`자산 ${originator.id} 의 온도가 ${data.value}°C 로 상승하여 자동으로 긴급 정비가 발행되었습니다.\n작업지시 ID: ${data.work_order_id}`

步骤 7 — 失败分支处理

作业指示发布本身也可能失败(例如资产已消失或权限问题)。立即向操作员发送推送通知。

从外部集成拖出 flow_send_push
从作业指示节点的 FAILURE 输出连线
检查器:

选项	值
`title`	`자동화 실패`
`body_template`	`${originator.id} 정비 자동 발행 실패: ${data.error}`

步骤 8 — 保存与测试运行

点击右上角保存按钮。系统会自动加载快照,以便后续回滚。
点击右上角测试运行按钮 → 在 JSON 编辑器中输入以下内容 → 发布

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.TEMP" },
  "data":     { "value": 92.5 },
  "metadata": { "ts": 1746247200000, "tag_id": "MOTOR-001.TEMP", "asset_id": "MOTOR-001" }
}

对话框下方显示 ✓ JSON OK (type=POST_TELEMETRY) → 点击发布。

在实时调试面板确认以下内容:
- 触发器节点点亮绿色 → 消息通过
- 过滤节点:因为 92.5 > 80,故走 TRUE 分支
- 转换节点:originator 变为 Asset/MOTOR-001
- 作业指示节点:SUCCESS,自动赋予 data.work_order_id
- 邮件节点:发送尝试

步骤 9 — 验证与部署

在作业指示画面确认是否登记了新的作业指示
确认邮件是否正常送达(测试环境的邮箱)
用低于80°C的消息再测试一次(应被过滤器拦截):

{ "type": "POST_TELEMETRY", "originator": {"entity_type":"Tag","id":"MOTOR-001.TEMP"},
  "data": {"value": 70}, "metadata": {"asset_id":"MOTOR-001"} }

过滤节点分支为 FALSE,后续节点为灰色 — 正常。

验证所有分支后,通过重置全部计数来重置统计窗口
打开顶部部署开关

现在在实际运行环境中,一旦电机温度超过80°C,就会自动发布维护作业指示并向负责人发送邮件。

步骤 10 — 运维监控

部署后建议观察5~10分钟以下内容以确保安全。

位置	确认项
列表画面	该流程行的执行次数是否在正常范围内增长(而非暴增)
实时调试面板	是否有节点失败
执行历史画面	若有失败消息,查明 `NODE_ERROR` 的原因
收件邮箱	是否以非预期频率发出通知

下一步

若要进一步扩展此流程:

添加重试连线 — 邮件/推送发送失败时进行退避重试(参考示例8)
自动调整波段 — 以6小时平均值 ± 3σ 自动校正阈值(参考示例6)
累积停机时间 — 将过热历史累积到资产 attribute(参考示例14)
隔离质量产线 — 若过热连续发生则自动停止该产线(参考示例15)

画面构成

流程由以下三个画面构成。

画面	用途
列表	已注册流程一览·搜索·批量部署·导入
编辑	以可视化画布放置·连接·设置节点
执行历史	按节点单位查询执行日志·时间线

列表画面

由顶部搜索·创建区域和流程一览表组成。

顶部工具

项目	说明
状态过滤	`전체` / `배포` / `해제`
名称·描述搜索	通过关键词筛选流程
新建流程	创建空流程(输入名称·描述)
导入	上传通过导出获得的 JSON 恢复流程
全部重新部署	一次性重新加载所有已激活流程
刷新	重新加载列表

一览表

列表表格以每页25行分页,每行同时显示处理趋势迷你图和错误率环形图。翻页时图表状态保持不变。

列	说明
选择	用于批量部署·解除的复选框
状态	`배포` / `해제` 徽章
流程 ID	`FLOW_NNNNN` 格式的序列 ID
名称 \| 描述	操作员指定的元信息
节点	包含的节点数量
触发器	触发器节点数量
执行次数	累积消息处理计数
处理时间	节点平均/最近处理时间
错误	累积错误计数
最终修改	图形最后保存时间
操作	`편집` / `삭제` 按钮

批量部署·解除

将所选流程一次性部署(激活)或解除(停用)。仅已部署的流程才会接收触发器事件。

全部重新部署

顶部全部重新部署按钮会重新加载所有已激活的流程。请在以下情况使用。

从外部批量导入图形后
想重新注册具有自身调度器的触发器(计划·外部 MQTT 订阅·外部数据库轮询)时
怀疑运行中缓存一致性存在问题时

编辑画面 (可视化画布)

从画布左侧调色板拖出节点放置后,点击拖动节点的输出端口与下一节点连接。

顶部工具栏

按钮	动作
名称·描述	编辑流程元信息
部署/解除	立即激活/停用当前流程
导出	将整个图形下载为 JSON 文件
测试运行	注入任意 JSON 消息执行一次并确认结果(参考下方测试运行使用方法)
保存	将当前图形保存到服务器。保存时自动加载快照,便于后续回滚

若尝试保存没有触发器的图形,会显示"仅可通过手动测试运行启动"的警告。可故意省去触发器,将其用作仅手动执行的流程。

测试运行使用方法

点击 테스트 실행 按钮会打开 JSON 编辑对话框。操作员可直接编写消息发布一次,无需等待触发器事件即可验证图形动作。

项目	说明
编辑器	带行号·语法高亮的 JSON 编辑器。可自由编写消息本体
验证显示	若 `type` 必填字段存在则显示 `✓ JSON OK (type=X)`,缺失则显示 `⚠ type 필수`
发布	通过 `발행` 按钮将消息注入分发器。结果可在实时调试面板和执行历史中查看

默认模板示例

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data":     { "speed": 1500, "temp": 75.3 },
  "metadata": { "ts": 1746247200000, "site_id": "SITE-01" }
}

若存在未保存的更改时点击测试运行,会提示"服务器将执行已保存的版本"。若要验证更改,请先保存。

左侧 — 节点调色板

可按类别折叠展开,并可通过搜索输入即时过滤。

类别	颜色	节点数
触发器 (Trigger)	灰色	22种
过滤 (Filter)	蓝色	6种
转换 (Transform)	绿色	8种
动作 (Action) — 集成·存储·资产发布·命令调用	橙色	7种
动作 (Action) — 域 CRUD	橙色	34种
外部集成 (External)	紫色	9种
流程控制 (Control)	灰色	7种
Edge (Edge)	青色	22种

中央 — 画布

工具	快捷键 / 操作	动作
放大/缩小	`Ctrl/⌘ +` / `Ctrl/⌘ -` · 鼠标滚轮	缩放画布
100%	`Ctrl/⌘ 0`	重置缩放
适应画面	`Ctrl/⌘ 1`	自动调整以显示所有节点
删除所选节点	`Del` / `Backspace`	删除所选节点/连线
画布平移 (panning)	在空白区域左键点击后拖动	在未点击节点的状态下按住空白画布背景并拖动,整个画布会随之移动

画布右下角显示小地图,点击小地图可立即跳转至该位置。

💡 panning 使用提示:在节点上拖动会移动节点 — 若要移动画布,请务必按住没有节点/连线的空白背景区域。大型流程中 panning 比小地图更快捷。

右侧 — 节点设置

在画布上点击节点后,右侧检查器会显示该节点的设置表单。输入字段会根据节点种类自动生成。

输入方式	说明
静态值	直接使用表单中输入的值
*`_field` 动态值**	从消息载荷路径(例:`data.tag_id`、`metadata.site_id`)提取值。若无值则回退为静态值

脚本节点(过滤·转换·switch)可直接在检查器中以代码编辑器编辑,也可扩展为单独对话框在大屏幕中编写。

代码编辑器字体采用可读性增强的等宽字体栈(优先 Cascadia Code · JetBrains Mono · Consolas · Menlo),韩文注释也能稳定对齐。

计数重置

节点设置面板右上角以图标形式显示两个重置按钮。将鼠标悬停在各按钮上会显示提示。

图标按钮	动作
🩹 (创可贴) — 重置错误计数	仅将该流程内所有节点的累积错误计数重置为0
🔄 (旋转箭头) — 重置全部计数	将处理·错误·处理时间全部 + 流程单位统计全部重置为0。在完成运维验证并想开始全新统计时使用

两个按钮均无需确认对话框即刻生效 — 仅重置统计数据,不影响节点动作或消息处理。

右侧 — 实时调试

实时调试面板显示在检查器下方。

项目	说明
刷新周期	2秒
级别颜色	左侧边框显示 INFO(蓝)/WARN(黄)/ERROR(红)
显示信息	节点显示名 · 处理时间(ms) · 消息预览
暂停	通过面板右上角开关暂停刷新
清空	仅在画面上清空累积的调试项

画布上节点右上角显示小灯。

颜色	含义
灰色	待机 — 未接收消息的状态
绿色	消息通过中
红色	处理中发生错误

节点右下角以 평균 X · 최근 Y 形式显示处理时间。

消息结构

在流程内部流动的消息由以下4个区域构成。

{
  "type": "POST_TELEMETRY",
  "originator": {
    "entity_type": "Asset",
    "id": "MOTOR-001"
  },
  "data": { "speed": 1500, "temp": 75.3 },
  "metadata": {
    "ts": 1746247200000,
    "site_id": "SITE-01",
    "shift": "DAY",
    "tag_id": "MOTOR-001.SPEED"
  }
}

区域	含义
`type`	消息分类。过滤节点的分支依据
`originator`	消息主体实体(针对哪个资产/标签/订单)
`data`	载荷本体
`metadata`	上下文(时刻·站点·班次·标签ID等)

消息类型

类型	发生入口点
`POST_TELEMETRY` / `TAG_POINT`	标签点位接入
`POST_ATTRIBUTES`	标签/资产元信息更新
`TAG_ALARM`	标签单位报警
`ENTITY_CREATED` / `UPDATED` / `DELETED`	实体生命周期事件
`ASSET_DATA` / `ASSET_EVENT` / `ASSET_ALARM` / `ASSET_COMMAND` / `ASSET_AGGREGATION` / `ASSET_CONTEXT`	资产域事件
`ASSET_HEALTH_STATUS` / `ASSET_CONNECTION_STATUS`	资产周期评估(健康/连接状态)
`OEE_EVENT` / `RAM_EVENT` / `EMS_EVENT`	ISO 分析结果事件
`OPC_STATUS` / `EDGE_STATUS`	OPC/Edge 设备状态
`DIAGNOSTIC` / `DOMAIN_CHANGED`	诊断·域变更
`ALARM`	报警发生
`WEBHOOK`	HTTP webhook 接收
`KAFKA_INBOUND` / `MQTT_INBOUND`	外部主题接收
`TIMER`	计划触发

各触发器载荷示例

编写脚本节点时需要准确了解可访问哪些字段。以下是各触发器生成的实际消息 JSON 示例。所有触发器消息通用包含 type、originator、data、metadata。

`flow_on_tag_point` — 标签点位接入

每当一个标签有值进入时触发。是最常见的触发器。

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.SPEED" },
  "data": {
    "value": 1500.7,
    "quality": "GOOD",
    "ts": 1746247200123
  },
  "metadata": {
    "tag_id":      "MOTOR-001.SPEED",
    "site_id":    "SITE-01",
    "area_id":    "AREA-A",
    "line_id":    "LINE-1",
    "asset_id":   "MOTOR-001",
    "opc_id":     "OPC-LINE-1",
    "java_type":  "Float",
    "unit":       "rpm",
    "shift":      "DAY"
  }
}

字段	含义	脚本访问
`data.value`	接收到的值(数值/字符串/布尔值)	`msg.data.value`
`data.quality`	OPC 质量(GOOD/BAD/UNCERTAIN)	`msg.data.quality`
`data.ts`	接收时刻(epoch ms)	`msg.data.ts`
`metadata.tag_id`	标签 ID	`msg.metadata.tag_id`

`flow_on_tag_alarm` — 标签报警发生

标签的报警波段(hi/lo/...)被跨越的瞬间触发。

{
  "type": "TAG_ALARM",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.TEMP" },
  "data": {
    "alarm_band": "HI_HI",
    "value":      95.3,
    "threshold":  90.0,
    "priority":   "ERROR",
    "band_message": "온도 위험"
  },
  "metadata": {
    "tag_id": "MOTOR-001.TEMP",
    "asset_id": "MOTOR-001",
    "site_id": "SITE-01",
    "ts": 1746247200123
  }
}

`data.alarm_band` 值	含义
`NORMAL` / `HI` / `LO` / `HI_HI` / `LO_LO` / `TRIP_HI` / `TRIP_LO`	数值型报警等级
`BOOL_TRUE` / `BOOL_FALSE`	布尔型报警

`flow_on_asset_data` / `flow_on_asset_event` — 资产事件

以资产单位汇总的事件(CEP 处理结果·插件评估结果等)发生时触发。

{
  "type": "ASSET_EVENT",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "event_type":  "STARTUP",
    "details":     { "rpm_target": 1500 }
  },
  "metadata": {
    "asset_id": "MOTOR-001",
    "site_id": "SITE-01",
    "ts": 1746247200123
  }
}

flow_on_asset_data 传递资产单位时序数据(data.values 为键-值映射),flow_on_asset_aggregation 传递分/时单位聚合值。

`flow_on_asset_health_status` / `flow_on_asset_connection_status` — 周期评估

平台以1分钟周期评估资产单位健康/连接状态。

{
  "type": "ASSET_HEALTH_STATUS",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "status":       "WARN",
    "info_count":   12,
    "warn_count":   3,
    "error_count":  0,
    "prev_status":  "OK"
  },
  "metadata": { "asset_id": "MOTOR-001", "ts": 1746247200123 }
}

`data.status` 值	含义
`OK` / `WARN` / `ERROR` / `UNKNOWN`	健康等级
`CONNECTED` / `LATENT` / `ERROR` / `DISCONNECTED` / `UNKNOWN`	连接等级(connection_status)

与 prev_status 比较,常用只在状态发生迁移的瞬间触发后续动作的过滤模式。

`flow_on_oee_event` / `flow_on_ram_event` / `flow_on_ems_event` — 插件事件

工单单位 OEE/RAM/EMS 评估结果更新时触发。发起方(originator)是设备,计算结果位于 data.payload。

{
  "type": "OEE_EVENT",
  "originator": { "entity_type": "Asset", "id": "ASSET_DJ_M_0001" },
  "data": {
    "event_type":     "OEE_CALCULATED",
    "plugin":         "OEE",
    "statement_type": "RESULT",
    "asset_id":       "ASSET_DJ_M_0001",
    "timestamp":      1746247200123,
    "payload": {
      "oee":            0.78,
      "availability":   0.95,
      "performance":    0.85,
      "quality":        0.97,
      "good_quantity":  { "quantity_string": "1560", "data_type": "integer", "unit_of_measure": "EA" },
      "scrap_quantity":  { "quantity_string": "42",   "data_type": "integer", "unit_of_measure": "EA" },
      "rework_quantity": { "quantity_string": "6",    "data_type": "integer", "unit_of_measure": "EA" },
      "run_time_ms":    25200000,
      "order_id":       "ORDER_000001"
    }
  }
}

good_quantity·scrap_quantity·rework_quantity 是工单单位的 Quantity 对象(ISA-95 / B2MML)。值是十进制字符串, 在脚本中计算时请像 Number(data.payload.good_quantity.quantity_string) 这样转换,单位通过 unit_of_measure 确认。旧的 good_count·bad_count 字段已不存在。字段规则见 Order 服务 — 数量。
scrap_quantity 只是报废,返工是 rework_quantity。需要不良合计时请把两者相加(良品 + 报废 + 返工 = 生产)。
未计算的比率(«未适用»)和数量是 null,而不是 0。

`flow_on_opc_status` / `flow_on_edge_status` — OPC/Edge 状态

{
  "type": "OPC_STATUS",
  "originator": { "entity_type": "OPC", "id": "OPC-LINE-1" },
  "data": {
    "connection_status": "CONNECTED",
    "scan_status":       "START",
    "prev_status":       "DISCONNECTED"
  },
  "metadata": { "opc_id": "OPC-LINE-1", "edge_id": "EDGE-A", "ts": 1746247200123 }
}

`flow_on_diagnostic` — 系统诊断消息

服务器模块产生的诊断消息进入时触发。

{
  "type": "DIAGNOSTIC",
  "originator": { "entity_type": "Module", "id": "cep-engine" },
  "data": {
    "level":   "WARN",
    "code":    "PATTERN_LAG",
    "summary": "EQL 패턴 평가 지연 1.2s",
    "module":  "cep-engine"
  },
  "metadata": { "ts": 1746247200123 }
}

`data.level`	含义
`INFO` / `WARN` / `ERROR`	诊断严重程度

`flow_on_domain_changed` — 域变更事件

资产·标签·站点·作业指示等域实体被 CRUD 时触发。

{
  "type": "DOMAIN_CHANGED",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "action":      "UPDATED",
    "before":      { "asset_name": "Motor1" },
    "after":       { "asset_name": "Motor 01 - Renamed" },
    "changed_by":  "admin"
  },
  "metadata": { "ts": 1746247200123 }
}

`flow_on_entity_event` — 实体生命周期

以 ENTITY_CREATED / ENTITY_UPDATED / ENTITY_DELETED 三种类型统一触发。单个节点即可接收全部三种生命周期。

{
  "type": "ENTITY_CREATED",
  "originator": { "entity_type": "Customer", "id": "CUST-9001" },
  "data": { "customer_name": "신규 고객", "external_id": "ERP-CUST-9001" },
  "metadata": { "ts": 1746247200123 }
}

`flow_on_webhook` — 外部 HTTP 推送

外部系统以 POST /webhook/flow/{flow_id} 发送的载荷会原样转换为消息。通过 URL 路径中的 {flow_id} 和请求头 X-API-Key 进行认证。

旧路径 /flow/webhook/{flow_id} 会返回 410 Gone

该路径可在无认证的情况下以任意 flow_id 触发流程,因此已被停用 (FlowController 返回 410)。新入口点 /webhook/flow/{flow_id} 中 WebhookController 会验证 API 密钥·HMAC。是前后颠倒的路径因此不易察觉 — 不是 flow/webhook 而是 webhook/flow。

{
  "type": "WEBHOOK",
  "originator": { "entity_type": "External", "id": "ERP" },
  "data": {
    "order_no":  "PO-20260512-001",
    "customer":  "ACME",
    "quantity":  1000
  },
  "metadata": {
    "http_method": "POST",
    "remote_addr": "10.20.0.55",
    "request_id":  "req-7c0a...",
    "ts": 1746247200123
  }
}

外部系统发送的完整 JSON 本体会原样进入 data。HTTP 请求头仅部分(remote_addr/method/request_id)显示在 metadata 中。

`flow_on_mqtt_subscribe` — MQTT 主题订阅

{
  "type": "MQTT_INBOUND",
  "originator": { "entity_type": "Topic", "id": "factory/line1/events" },
  "data": { "event": "STARTUP", "rpm": 1500 },
  "metadata": {
    "topic":   "factory/line1/events",
    "qos":     1,
    "broker":  "tcp://mqtt.example.com:1883",
    "ts": 1746247200123
  }
}

`flow_jdbc_poll` — 外部数据库周期轮询

周期性执行配置的 SELECT 查询,为每一行触发一条消息。

{
  "type": "KAFKA_INBOUND",
  "originator": { "entity_type": "DB", "id": "mes_db" },
  "data": {
    "PO_NO":    "PO-20260512-001",
    "CUSTOMER": "ACME",
    "QTY":      1000,
    "DUE_DATE": "2026-05-20"
  },
  "metadata": {
    "datasource": "mes_db",
    "query":      "SELECT * FROM po WHERE status='NEW'",
    "row_index":  0,
    "ts": 1746247200123
  }
}

若有100行,则依次触发100条消息。为避免重复处理相同行,请在 SELECT 查询中同时更新处理标志,或加入 processed_at 列的比较条件。

`flow_schedule` — 基于时间触发

以 Cron 表达式或固定周期触发。载荷为空,仅填充 metadata.ts。

{
  "type": "TIMER",
  "originator": { "entity_type": "Schedule", "id": "daily-report" },
  "data": {},
  "metadata": {
    "cron":      "0 0 8 * * ?",
    "fired_at":  1746247200000,
    "ts":        1746247200000
  }
}

载荷转换注意事项

originator.id 为域 ID — 通过 flow_change_originator 节点变更后,其后的 flow_save_attributes·flow_publish_asset_* 动作节点会基于新的 originator 工作。
用脚本替换 data/metadata 时,推荐直接赋值而非浅拷贝(Object.assign) — 修改原始对象可能影响订阅同一触发器的其他流程。
所有 epoch 时刻字段均为毫秒(ms)。若需秒单位,请使用 Math.floor(msg.metadata.ts / 1000)。

节点目录

详细的节点 ID 和选项可在编辑画面的检查器中确认。

触发器 (22种)

所有流程的起点都是触发器节点。没有单独的入口/终点节点。

域自动接收 — 系统内部事件会自动分发。

类别	触发器节点
标签	`flow_on_tag_point` (遥测), `flow_on_tag_alarm`
资产	`flow_on_asset_data`, `flow_on_asset_event`, `flow_on_asset_alarm`, `flow_on_asset_command`, `flow_on_asset_aggregation`, `flow_on_asset_context`, `flow_on_asset_health_status`, `flow_on_asset_connection_status`
插件	`flow_on_oee_event`, `flow_on_ram_event`, `flow_on_ems_event`
OPC/Edge	`flow_on_opc_status`, `flow_on_edge_status`
诊断·域	`flow_on_diagnostic`, `flow_on_domain_changed`
实体	`flow_on_entity_event`

没有名为 flow_on_alarm 的节点

报警触发器根据对象分为两种 — 标签报警是 flow_on_tag_alarm, 资产报警是 flow_on_asset_alarm。旧文档的示例曾使用 flow_on_alarm, 但调色板中没有该名称,直接照做将找不到该节点。

所有触发器节点均可通过 *_pattern 选项(glob:*、?)进行按消息单位的预过滤。未匹配模式的消息不会传递给后续节点,执行计数也不会增加(标记为 SKIPPED)。为最小化运维负载,建议优先在触发器阶段进行筛选。

外部接入

节点	动作
`flow_on_webhook`	接收外部系统通过 HTTP 推送的载荷
`flow_on_mqtt_subscribe`	订阅外部 MQTT broker 主题
`flow_jdbc_poll`	周期性读取外部数据库 SELECT 结果,逐行触发

基于时间

节点	动作
`flow_schedule`	Cron/周期计划 — 触发 `TIMER` 消息

过滤 (6种)

节点	说明
`flow_msg_type_filter`	若 `type` 包含在指定列表中则 `TRUE`
`flow_originator_type_filter`	若 `originator.entity_type` 包含在指定列表中则 `TRUE`
`flow_script_filter`	用脚本进行 boolean 判断
`flow_check_existence_field`	`data`/`metadata` 特定字段是否存在
`flow_switch`	多重 case 分支(每个 case 有不同 relation)
`flow_check_relation`	根据前一步的 relation 进行分支

转换 (8种)

节点	画面名称	说明
`flow_script_transform`	脚本转换	用脚本转换 `data`/`metadata`
`flow_change_originator`	发信者变更	将 `originator` 变更为其他实体
`flow_rename_keys`	键名变更	批量变更 `data` 字段名
`flow_template`	模板	生成 `${path}` 替换文本
`flow_split`	拆分	若 `data` 为数组,按各元素拆分为消息
`flow_merge`	合并	将多条消息以时间窗口合并 — `flow_split` 的反操作
`flow_flatten`	平坦化	将嵌套对象的子键提升到顶层 (`data.data_map.x` → `data.x`)
`flow_to_email`	邮件转换	将消息转换为邮件格式

flow_flatten 用于希望后续节点直接以 ${data.x} 引用像标签点位那样在 data_map 内又多包一层值的消息时使用。

动作 — 集成·存储 (2种)

节点	说明
`flow_save_tag_point`	加载标签点位(与正常摄取路径相同)
`flow_save_attributes`	部分更新标签/资产元信息

若您在寻找 flow_dds_publish(向内部消息通道自由发布),此处没有该节点, 它位于 外部集成 调色板组中。

动作 — 资产事件发布 (4种)

以与 CEP(EQL)相同的处理路径发布资产事件(永久存储 + 缓存 + 时间线 + 插件 + 消息通道一致处理)。

节点	频道
`flow_publish_asset_event`	资产事件
`flow_publish_asset_context`	资产上下文
`flow_publish_asset_aggregation`	资产聚合
`flow_publish_asset_command`	资产命令

`flow_asset_command_invoke` — 执行而非发布

节点	画面名称	作用
`flow_asset_command_invoke`	资产命令调用	同步执行资产上定义的命令并等待结果

容易与 flow_publish_asset_command 混淆

flow_publish_asset_command — 向频道发布资产命令事件。发布即结束。
flow_asset_command_invoke — 实际执行资产上定义的命令,并等待收到结果。

若希望命令被执行却使用了发布节点,会看起来像什么都没发生一样。

动作 — 域 CRUD (34种)

所有域操作均委托给同一域服务,保证审计·一致性。可以通过 *_field 动态选项从消息载荷中提取值。

域	Create	Update	Delete
资产 (Asset)	`flow_create_asset`	`flow_update_asset`	`flow_delete_asset`
标签 (Tag)	`flow_create_tag`	`flow_update_tag`	`flow_delete_tag`
站点/区域/产线	`flow_create_site`	`flow_update_site`	`flow_delete_site`
作业指示 (WorkOrder)	`flow_create_work_order`	`flow_update_work_order`	`flow_delete_work_order`
报警设置 (EQL)	`flow_create_alarm_config`	`flow_update_alarm_config`	`flow_delete_alarm_config`
客户 (Customer)	`flow_create_customer`	`flow_update_customer`	`flow_delete_customer`
产品 (Product)	`flow_create_product`	`flow_update_product`	`flow_delete_product`
作业者 (Employee)	`flow_create_employee`	`flow_update_employee`	`flow_delete_employee`
日历(班次)	`flow_create_calendar`	`flow_update_calendar`	`flow_delete_calendar`

标签报警波段部分更新 (2种)

节点	说明
`flow_update_tag_alarm_band_numeric`	仅部分更新输入字段的数值型报警波段(`hi`/`lo`/`hi_hi`/`lo_lo`/`trip_hi`/`trip_lo`/`band_message`/`use_alarm`)
`flow_update_tag_alarm_band_boolean`	仅部分更新输入字段的布尔型报警波段(`bool_true`/`bool_false`/优先级/消息/`use_alarm`)

作业指示(WorkOrder)状态迁移 (5种)

不直接更新数值列,而是调用域服务的状态迁移方法,因此保持 OEE/RAM/EMS 可见性。

节点	迁移
`flow_start_work_order`	`WAIT` → `START`
`flow_pause_work_order`	`START` → `PAUSED`
`flow_resume_work_order`	`PAUSED` → `START`
`flow_end_work_order`	`START` 或 `PAUSED` → `END`
`flow_abort_work_order`	`START` 或 `PAUSED` → `ABORTED` (`abort_code`、`notes` 选项)

NOT NULL 自动补全 — Create 节点会自动为 NOT NULL 列填入 default 值。例如:作业指示 status="WAIT" / master_id 为 MES 主 ID(无则自动回填),客户经理信息 "admin"/"admin@example.com",作业者 org_id 回退为 site_id,所有行的 insert_user_id="flow"。FK 列(如 customer_id/product_id)中的空字符串会转换为 NULL。

Update 节点 — 部分更新 — Customer/Product/Employee/Calendar 和报警波段 Update 节点会先查询已有记录,再仅合并输入字段后保存。空字符串·null 值会被忽略,保持原值。若需要全量覆盖,请使用 Delete + Create 组合。

有意排除了直接触发报警的节点。报警必须仅通过 flow_create_alarm_config 路径产生,才能保证报警历史的一致性。

Edge (Edge, 22种)

通过调用 Edge 设备(OPC Agent)的 REST API,自动化 OPC 服务器注册·标签 CRUD·标签值读写查询·Docker 应用控制。所有节点在语义上共享相同动作,仅默认 HTTP 方法(GET/POST/PUT/DELETE)不同。

组	节点
OPC 服务器管理	`flow_edge_opc_create`, `flow_edge_opc_update`, `flow_edge_opc_delete`, `flow_edge_opc_start`, `flow_edge_opc_stop`, `flow_edge_opc_list`
标签管理	`flow_edge_tag_create`, `flow_edge_tag_update`, `flow_edge_tag_delete`, `flow_edge_tag_read`, `flow_edge_tag_write`, `flow_edge_tag_list`
查询	`flow_edge_monitoring`, `flow_edge_info`, `flow_edge_transfer`
Docker 应用控制	`flow_edge_app_list`, `flow_edge_app_inspect`, `flow_edge_app_start`, `flow_edge_app_stop`, `flow_edge_app_restart`, `flow_edge_app_logs`, `flow_edge_app_stats`

查询3种

节点	画面名称	作用
`flow_edge_monitoring`	监控	查询 Edge 监控指标
`flow_edge_info`	Edge 信息	查询 Edge ID·版本·uptime
`flow_edge_transfer`	传输健康	MQTT/Sparkplug 传输状态

节点	画面名称	作用
`flow_edge_opc_list`	OPC 列表	查询 Edge 的 OPC 服务器列表
`flow_edge_tag_list`	标签列表	查询 OPC 的标签列表

Docker 应用控制 7种

将Edge 详情画面Docker 面板中手动执行的操作以流程自动化。

节点	画面名称	作用
`flow_edge_app_list`	应用列表	Edge Docker 容器列表
`flow_edge_app_inspect`	应用详情	容器详情(inspect)
`flow_edge_app_start`	应用启动	启动容器
`flow_edge_app_stop`	应用停止	停止容器
`flow_edge_app_restart`	应用重启	重启容器
`flow_edge_app_logs`	应用日志	容器日志 (`line=N`)
`flow_edge_app_stats`	应用统计	容器 CPU / 内存 / I/O

应用控制节点会实际操作现场设备

app_stop · app_restart 会真正停止并重新启动运行在现场 Edge 上的容器。若挂接在触发器上自动执行,请将条件设置得较窄 — 若对不稳定报警挂上 app_restart,容器会反复重启。

通用设置

选项	说明
`url`	Edge REST 端点。支持 `${data.x}`/`${metadata.y}` 模板替换
`method`	HTTP 方法(未设置时按节点默认值 — 如 create=POST, update=PUT, delete=DELETE, read/monitoring=GET)
`headers`	JSON 请求头(例:`{"Authorization":"Bearer ${TOKEN}"}`)
`body_template`	请求体(未设置时原样发送 `data`,GET/DELETE 不发送本体)
`timeout_ms`	超时(默认5000)

响应·分支

data.response_status — HTTP 状态码
data.response — 响应本体(字符串)
data.error — 错误消息(失败时)
SUCCESS (200~399) / FAILURE (其他或异常)

外部集成 (9种)

所有外部节点均支持 *_field 动态选项。

节点	动态选项
`flow_dds_publish`	向内部消息通道自由发布(虽非外部调用,但归于此组)
`flow_http_request`	`url_field` / `method_field` / `body_field`
`flow_kafka_publish`	`topic_field` / `key_field`
`flow_mqtt_publish`	`topic_field`
`flow_webhook_callback`	`url_field`
`flow_send_email`	`to_field` / `cc_field` / `subject_field` / `body_field`
`flow_send_sms`	`to_field` / `text_field`
`flow_send_push`	`title_field` / `body_field`
`flow_jdbc_query`	SQL 静态(SELECT/INSERT/UPDATE/DELETE)

流程控制 (7种)

节点	说明
`flow_log`	调试日志(level / prefix)
`flow_noop`	通过
`flow_delay`	`delay_ms` 后传递给下一节点
`flow_throttle`	`max_msgs` / `window_ms` 限制(超出时 `THROTTLED` relation)
`flow_debounce`	`window_ms` 稳定后仅触发最后一条消息
`flow_merge`	在 `window_ms` 期间累积输入,以 `data.merged` 数组形式一次性 emit
`flow_subflow`	`target_flow_id` — 调用其他流程
`flow_retry`	`max_attempts`(默认3) / `backoff_ms`(默认1000) / `backoff_multiplier`(默认2.0)。退避等待后以 `SUCCESS` 分支传递消息,达到最大尝试次数时以 `EXHAUSTED` 分支。自动更新 `metadata.retry_count` / `metadata.retry_exhausted`

Retry 推荐连线模式

[risky_node] ─[FAILURE]─▶ [flow_retry] ─[SUCCESS]─▶ (다시 risky_node 로 루프 연결)
                                       └[EXHAUSTED]─▶ [에러 핸들러 / 알림]

节点选项详细参考

不仅以一行表格,还详细说明复杂节点的选项默认值·示例·失败处理。

`flow_script_filter` — 脚本过滤

选项	类型	默认值	说明
`script`	text	(必填)	评估式 — 返回 boolean。`true` → `TRUE` 分支 / `false` → `FALSE` 分支
`script_type`	enum	`javascript`	`javascript` / `eql`
`on_error`	enum	`FALSE`	脚本异常时 — 发送到 `TRUE` / `FALSE` / `FAILURE` 哪个分支

脚本上下文

变量	含义
`msg.type`	消息类型(POST_TELEMETRY 等)
`msg.data`	载荷(可修改但在过滤中无意义)
`msg.metadata`	上下文
`msg.originator`	originator 对象

示例

// 온도가 임계 초과 + 야간 시프트만
msg.data.value > 80 && msg.metadata.shift === 'NIGHT'

// 사이트별 임계 분기
var th = {'SITE-A': 80, 'SITE-B': 90, 'SITE-C': 75};
msg.data.value > (th[msg.metadata.site_id] || 100);

`flow_script_transform` — 脚本转换

选项	类型	默认值	说明
`script`	text	(必填)	转换式 — 修改 `msg` 对象或返回新对象
`script_type`	enum	`javascript`	`javascript` / `eql`
`mode`	enum	`mutate`	`mutate`(in-place) / `return`(使用 return 值)

示例

// data 에 계산 필드 추가
msg.data.fahrenheit = msg.data.value * 9/5 + 32;
msg.metadata.processed_at = Date.now();

// 페이로드 통째로 교체 (mode=return)
return {
  type: 'WEBHOOK',
  originator: msg.originator,
  data: { temp: msg.data.value, level: msg.data.value > 80 ? 'HIGH' : 'OK' },
  metadata: msg.metadata
};

在 mutate 模式下,即使有 return 语句也会被忽略。若要替换为新对象,必须设置为 mode=return。

`flow_switch` — 多重分支

选项	类型	默认值	说明
`cases`	array	(必填)	`[{expression, relation}]` 数组 — 自上而下评估,采用首个匹配
`default_relation`	string	`DEFAULT`	所有 case 均不匹配时
`script_type`	enum	`javascript`	—

示例

// cases 설정
[
  { "expression": "msg.data.value > 90", "relation": "CRITICAL" },
  { "expression": "msg.data.value > 80", "relation": "WARN" },
  { "expression": "msg.data.value > 70", "relation": "INFO" }
]
// default_relation: "NORMAL"

后续节点可以对4个分支标签(CRITICAL/WARN/INFO/NORMAL)分别进行不同处理。

`flow_retry` — 自动重试

自动恢复外部 IO 节点的临时性失败。

选项	类型	默认值	说明
`max_attempts`	int	3	最大尝试次数(超过此值则 `EXHAUSTED`)
`backoff_ms`	long	1000	首次等待时间(ms)
`backoff_multiplier`	double	2.0	指数退避倍数 — 第1次1秒 → 第2次2秒 → 第3次4秒
`max_backoff_ms`	long	30000	单次等待上限
`jitter_pct`	int	0	在退避中加入 ±N% 随机抖动(避免 thundering herd)

`metadata` 自动补全

字段	含义
`metadata.retry_count`	到目前为止的尝试次数
`metadata.retry_exhausted`	true 时进入 EXHAUSTED 分支
`metadata.retry_last_error`	最后一次失败原因

若退避总和超过60秒,可能阻塞触发器处理队列。若外部系统响应持续缓慢,请先通过 flow_throttle 限制进入速度。

`flow_on_webhook` — HTTP 触发器

选项	类型	默认值	说明
`auth_required`	boolean	`true`	`X-API-Key` 请求头是否必填 — 关闭则任何人都可调用
`allowed_origins`	csv	`*`	CORS Origin 白名单
`max_body_kb`	int	256	本体大小上限(超出则返回413)
`payload_pattern`	glob	`*`	消息预过滤 glob

调用方法

curl -X POST \
  https://platform.example.com/webhook/flow/{flow_id} \
  -H "X-API-Key: {edge_or_token_key}" \
  -H "Content-Type: application/json" \
  -d '{"order_no":"PO-001","customer":"ACME","quantity":1000}'

路径中的 {flow_id} 可在列表画面中复制
X-API-Key 是在 Edge API Key 或 API 认证令牌画面颁发的令牌
响应:200 OK (已 enqueue 到消息队列) / 401(认证失败) / 404(无流程或未部署) / 413(本体超出)

`flow_http_request` — 外部 HTTP 调用

选项	类型	默认值	说明
`url`	string	(必填)	要调用的 URL。支持 `${data.x}` 模板替换
`url_field`	string	—	从载荷动态获取 URL 时 — 如 `data.endpoint`
`method`	enum	`GET`	`GET` / `POST` / `PUT` / `DELETE` / `PATCH`
`method_field`	string	—	从载荷动态获取方法
`headers`	json	`{}`	`{"Authorization": "Bearer ${TOKEN}"}` 形式
`body`	text	—	静态本体(支持模板替换)
`body_field`	string	—	从载荷获取本体时 — 通常为 `data`
`timeout_ms`	int	5000	响应等待上限
`follow_redirect`	boolean	`true`	自动跟随 3xx 重定向
`verify_ssl`	boolean	`true`	TLS 证书验证(仅用于测试请关闭)

响应载荷补全

字段	含义
`data.response_status`	HTTP 状态码 (200 / 404 / 500 ...)
`data.response_body`	响应本体(JSON 则自动解析)
`data.response_headers`	响应头对象

分支

SUCCESS — 2xx/3xx
FAILURE — 4xx/5xx 或异常/超时

`flow_send_email` — 邮件发送

选项	类型	默认值	说明
`to`	string	—	静态收件人(逗号分隔)
`to_field`	string	—	从载荷提取收件人 — 如 `data.recipient`
`cc` / `cc_field`	string	—	抄送
`bcc` / `bcc_field`	string	—	密送
`subject` / `subject_field`	string	(必填其一)	主题 — 支持模板替换
`body` / `body_field`	text	(必填其一)	本体(允许 HTML)
`is_html`	boolean	`true`	若为纯文本邮件请关闭
`attachments`	json	`[]`	`[{"url":"...","filename":"..."}]`

SMTP 设置由操作员预先在系统 → 设置的邮件设置中注册。注册前所有邮件节点均落入 FAILURE 分支。

`flow_jdbc_poll` — 外部数据库轮询

选项	类型	默认值	说明
`datasource_id`	string	(必填)	系统 → 设置中注册的外部数据库标识符
`query`	sql	(必填)	SELECT 查询 — 单次最多返回 1,000 行
`poll_interval_ms`	int	60000	轮询周期(默认1分钟)
`marker_column`	string	—	"最后处理时刻"列 — 仅 SELECT 标记之后的行
`marker_initial`	string	`1970-01-01 00:00:00`	首次轮询时标记起始值
`row_limit`	int	1000	单次轮询最大行数(即使超出也安全)
`on_error_continue`	boolean	`true`	数据库出错时仅记录诊断并进入下一轮询

`marker_column` 使用示例

SELECT po_no, customer, qty, created_at
FROM po
WHERE created_at > :marker
ORDER BY created_at

→ :marker 处会自动填入上次轮询中获取的最大 created_at 值。

`flow_kafka_publish` / `flow_mqtt_publish` — 外部发布

选项	类型	默认值	说明
`broker`	string	(必填)	`kafka:9092` 或 `tcp://mqtt:1883`
`topic`	string	—	静态主题 — 可 `${data.x}` 替换
`topic_field`	string	—	从载荷提取主题(如 `data.target_topic`)
`key` / `key_field`	string	—	(仅 Kafka)消息键
`body` / `body_field`	json/text	(必填其一)	发布本体 — 未设置时原样使用 `data`
`qos`	int	1	(仅 MQTT) 0/1/2
`retain`	boolean	false	(仅 MQTT) Retained 标志

`flow_publish_asset_*` — 资产事件发布

资产事件(事件/上下文/聚合/命令)会以 CEP/插件/时间线同时识别的统一频道发布。4个节点的共同选项:

选项	类型	默认值	说明
`asset_id`	string	—	静态资产 ID
`asset_id_field`	string	—	从载荷提取资产 ID(通常为 `metadata.asset_id`)
`event_type`	string	—	资产事件分类(例:`STARTUP`, `SHUTDOWN`, `MAINTENANCE`)
`event_type_field`	string	—	从载荷提取分类
`payload`	json	`${data}`	发布本体 — 未设置时原样使用 `data`

若是 flow_publish_asset_command,会立即通过资产的命令接收主题(asset_id/cmd/{event_type})到达 Edge。

`flow_create_` / `flow_update_` — 域 CRUD

所有 Create/Update 节点共享以下选项模式。

选项	类型	说明
`{컬럼명}`	string	静态值(不输入则为 NULL/default)
`{컬럼명}_field`	string	从载荷提取值(`data.foo` / `metadata.bar`)
`id_strategy`	enum	`auto`(系统颁发) / `field`(从 `{도메인}_id_field` 提取)
`on_duplicate`	enum	`error`(默认) / `skip` / `update` — 仅 Create 节点

NOT NULL 自动补全

Create 节点会自动为 NOT NULL 列填入 default 值。

域	自动填充的列
作业指示	`status="WAIT"` / `master_id`(无则自动回填) / `insert_user_id="flow"`
客户	经理信息 `"admin"` / `"admin@example.com"` (未注册的情况)
作业者	`org_id=site_id`(回退)
通用	`insert_date=now()` / `insert_user_id="flow"`

FK 空字符串处理

若 FK 列(customer_id/product_id 等)输入空字符串 "",会自动转换为 NULL。在 JS 中 msg.data.customer_id = '' 比 delete msg.data.customer_id 更安全。

`flow_edge_*` — Edge REST 调用

调用 Edge 设备 REST 端点的22种节点共享通用选项。

选项	类型	默认值	说明
`edge_id`	string	—	静态 Edge ID
`edge_id_field`	string	—	从载荷提取 Edge ID
`path`	string	(按节点默认)	Edge REST 路径(例:`/api/v1/opc`, `/api/v1/app/grafana/start`)
`body_template`	text	—	请求本体 — 未设置时原样使用 `data`
`timeout_ms`	int	5000	—

自动认证

Edge 节点在调用时会自动从 mm_edge 主表 lookup 该 Edge 的 api_key,并以 X-API-Key 请求头附加。操作员无需另行设置。

响应载荷

字段	含义
`data.edge_response_status`	Edge 响应码
`data.edge_response`	响应本体
`data.edge_id`	调用目标 Edge ID(用于确认)

分支与 flow_http_request 相同(SUCCESS / FAILURE)。

编写脚本节点

flow_script_filter / flow_script_transform / flow_switch 节点支持两种表达式。

JavaScript(默认推荐)

标准 ECMAScript 语法。支持多行·var/let/const·函数·对象字面量等全部特性。

绑定

变量	说明
`msg`	完整消息。`msg.data.x`、`msg.metadata.topic`、`msg.type`、`msg.originator.id` 均可直接访问
`data`	`msg.data` 的简写别名
`metadata`	`msg.metadata` 的简写别名

过滤示例

data.temp > 80

转换示例

data.temp_f = data.temp * 1.8 + 32;
data.alert = data.temp > 80 ? 'HIGH' : 'OK';
msg

Switch 示例(每个 case 为 boolean)

data.t > 100   // case "Critical"

常用转换模式

// 1) 단위 변환 (섭씨 → 화씨) + 라벨링
data.temp_f = data.temp * 1.8 + 32;
data.alert  = data.temp > 80 ? 'HIGH' : 'OK';
msg

// 2) 메타데이터 보강 — 시간대·시프트 자동 부여
const h = new Date(metadata.ts).getHours();
metadata.shift = (h >= 6 && h < 18) ? 'DAY' : 'NIGHT';
msg

// 3) 외부 페이로드를 도메인 모델로 매핑 (MES PO → WorkOrder)
const po = data;
data = {
  master_id: 'WO-MES-' + po.po_no,
  asset_id:  po.line_id || 'UNASSIGNED',
  title:     po.product_name + ' (' + po.qty + ')',
  due_date:  po.delivery_date,
  qty:       po.qty
};
msg

// 4) 실패 분기로 명시적 라우팅 (필수 필드 누락 시)
if (!data.tag_id || data.value == null) throw new Error('필수 필드 누락');
msg

// 5) 배열 분할 후 데이터 정제 — split 노드 후에 사용
data.value     = parseFloat(data.raw);
data.threshold = data.value > 100;
msg

常用过滤模式

// 우선순위 화이트리스트
['ERROR', 'CRITICAL'].includes(data.priority)

// 시간대 기반 필터 (주간만 허용)
new Date(metadata.ts).getHours() >= 8 && new Date(metadata.ts).getHours() < 20

// 자산 ID 패턴 매칭
/^MOTOR-.*$/.test(originator.id)

// 임계값 + 안정성 (값이 5번 이상 누적된 경우)
data.value > data.threshold && data.consecutive_count >= 5

用户脚本在安全沙箱中执行,文件·网络·线程·任意类访问均被阻断。若需要调用外部系统,请单独连线 flow_http_request 等外部集成节点。

EQL 表达式

用于与现有 EQL 用户兼容。仅支持单条表达式(不支持多行·分号分隔),只能使用具有副作用的模式。

#msg.getData().getInt('temp') > 80

若需要多重动作,请使用 JavaScript。

JavaScript 执行环境规范

脚本在隔离的沙箱中执行。需要准确了解哪些功能可用/不可用,才能编写稳定的脚本。

可用 (✅)

功能	备注
标准 ECMAScript	`var`/`let`/`const`·函数·类·解构·spread·`?.`·`??` 等
对象字面量	`{ key: value, ... }`
数组方法	`map` / `filter` / `reduce` / `forEach` / `find` / `some` / `every` / `flat` / `slice`
字符串方法	`split` / `replace` / `includes` / `match` / `padStart` / `repeat`
数学函数	`Math.*` 全部
JSON	`JSON.parse` / `JSON.stringify` (若 `data` 已是对象则无需再次 stringify)
Date	`new Date()` / `Date.now()` / `getHours()` / `toISOString()` 等
正则表达式	`/pattern/` 字面量 + `RegExp` 构造器
抛出错误	`throw new Error('...')` — 自动分支到 `FAILURE`
try/catch/finally	异常处理

不可用 (❌)

功能	原因 / 替代方案
网络调用 (`fetch` / `XMLHttpRequest`)	沙箱阻断 — 请单独连线 `flow_http_request` 节点
文件系统 (`require('fs')`)	沙箱阻断
线程 (`setTimeout` / `setInterval` / `Worker`)	仅允许同步执行 — 若需延迟请用 `flow_delay` 节点
`require` / `import`	无法加载外部模块 — 所需函数需在同一脚本内定义
`eval` / `new Function(string)`	出于安全考虑被阻断
任意 Java 类	即使在 EQL 兼容模式下也不对用户代码暴露
`process` / `global` / `window`	未定义
WebSocket / EventSource	阻断 — 消息接收由触发器节点负责

执行限制 — 没有限制

脚本没有设置时间·内存限制

旧文档曾刊载执行时间500ms·内存16MB·堆栈1024·输出2MB的表格, 但这四项均未实际应用。

节点设置中虽然看似可以使用 timeout_ms,但脚本节点不会读取该值。 (ScriptTransformNode · ScriptFilterNode 只会读取 script 和 language。)
JS 执行上下文也没有设置时间·语句数·内存限制。

因此 while (true) {} 这样的脚本会不间断地占用工作线程。 虽然流程整体上限30秒(flow_timeout_ms)会在执行循环取出下一任务时检查, 但脚本内运行的无限循环无法到达那个检查点。

请只编写自行结束的脚本。 避免无限循环·巨大循环·大字符串累积, 循环务必设置上限。

实际被阻断的 — 安全沙箱

与时间·内存不同,触及外部的行为是被明确阻断的。

项目	状态
Java 类访问 (`Java.type` 等)	阻断
文件·网络 IO	阻断
线程创建	阻断
原生访问	阻断
实验性选项	阻断
宿主对象访问	仅允许列表中的对象

若需要外部调用,请不要在脚本中处理,而应通过 flow_http_request 等专用节点单独分离 — 脚本中反正无法执行,专用节点则会实际生效 timeout_ms。

请勿将繁重处理集中于单个脚本,应拆分到多个节点。

`msg` 返回约定

// flow_script_transform 의 두 가지 모드
// 1) mutate 모드 (기본) — msg 객체를 직접 수정, 마지막에 msg 또는 아무 값 반환
data.foo = 'bar';
msg

// 2) return 모드 — 완전히 새 객체로 교체
return {
  type: msg.type,
  originator: msg.originator,
  data: { ...data, foo: 'bar' },
  metadata: msg.metadata
};

标准时刻·地区设置

项目	动作
服务器系统时区	UTC(建议直接使用 epoch ms)
韩国时刻显示	`toLocaleString('ko-KR', { timeZone: 'Asia/Seoul' })`
韩国时刻计算	`new Date(ts + 9*3600000).getUTCHours()` 或以上地区设置
0秒单位 timestamp	`Math.floor(Date.now() / 1000) * 1000`

内存安全编写模式

脚本节点虽有16MB内存上限,但在频繁调用的节点上如果小的内存泄漏累积,会增加引擎 GC 负担。请避免以下模式。

反模式 1 — 闭包保留大数据

// ❌ 나쁜 예 — 큰 배열을 변환 함수에 캡쳐
const heavy = data.records || [];          // 1만 건
const summarize = (item) => heavy.find(r => r.id === item.id);
data.matched = data.targets.map(summarize);
msg

// ✅ 좋은 예 — 인덱스 미리 만들고 함수 안에서만 사용
const index = {};
(data.records || []).forEach(r => { index[r.id] = r; });
data.matched = data.targets.map(t => index[t.id]);
data.records = undefined;     // 변환 후 큰 원본 제거
msg

反模式 2 — 直接传递大量数组

// ❌ data.records 가 10,000 건이면 모든 후속 노드에서 메모리 차지
msg

// ✅ 필요한 통계만 남기고 원본 제거
data.summary = {
  count: (data.records || []).length,
  total: (data.records || []).reduce((s, r) => s + r.value, 0)
};
delete data.records;
msg

反模式 3 — 深度对象拷贝

// ❌ JSON.parse(JSON.stringify(obj)) 는 큰 객체에서 매우 느림
data.copy = JSON.parse(JSON.stringify(data.original));

// ✅ 얕은 복사 또는 필요한 필드만 직접 선택
data.summary = { id: data.original.id, name: data.original.name };

反模式 4 — 正则表达式爆炸(灾难性回溯)

// ❌ (a+)+b 형태의 중첩 그룹은 입력에 따라 지수 시간
const re = /^(a+)+b$/;
if (re.test(data.text)) ...

// ✅ 비포획 그룹 + atomic 그룹 패턴 또는 단순 매칭
const re = /^a+b$/;
if (re.test(data.text)) ...

反模式 5 — 在函数外累积 `let` 变量

// ❌ 함수 밖 let 은 매 노드 호출마다 0으로 리셋되지만, 의도와 다르게 동작 가능
let counter = 0;
data.items.forEach(() => counter++);
data.counter = counter;

// ✅ 명시적으로 함수 안에서 선언
data.counter = data.items.length;     // 같은 결과, 더 명확

反模式 6 — 嵌套 try/catch 无限尝试

// ❌ 실패해도 다시 던지지 않으면 그래프가 잘못된 분기로 이동
try {
  riskyCall();
} catch (e) { /* 무시 */ }
msg

// ✅ 실패 의도면 throw, 정상 의도면 명시적으로 기록
try {
  riskyCall();
  data.status = 'OK';
} catch (e) {
  data.status = 'ERROR';
  data.error_msg = e.message;
}
msg

脚本内 throw 的异常会进入 FAILURE 分支且消息会被保留。外部 IO 节点自身拥有独立分支,请不要用 try/catch 包裹。

内存压力诊断

没有像 NODE_SCRIPT_OOM 这样的代码

旧文档曾将 NODE_SCRIPT_OOM · FLOW_MSG_TRIMMED · ENGINE_GC_PRESSURE 作为内存压力信号提示,但产品不会产生此类代码。 脚本内存也没有上限, 也不会自动截断载荷。

请以指标和症状而非代码来判断内存压力。

观察对象	压力信号
`FLOW_EXECUTOR_DROPPED_COUNT`	从0开始上升 — 处理池饱和,正在丢弃消息
`FLOW_EXECUTOR_QUEUE_SIZE`	持续增加不减少
`FLOW_EXECUTION_TIME`	最大值飙升至平时的数倍
服务器日志	`work_queue full ... task dropped` 警告、`FlowExecutor timeout` 警告
JVM	GC 时间增加·堆使用率上升

若出现上述信号,请首先检查处理大载荷的流程,确认脚本中是否正在累积大量字符串数组。

移动推送通知渠道

用 flow_send_push 节点向操作员的移动应用发送推送通知。

节点选项

选项	类型	默认值	说明
`to` / `to_field`	string	(必填其一)	收件人用户ID(逗号分隔)或载荷路径
`title` / `title_field`	string	(必填其一)	通知标题(建议50字以内)
`body` / `body_field`	text	(必填其一)	正文(建议120字以内)
`priority`	enum	`NORMAL`	`NORMAL` / `HIGH` — HIGH 会显示在锁屏
`sound`	enum	`default`	`default` / `silent` / 自定义提示音
`data`	json	`{}`	应用接收处理的附加数据(载荷4KB上限)
`deep_link`	string	—	点击通知后打开的画面(例:`pp://asset/MOTOR-001`)
`ttl_sec`	int	86400	未接收时保留时间(秒) — 过期后自动删除

用户令牌自动路由

操作员首次登录移动应用时,设备令牌会自动登记到安全 → API 认证令牌。流程中不直接处理令牌,只需指定用户 ID 即可。

若 iOS / Android 双方令牌均已登记,两台设备均会发送
若令牌失效(应用删除等)会自动注销

简单发送示例

[flow_on_asset_alarm]
  ↓ where priority='ERROR'
[flow_send_push]
   to_field:    "metadata.responsible_user"
   title:       "🚨 ${metadata.asset_id} 알람"
   body_field:  "data.band_message"
   priority:    HIGH
   deep_link:   "pp://alarm/view/${data.alarm_id}"

多语言推送

// flow_script_transform — 사용자 로케일에 따라 본문 분기
const locale = metadata.user_locale || 'ko-KR';
const templates = {
  'ko-KR': { title: '🚨 ${asset} 위험 알람',           body: '값 ${value}, 즉시 점검 바랍니다.' },
  'en-US': { title: '🚨 ${asset} Critical Alarm',     body: 'Value ${value}, please inspect immediately.' },
  'ja-JP': { title: '🚨 ${asset} 危険警報',            body: '値 ${value}, 即時点検が必要です。' }
};
const tpl = templates[locale] || templates['ko-KR'];
data.push_title = tpl.title.replace('${asset}', metadata.asset_id).replace('${value}', data.value);
data.push_body  = tpl.body.replace('${value}', data.value);
msg

之后在 flow_send_push 的 title_field=data.push_title / body_field=data.push_body 中使用。

分组通知合并 (Inbox-style)

将多个报警合并为一条通知(以5分钟为单位):

[flow_on_asset_alarm]
   ↓
[flow_merge window=300s]
   ↓ (data.merged 배열)
[flow_script_transform — 요약 만들기]
   ↓ data.push_body="알람 N건: A자산, B자산, ..."
[flow_send_push]

用户不在时自动升级通知

推送未确认30分钟后自动通过 SMS 或邮件升级通知。

[flow_on_asset_alarm priority=ERROR]
   ↓
[flow_send_push]
   ↓ SUCCESS
   ↓ data.alert_id = response_id
[flow_delay 1800s (30분)]
   ↓
[flow_http_request GET /api/alert/${alert_id}/status]
   ↓ (response_body.read=false 면)
[flow_script_filter (data.response_body.read === false)]
   ↓ TRUE
[flow_send_sms]
   to_field: "metadata.responsible_phone"
   text:     "푸시 미확인 30분 경과: ${data.title}"

建议发送频率限制

优先级	建议频率
HIGH(锁屏显示)	每用户每小时5件以下 — 防止报警疲劳
NORMAL	每用户每小时20件以下

请在发送前连线 flow_throttle 节点,或以 flow_debounce 稳定后再发送同一资产的报警。

外部认证令牌自动刷新

在调用外部系统时自动刷新如 OAuth2 一类会过期的令牌的模式。

简单刷新 — 基于时间(cron)

[flow_schedule cron='0 */50 * * * ?']    ← 50분마다 (만료 1시간 전)
   ↓
[flow_http_request]
   url:    "https://auth.example.com/oauth2/token"
   method: POST
   body:   "grant_type=client_credentials&client_id=${creds.client_id}&client_secret=${creds.client_secret}"
   ↓
[flow_script_transform]
   ↓ data.access_token = data.response_body.access_token
[flow_save_attributes]    ← 자격 증명 저장소에 저장
   target_id:   "creds:erp_api"
   attributes:  {"access_token": "${data.access_token}", "expires_at": ${data.response_body.expires_in * 1000 + ts}}

之后在其他流程的 flow_http_request 中:

Headers: Authorization: Bearer ${creds:erp_api.access_token}

主动刷新 — 收到401时

[main flow]
   ↓
[flow_http_request]
   ↓ SUCCESS → 정상 처리
   ↓ FAILURE
[flow_script_filter (response_status === 401)]
   ↓ TRUE
[flow_subflow target_flow_id="refresh-token"]    ← 토큰 갱신
   ↓
[원래 노드로 루프]    ← 갱신된 토큰으로 재시도

使用 Refresh Token

[flow_http_request]
   url:    "https://auth.example.com/oauth2/token"
   method: POST
   body:   "grant_type=refresh_token&refresh_token=${creds.refresh_token}"
   ↓
[flow_script_transform]
   // 새 access_token + 새 refresh_token (rotation)
   data.access_token  = data.response_body.access_token;
   data.refresh_token = data.response_body.refresh_token;
   data.expires_at    = Date.now() + data.response_body.expires_in * 1000;
   msg
   ↓
[flow_save_attributes]

令牌过期临界自动报警

[flow_schedule cron='0 0 * * * ?']    ← 매시
   ↓
[flow_jdbc_query]
   sql: "SELECT id, expires_at FROM credentials WHERE expires_at < NOW() + INTERVAL '1 day'"
   ↓
[flow_split]
   ↓
[flow_send_email]
   subject: "API 토큰 만료 임박: ${data.id}"
   body:    "${data.id} 토큰이 ${data.expires_at} 만료 예정입니다."

建议的凭据保管位置

种类	建议位置
静态(几乎不变)	系统 → 设置的凭据存储库
动态(自动刷新)	用上述模式,以 `flow_save_attributes` 存储于资产元信息
用户级 OAuth	安全 → API 认证令牌画面

请勿将凭据以明文留在图形中,务必以外部存储引用(${creds.x})形式编写。导出/导入时明文令牌也不会包含在 JSON 中。

以 language 选项选择使用哪种语言(JS 或 EQL)。

执行历史画面

可按节点单位按时间顺序查询执行事件。

搜索条件

项目	说明
时间	指定查询期间
级别	`전체` / `INFO` / `WARN` / `ERROR`
流程 ID	仅过滤特定流程
消息 ID	用于追踪单条消息
数量	最近200/500/1,000条

快速时间范围

通过时间线面板顶部按钮立即跳转:10분전 / 30분전 / 1시간전 / 6시간전 / 12시간전 / 전체기간。

结果列

列	说明
级别	INFO / WARN / ERROR
时间	事件发生时刻
事件	`FLOW_START` / `NODE_IN` / `NODE_OUT` / `NODE_ERROR` / `FLOW_END`
流程 ID	属于哪个流程
节点	节点显示名
节点类型	例:`flow_script_transform`
关系	`SUCCESS` / `FAILURE` / `TRUE` / `FALSE` / `MATCH` / `NO_MATCH` / `DEFAULT` / `THROTTLED` / `EXHAUSTED` (均为大写)
消息	消息类型·主体·关系·处理时间·数据预览摘要

可通过 CSV 下载 按钮导出当前查询结果。

执行历史的保留周期为7天。如需长期保留,请导出至外部日志系统。

常规工作流

创建新流程 — 列表画面 새 플로우 按钮 → 输入名称·描述
进入编辑画面 — 自动打开空画布
放置触发器节点 — 从左侧调色板拖出触发器节点
添加处理节点 — 依次放置过滤 → 转换 → 动作,并以连线连接
节点设置 — 点击各节点,在右侧检查器中输入选项
保存 — 右上角 저장 按钮(自动加载快照)
测试运行 — 注入任意消息以确认结果
部署 — 通过 배포 开关激活 → 触发器事件进入时自动执行
监控 — 通过实时调试面板和执行历史画面确认

示例流程

每个示例由节点连线图 + 核心节点设置 + 动作说明构成。JSON 形式的节点设置与右侧检查器中输入的值1:1对应。

示例 1: MES 作业指示自动创建

每分钟轮询 MES 中的新 PO 并转换为作业指示。

[flow_schedule: 매분]
   │
   ▼ TIMER
[flow_http_request: MES /api/po/list?status=NEW]
   │
   ▼ SUCCESS
[flow_split: data → 각 PO별 메시지]
   │
   ▼
[flow_script_transform: PO → WorkOrder 매핑]
   │
   ▼
[flow_create_work_order]
   │
   ├── SUCCESS → [flow_log: 워크오더 생성됨]
   └── FAILURE → [flow_send_email: 실패 알림]

节点设置

节点	核心设置
`flow_schedule`	`cron`: `0 * * * * ?` (每分钟0秒)
`flow_http_request`	`method`: `GET`, `url`: `https://mes.example.com/api/po/list?status=NEW`, `headers`: `{"Authorization":"Bearer ${MES_TOKEN}"}`
`flow_split`	`path`: `data` (拆分为数组)
`flow_script_transform`	`language`: `JS`, 参考下方脚本
`flow_create_work_order`	`master_id_field`: `data.master_id`, `asset_id_field`: `data.asset_id`, `order_name_field`: `data.order_name`, `planned_quantity_field`: `data.planned_quantity`, `unit_of_measure_field`: `data.unit_of_measure`, `external_system_field`: `data.external_system`, `external_id_field`: `data.external_id`
`flow_send_email`	`to_field`: `metadata.alert_to`, `subject`: `[MES 동기화 실패] ${data.po}`

Script Transform 示例

data.master_id = 'WO-MES-' + data.po;
data.order_name = data.product_name + ' (' + data.qty + ' ' + data.uom + ')';
data.asset_id  = data.line_id || 'UNASSIGNED';
data.planned_quantity  = String(data.qty);   // 计划数量 — 大于 0 的十进制数
data.unit_of_measure   = data.uom;           // UN/ECE Rec 20 代码(EA·KGM·LTR …) — 必填,无默认值
data.external_system   = 'MES';              // 外部工单以 (系统, 编号) 对来标识
data.external_id       = data.po;
data.due_date  = data.delivery_date;
metadata.alert_to = 'ops@example.com';
msg

示例 2: 通过外部 webhook 发布资产命令

验证外部系统发送的 HTTP 载荷后,转换为内部资产命令。

[flow_on_webhook]
   │
   ▼ WEBHOOK
[flow_script_filter: payload 검증]
   │
   ├── TRUE
   ▼
[flow_publish_asset_command]
   │
   ├── SUCCESS → [flow_log]
   └── FAILURE → [flow_webhook_callback: 외부에 실패 통보]

调用方法:通过 POST /webhook/flow/{flow_id} 发送 JSON 本体,原样放入 data 区域并触发(旧的 /flow/webhook/… 为410)。

Script Filter 示例

// 인증 토큰 일치 + 필수 필드 존재 검사
if (data.token !== 'EXPECTED_TOKEN') return false;
if (!data.asset_id || !data.cmd_key) return false;
true

示例 3: 报警 → 紧急作业指示自动发布

接收到 ERROR 级别以上的报警时,针对上层资产发布紧急维护工单。

[flow_on_tag_alarm]
   │
   ▼ ALARM
[flow_script_filter: priority = ERROR/CRITICAL]
   │
   ├── TRUE
   ▼
[flow_change_originator: Tag → 상위 Asset]
   │
   ▼
[flow_create_work_order: 긴급 정비]
   │
   └── SUCCESS → [flow_send_email: 정비 담당자]

Script Filter 示例

['ERROR', 'CRITICAL'].includes(data.priority)

示例 4: 外部数据库同步 — 资产批量注册

轮询遗留数据库中的新设备行,自动注册为资产。

[flow_jdbc_poll: SELECT * FROM legacy_assets WHERE sync_status='NEW']
   │
   ▼
[flow_script_transform: 컬럼 매핑]
   │
   ▼
[flow_create_asset]
   │
   ├── SUCCESS → [flow_jdbc_query: UPDATE legacy_assets SET sync_status='OK' WHERE id=?]
   └── FAILURE → [flow_log: ERROR + flow_send_push]

节点设置

节点	核心设置
`flow_jdbc_poll`	`dsn`: 外部数据库连接, `sql`: `SELECT * FROM legacy_assets WHERE sync_status='NEW' LIMIT 100`, `interval_ms`: `60000`
`flow_create_asset`	`asset_id_field`: `data.legacy_id`, `asset_name_field`: `data.name`, `site_id_field`: `data.plant_code`
`flow_jdbc_query`	`sql`: `UPDATE legacy_assets SET sync_status='OK' WHERE id=?`, `params_field`: `data.legacy_id`

示例 5: 作业指示自动状态迁移

以资产上发生的启停事件自动迁移作业指示状态。

[flow_on_asset_event]
   │
   ▼ ASSET_EVENT
[flow_switch: data.event_type 기준]
   │
   ├── case "RUN"   ─▶ [flow_start_work_order] ─▶ [flow_log]
   ├── case "STOP"  ─▶ [flow_pause_work_order] ─▶ [flow_log]
   ├── case "DONE"  ─▶ [flow_end_work_order]   ─▶ [flow_log]
   └── DEFAULT      ─▶ [flow_noop]

Switch 节点 case 示例

RUN: data.event_type === 'RUN'
STOP: data.event_type === 'STOP'
DONE: data.event_type === 'DONE' && data.qty_done >= data.qty_planned

若状态迁移失败(当前状态不正确),会自动路由到 FAILURE,因此无需另外使用 Filter 即可安全连线。

示例 6: 报警波段自动调整

基于资产聚合值(例如6小时平均值)动态调整标签的上限/下限报警波段。

[flow_on_asset_aggregation]
   │
   ▼ ASSET_AGGREGATION
[flow_script_transform: 통계 → 임계값 산출]
   │
   ▼
[flow_update_tag_alarm_band_numeric]
   │
   ├── SUCCESS → [flow_log]
   └── FAILURE → [flow_send_email: 운영자 알림]

Script Transform 示例

// 6h 평균 ± 3σ 를 임계값으로 사용
const mean   = data.mean;
const stddev = data.stddev || 1;
data.tag_id  = originator.id + '.TEMP';
data.hi      = mean + 3 * stddev;
data.lo      = mean - 3 * stddev;
data.hi_hi   = mean + 4 * stddev;
data.lo_lo   = mean - 4 * stddev;
data.use_alarm = true;
msg

示例 7: Edge 设备自动注册

接收新 OPC 服务器信息后批量注册到 Edge 设备。

[flow_on_webhook] (POST 본문에 OPC + 태그 목록)
   │
   ▼
[flow_edge_opc_create]
   │
   ▼ SUCCESS
[flow_split: data.tags 배열]
   │
   ▼
[flow_edge_tag_create]
   │
   ▼ SUCCESS (모든 태그 등록 완료 후)
[flow_edge_opc_start]
   │
   └── SUCCESS → [flow_log: 엣지 디바이스 가동]

调用本体示例

{
  "type": "WEBHOOK",
  "data": {
    "opc_id":   "OPC-LINE-A",
    "endpoint": "opc.tcp://line-a.local:4840",
    "tags": [
      { "tag_id": "MOTOR-001.SPEED", "address": "ns=2;s=Motor1.Speed" },
      { "tag_id": "MOTOR-001.TEMP",  "address": "ns=2;s=Motor1.Temp"  }
    ]
  }
}

示例 8: 外部 API 可靠性增强 (Retry)

对间歇性失败的外部 API 调用应用退避重试,超过最大尝试次数时通知操作员。

[flow_on_asset_event]
   │
   ▼
[flow_http_request: 외부 ERP API]
   │
   ├── SUCCESS  → [flow_log]
   └── FAILURE  ─▶ [flow_retry]
                       │
                       ├── SUCCESS    ─▶ (다시 flow_http_request 로 루프 결선)
                       └── EXHAUSTED  ─▶ [flow_send_email: 'ERP 동기화 N회 실패']

Retry 节点设置

max_attempts: 5
backoff_ms: 2000
backoff_multiplier: 2.0 → 2秒、4秒、8秒、16秒、32秒间隔

示例 9: 插件事件路由(OEE 低下通知)

当 OEE 值降至临界以下时,向产线负责人发送推送通知。

[flow_on_oee_event]
   │
   ▼
[flow_script_filter: data.availability * data.performance * data.quality < 0.6]
   │
   ├── TRUE
   ▼
[flow_template: '라인 ${originator.id} OEE ${data.oee_pct}%']
   │
   ▼
[flow_send_push: 라인 매니저]

Template 示例

body: 라인 ${originator.id} OEE ${data.oee_pct}% (목표 60% 미달, 주요 손실: ${data.top_loss})

示例 10: 消息精炼流水线(Throttle + Debounce)

将高频标签变化限制为每分钟1次,并且仅当额外5秒内没有变化时才触发后续节点。

[flow_on_tag_point]
   │
   ▼
[flow_throttle: max_msgs=1, window_ms=60000]
   │
   ├── SUCCESS    → [flow_debounce: window_ms=5000]
   │                                │
   │                                ▼
   │                          [flow_save_attributes]
   └── THROTTLED  → [flow_log: 차단됨]

示例 11: 多重触发器合并(Merge)

将 OEE/RAM/EMS 三种事件以30秒为单位合并,一次性发送报告消息。

[flow_on_oee_event]  ─┐
[flow_on_ram_event]  ─┼─▶ [flow_merge: window_ms=30000]
[flow_on_ems_event]  ─┘                │
                                       ▼
                              [flow_script_transform: 요약 메시지 생성]
                                       │
                                       ▼
                              [flow_send_email: 일일 요약]

在同一流程中放置多个触发器节点时,均成为入口点。flow_merge 的 data.merged 数组中会累积窗口期间进入的所有消息。

示例 12: 以子流程模块化公共处理

将公共消息精炼逻辑(重复检查 + 单位转换 + 加载)分离为单独流程,由多个触发器调用。

主流程(各自独立)

[flow_on_tag_point]   ─▶ [flow_subflow: target_flow_id=FLOW_00099]
[flow_on_asset_data]  ─▶ [flow_subflow: target_flow_id=FLOW_00099]

子流程 FLOW_00099

(트리거 없음 — 호출 전용)
[flow_log: 'subflow in']
   │
   ▼
[flow_check_existence_field: data.value 존재]
   │
   ├── TRUE
   ▼
[flow_script_transform: 단위 변환]
   │
   ▼
[flow_save_tag_point]

子流程即使没有触发器也可保存,可作为纯外部调用使用(保存时会显示警告)。

示例 13: 班次变更时自动发送日报

每天夜班结束时刻(例:06:00),将昨天~今天的生产·质量·停机摘要以邮件发送。

[flow_schedule: 0 0 6 * * ?]   (매일 06:00)
   │
   ▼ TIMER
[flow_http_request: 내부 통계 API /api/report/daily]
   │
   ▼ SUCCESS
[flow_script_transform: 본문 마크다운 생성]
   │
   ▼
[flow_send_email]

Script Transform 示例

const r = data.report;
data.subject = `[${r.site_id}] ${r.date} 일일 운영 요약`;
data.body    =
  `■ 생산: ${r.qty_done}/${r.qty_planned} (${(100*r.qty_done/r.qty_planned).toFixed(1)}%)\n` +
  `■ 가동률: ${r.availability}%\n` +
  `■ 품질률: ${r.quality}%\n` +
  `■ 다운타임 Top3:\n` +
  r.downtimes.slice(0,3).map(d => ` - ${d.code} ${d.minutes}분`).join('\n');
msg

示例 14: 停机时间自动汇总

当资产 STOP 事件进入时,更新班次单位累积停机时间,超过临界值时发送通知。

[flow_on_asset_event]
   │
   ▼
[flow_msg_type_filter: type in [ASSET_EVENT]]
   │
   ▼ TRUE
[flow_script_filter: data.event_type === 'STOP']
   │
   ▼ TRUE
[flow_script_transform: 다운타임 분 단위 계산]
   │
   ▼
[flow_save_attributes: 자산 누적 다운타임 갱신]
   │
   ▼ SUCCESS
[flow_script_filter: data.shift_downtime_min > 30]
   │
   ▼ TRUE
[flow_send_push: '시프트 다운타임 30분 초과']

示例 15: 质量不良产线自动隔离

若质量检验中连续5件以上不良,向产线资产发布停止命令并中止作业指示。

[flow_on_asset_event]   (event_type=QUALITY_FAIL)
   │
   ▼
[flow_script_transform: data.consecutive_fail = (... + 1)]
   │
   ▼
[flow_save_attributes]
   │
   ▼ SUCCESS
[flow_script_filter: data.consecutive_fail >= 5]
   │
   ▼ TRUE
[flow_publish_asset_command: cmd_key='STOP']
   │
   ▼
[flow_abort_work_order: abort_code='QUALITY']
   │
   └── SUCCESS → [flow_send_email: 품질 매니저 + 라인 매니저]

示例 16: 能耗超阈值 — 产线暂停建议

当产线每小时能耗超出预算时,向操作员发送带建议消息的 SMS。

[flow_on_ems_event]
   │
   ▼
[flow_script_filter: data.power_kwh > data.budget_kwh * 1.2]
   │
   ▼ TRUE
[flow_template: '${originator.id} 시간당 ${data.power_kwh}kWh (예산 ${data.budget_kwh}kWh 초과)']
   │
   ▼
[flow_send_sms]
   │
   └── SUCCESS → [flow_save_attributes: 자산에 마지막 경고 시각 기록]

示例 17: 外部系统双向同步 — 作业指示状态镜像

将外部 ERP 传入的状态与内部作业指示状态进行双向同步。

向下 (ERP → 内部)

[flow_on_mqtt_subscribe: erp/work-order/status]
   │
   ▼
[flow_script_transform: 메시지 → 도메인 매핑]
   │
   ▼
[flow_switch: data.status]
   │
   ├── case "STARTED" ─▶ [flow_start_work_order]
   ├── case "PAUSED"  ─▶ [flow_pause_work_order]
   ├── case "DONE"    ─▶ [flow_end_work_order]
   └── DEFAULT        ─▶ [flow_log: WARN]

向上 (内部 → ERP)

[flow_on_entity_event]   (originator.entity_type=Order)
   │
   ▼
[flow_msg_type_filter: type in [ENTITY_UPDATED]]
   │
   ▼ TRUE
[flow_template: ERP 형식으로 변환]
   │
   ▼
[flow_mqtt_publish: erp/work-order/status]

为避免双向同步中的无限循环,请以 metadata.source 等键标记消息来源,并在触发器阶段过滤掉自身发布的消息。

导入/导出

可将流程定义序列化为 JSON,以在其他环境移植或备份。

动作	位置	说明
导出	编辑画面顶部 `내보내기` 按钮	将图形 + 节点 + 连线 + 节点设置整体下载为 JSON
导入	列表画面顶部 `가져오기` 按钮	粘贴或上传 JSON 文本
自动快照	保存时自动	保存图形时按版本单位加载(以备回滚)

导入动作

始终会颁发新流程 ID(防止覆盖现有 ID)

节点 ID 也会重新赋予,连线链接会自动重新映射

导入后的流程立即以解除状态加载,需要操作员审核后自行部署

错误处理

节点单位

若节点处理过程中发生异常,会自动路由到 FAILURE relation。
若 FAILURE 输出未连接任何节点,消息会被丢弃,仅留下错误日志。
触发器节点的 *_pattern 不匹配的消息会被标记为 SKIPPED,不传递给后续节点,也不计入处理/错误计数器。
所有错误均会显示在实时调试面板和执行历史画面。

流程单位

项目	默认值	说明
max_depth	100	处理一条消息期间累积的节点访问次数限制
max_revisit	3	同一节点的再访问次数限制(防止循环无限循环)
flow_timeout_ms	30,000	超出处理时间时强制终止

外部 IO 节点重试

HTTP·Kafka·外部数据库等外部 IO 节点可通过 retry_count / retry_delay_ms 选项在节点内部立即重试,所有重试失败时路由到 FAILURE relation。若需要更精细的退避或 EXHAUSTED 分支处理,请单独使用 flow_retry 节点。

运维诊断

管理员可在系统菜单中确认流程引擎的分发队列状态(工作者运行情况、待处理队列大小、累积处理/失败/丢弃计数、最后错误)。

服务器会在后台定期检查流程工作者池的负载,若负载持续一定时间以上,会在运维日志中记录一行状态变化(HEALTHY → DEGRADED → CRITICAL)。恢复正常后会再记录一行恢复日志。

权限

流程不引入单独的权限模型,而是原样使用现有系统认证/权限。

功能	所需权限
查看列表 / 查看执行历史	所有已认证用户
创建·编辑·部署·删除流程	ADMIN
导入/导出 / 全部重新部署	ADMIN

运维模式 (Recipes)

汇集常用连线形式的库。各模式可原样复制,作为新流程的起点使用。

模式 1 — 处理 + 通知分支

根据处理结果,SUCCESS 加载,FAILURE 通过通知进行双重分支。

[Action Node]
   ├── SUCCESS → [flow_log] / [flow_save_attributes] / ...
   └── FAILURE → [flow_send_email] / [flow_send_push]

模式 2 — 安全重试

构建退避 + EXHAUSTED 处理器,使外部 IO 能抵御临时性失败。

[risky_node] ─[FAILURE]→ [flow_retry] ─[SUCCESS]→  (risky_node 로 루프)
                                       └[EXHAUSTED]→ [에러 핸들러]

模式 3 — 用预过滤阻断负载

以触发器阶段的 *_pattern 提前过滤消息,减少后续处理量。未匹配模式会被标记为 SKIPPED,不计入计数器。

[flow_on_tag_point]    (옵션 tag_id_pattern: "MOTOR-*.SPEED")
   │
   ▼
[필터/변환/액션 ...]

模式 4 — 多重 case 分支

按状态/类型分成多条分支。

[flow_switch]   (case별 boolean 표현식)
   ├── case A → [...]
   ├── case B → [...]
   └── DEFAULT → [...]

模式 5 — 窗口累积 + 一次性 emit

将高频输入汇集一定时间,转换为一条消息。

[High-rate Trigger]
   │
   ▼
[flow_merge: window_ms=10000]   → data.merged 배열에 누적
   │
   ▼
[flow_script_transform: 요약]
   │
   ▼
[Action / 외부 발송]

模式 6 — 重试 + 达到重试上限后绕行

重试结束后,通过备用路径(其他 API、通知、数据库加载)绕行。

[Primary HTTP] ─[FAILURE]→ [flow_retry]
                            ├─[SUCCESS]    → (Primary HTTP)
                            └─[EXHAUSTED]  → [Backup HTTP] ─[FAILURE]→ [flow_log/Email]

模式 7 — 域变更后后续动作

将标签单位报警转换为上层资产单位,委托给按资产的处理。

[flow_on_tag_alarm]
   │
   ▼
[flow_change_originator: Tag → Asset]
   │
   ▼
[자산 단위 액션 (Create Work Order / Publish Asset Event 등)]

模式 8 — 结合节流 + 防抖

节流为每分钟1次以下 + 仅在5秒内无变化时才处理。

[High-rate Trigger]
   │
   ▼
[flow_throttle: max_msgs=1, window_ms=60000]
   │
   ▼ SUCCESS
[flow_debounce: window_ms=5000]
   │
   ▼
[Action]

模式 9 — 以子流程模块化公共逻辑

当多个入口点需要共享相同后续处理(校验·精炼·加载)时,分离为子流程。

메인1: [Trigger A] → [flow_subflow: target=FLOW_99]
메인2: [Trigger B] → [flow_subflow: target=FLOW_99]

서브 (FLOW_99): (트리거 없음)
[검증] → [정제] → [적재]

模式 10 — 直接触发绕行

若要将其他流程的结果流入下一流程的输入,通过 flow_dds_publish 发布到域通道,另一流程则以相同消息类型作为触发器接收。

플로우 A: [...] → [flow_dds_publish: type=ASSET_EVENT, originator=...]
플로우 B: [flow_on_asset_event] → [...]

应用示例(一览)

场景	结构
MES 集成	`flow_schedule` → `flow_http_request` → `flow_script_transform` → `flow_create_work_order`
事件外部中继	`flow_on_asset_event` → `flow_msg_type_filter` → `flow_mqtt_publish`
数据精炼·加载	`flow_on_tag_point` → `flow_script_transform` → `flow_save_tag_point`
报警自动化	`flow_on_tag_alarm` → `flow_script_filter` → `flow_send_email` + `flow_create_work_order`
外部数据库同步	`flow_jdbc_poll` → `flow_script_transform` → `flow_create_asset`
webhook 接收	`flow_on_webhook` → `flow_script_filter` → `flow_publish_asset_command`
报警波段自动调整	`flow_on_asset_aggregation` → `flow_script_transform` → `flow_update_tag_alarm_band_numeric`
作业指示状态自动化	`flow_on_asset_event` → `flow_switch` → `flow_start/end/pause/resume_work_order`
API 可靠性增强	`flow_http_request` ─FAILURE→ `flow_retry` ─SUCCESS→ 循环 / EXHAUSTED→ 通知
OEE 低下通知	`flow_on_oee_event` → `flow_script_filter` → `flow_template` → `flow_send_push`
Edge 批量注册	`flow_on_webhook` → `flow_edge_opc_create` → `flow_split` → `flow_edge_tag_create` → `flow_edge_opc_start`
高频精炼	`flow_throttle` → `flow_debounce` → 后续
多重事件合并	多重触发器 → `flow_merge` → 摘要转换 → 发送

分步调试指南

流程未按预期动作时,请遵循以下标准步骤。

步骤1 — 确认是否触发

在列表画面观察该流程的执行次数是否增加。

观察	含义·下一步操作
计数为0	触发器未触发。检查流程部署状态 + 触发器节点连线 + `*_pattern` 选项
计数增加但错误也一并增加	动作节点失败。转到第2步
计数增加但后续处理未执行	分支连线遗漏。检查 `FAILURE`/`THROTTLED`/`EXHAUSTED` 等所有输出是否处理

步骤2 — 用实时调试确认各节点流程

打开编辑画面,打开实时调试面板(2秒刷新)。

观察	含义
特定节点持续保持灰色	消息未到达 — 上一节点走了 `FAILURE` 分支或被过滤拦截
点亮红色 + ERROR 行	节点处理中异常。检查消息的 `data.error` 字段和 `/flow/log` 的 `NODE_ERROR` 事件
点亮绿色但下一节点灰色	输出 relation 标签不一致。确认连线标签是否与节点的输出 relation(SUCCESS/TRUE 等)精确一致

步骤3 — 用执行历史画面追踪消息

在 /flow/log 画面按消息 ID 追踪单条消息的整个流程。

在消息 ID 过滤中输入在实时调试中看到的 msg_id
按 FLOW_START → NODE_IN → NODE_OUT → FLOW_END 顺序按时间排列
若看到 NODE_ERROR,该节点的 error_message 即为原因

步骤4 — 检查节点设置

常见失误:

失误	检查项
`*_field` 路径拼写错误	是否为 `data.tag_id`?是 `metadata.tag_id` 吗?用实时调试的消息预览确认实际键
`${...}` 模板变量未替换	变量路径是否存在于消息中,是否有拼写错误
外部 IO 超时	增加 `timeout_ms`。外部系统本身的响应时间
权限·认证请求头	`headers` JSON 格式是否正确,令牌是否有效

步骤5 — 隔离测试

将有问题的节点单独移到新的临时流程中,通过测试运行触发一条消息,隔离验证结果。若正常动作,则原流程的连线·前一阶段消息形式值得怀疑。

步骤6 — 重置计数后重现

重置全部计数后,触发一条消息以清晰统计重现,问题会更明确。

常见问题

症状	原因·处理
触发器未触发	确认流程是否处于部署状态,触发器节点的 `_pattern` 是否过窄。计划/外部订阅节点请在全部重新部署*后重试
实时调试为空	确认消息计数器是否未增加 — 触发器模式未匹配(`SKIPPED`)不计入计数器。也确认面板上方的暂停开关是否打开
计数器数值异常累积	通过节点设置右上角重置全部计数按钮重置统计窗口
同一消息被反复处理	怀疑存在循环。`max_revisit` 默认仅允许3次同一节点的再访问。通过 `flow_subflow` 分离后用 `flow_throttle`/`flow_debounce` 调整输入量
外部系统调用间歇性失败	用 `retry_count`/`retry_delay_ms` 节点选项或 `flow_retry` 节点连线退避重试
导入后计划不动作	执行列表顶部的全部重新部署
Update 节点执行后部分字段保持原样	属预期行为。Update 节点仅更新输入字段,其余保持不变。全量更新请用 Delete + Create 组合
Create 节点因 NOT NULL 错误而失败	确认节点设置中的 `*_field` 动态选项路径是否实际存在于消息中。部分 NOT NULL 列会自动补全,但核心 ID(`asset_id`、`tag_id` 等)需直接填写
`flow_kafka_publish` / `flow_mqtt_publish` 未发布	检查外部 broker 连接信息(`bootstrap_servers`/`broker_url`)和主题权限。旁边连线 `flow_log` 确认发布前消息是否到达
`flow_http_request` 无响应	增加 `timeout_ms`(默认5000),本体在 `body_template` 中以 `${msg.data.x}` 形式直接明确。响应加载于 `data.response_status` / `data.response`
`flow_retry` 未重试	属连线错误情况。需要将 `flow_retry` 的 `SUCCESS` 输出重新连回原失败节点才会重试(参考下方模式2)
`flow_jdbc_poll` 等行每次都重新进入	轮询 SQL 的 `WHERE` 条件中需包含处理后状态更新(例:`WHERE sync_status='NEW'` + 同一流程末尾以 `flow_jdbc_query` 设置 `UPDATE ... sync_status='OK'`)
找不到报警直接触发节点	属预期排除。报警必须仅通过 `flow_create_alarm_config` 路径产生才能保持报警历史一致性
脚本节点因 `Java class` 访问错误而失败	脚本沙箱中被阻断。外部调用请单独连线外部集成节点
作业指示状态迁移节点仅出现 `FAILURE`	当前状态不是可迁移的起始状态。例如 `flow_pause_work_order` 仅在 `START` 状态下动作。请预先用 `flow_check_existence_field` / `flow_script_filter` 确认状态
测试运行虽然动作了但图形保持变更状态	测试运行始终触发已保存版本。如需验证更改,请先保存 → 测试运行
导入的流程处于非激活状态	属预期行为。审核后请自行打开部署开关

常见问题 (FAQ)

汇集了操作员首次使用流程时常问的问题。

Q. 一个流程能设置多个触发器吗? A. 可以。在同一流程中放置多个触发器节点均会成为入口点,各自独立触发。与 flow_merge 结合可将多种事件合并为一个后续处理。

Q. 可以创建没有触发器的流程吗? A. 可以。保存时会显示警告但仍可保存。可仅通过测试运行或其他流程的 flow_subflow 调用来触发,以"库"形式使用。

Q. 一条消息可以同时经过多个节点吗? A. 可以。若在一个节点的输出端口连接多条连线,同一消息会同时分支并传递给所有后续节点。

Q. 转换节点可以直接创建消息,将其变为其他消息吗? A. 可以。在 flow_script_transform 中可自由更改 msg.type、msg.originator、msg.data、msg.metadata。但即使让消息看起来像其他触发器类型,也不会自动调用其他流程(维持原本的触发器映射)。若要调用其他流程,请使用 flow_subflow 或 flow_dds_publish。

Q. 自动快照最多保留多少个?可以直接恢复吗? A. 每次图形保存都会自动加载,保留策略遵循运行环境设置。画面上不提供直接恢复的 UI,如有需要可请管理员协助恢复到特定版本。最安全的方法是在变更前用 내보내기 将 JSON 文件保存到外部。

Q. 执行历史保留多久? A. 默认7天。如需长期保留,请通过 flow_kafka_publish 或 flow_jdbc_query 加载到外部日志系统。

Q. 我只想重置某个流程的统计。 A. 请使用编辑画面节点设置右上角的重置错误计数(仅错误)或重置全部计数(全部)。不会影响其他流程。

Q. 想把流程迁移到其他环境(开发/预生产/生产)。 A. 用编辑画面的 내보내기 下载 JSON,在目标环境的列表画面通过 가져오기 上传即可。导入的流程始终以新 ID + 解除状态加载,可由操作员审核后自行部署。

Q. 同一载荷被处理了两次,该如何防止? A. 请在触发器阶段用 *_pattern 收窄,或用 flow_throttle/flow_debounce 限制频率。若为外部输入(webhook·MQTT),请让发送端带幂等键,并用 flow_check_existence_field 或基于消息 ID 的过滤器去重。

Q. 循环(loop)连线安全吗? A. 像 flow_retry 这样有意的循环是安全的。非预期的循环会被 max_revisit(默认3次)自动阻断。但运行中建议通过 flow_log 使流程可视化,以尽早发现非预期的循环。

Q. 流程更改会实时生效吗? A. 保存图形时立即生效。但对于计划·外部订阅·外部数据库轮询等自带调度器的触发器,请通过全部重新部署或流程解除→部署开关重新注册。

Q. 可以更改已注册节点的 ID 吗? A. 节点 ID 由系统自动分配。显示名可在节点设置表单中更改,实时调试·执行历史中使用的是显示名。

Q. 没有权限的用户会不小心变更流程吗? A. 创建·编辑·部署·删除流程需要 ADMIN 权限。普通用户只能查看列表和执行历史。

运维最佳实践

以下是在生产环境安全运维流程时应遵守的原则。

负载管理

请在触发器阶段用 *_pattern 提前过滤,减少后续处理量。每分钟数万件涌入的标签点位是成本最高的入口点。
调用外部系统(flow_http_request/flow_jdbc_query 等)时,请与 flow_throttle / flow_debounce 结合以控制外部负载。
若同一输入需要挂多个后续节点,请分离为 flow_subflow 以减少整体图形节点数。节点越多,实时调试轮询成本也越高。
仅在运维验证阶段打开调试模式。加载量越大,7天的执行历史保留窗口越快被填满。

安全编写

所有动作节点务必连线 FAILURE 分支(哪怕仅是 flow_log)。若 FAILURE 为空,消息会被静默丢弃,难以追踪事故。
外部 IO 节点尽可能与 flow_retry 结合,吸收临时性故障。
Update 节点仅更新输入字段(部分更新)。全量覆盖请用 Delete + Create 组合。
没有触发器节点的图形仅可用于手动测试运行。保存时请确认警告,以免意外遗漏触发器。
循环(返回同一节点的连线)只应以 flow_retry 等有意形式创建,其他情况下虽有 max_revisit 保护,但可疑图形请以 flow_log 可视化流程。

变更流程(Change Management)

变更运行中流程时的推荐顺序。

备份 — 用编辑画面的 내보내기 将当前图形下载为 JSON 文件保存(虽然自动快照也会加载,但外部保管更安全)。
复制或以新流程操作 — 与其立即修改运行中的流程,建议在新流程(或用导入创建的副本)中变更后再验证。
测试运行 — 直接注入 JSON 消息,依次触发所有分支(SUCCESS/FAILURE/EXHAUSTED 等),在执行历史中确认结果。
反映到生产 — 将验证过的流程图形 JSON 내보내기 → 在生产环境 가져오기 → 操作员审核后部署开关。
发生问题时回滚 — 立即用解除开关停用。由于自动快照已加载,可请开发者协助恢复到之前版本。

计数器·统计运用

若列表画面的处理趋势迷你图突然趋于平坦,请检查触发器未触发或 SKIPPED 处理的可能性。
若错误率环形图异常,请怀疑节点设置中的 *_field 路径。若消息载荷中没有该键会频繁失败。
运维验证后请用重置全部计数重置统计窗口,重新测量平时的基线。

紧急应对流程

运行中发生问题时应迅速使用的分步流程。

场景 1 — 特定流程暴增(消息骤增)

症状:某流程的执行次数相较正常值暴增数十~数百倍,错误也同时增加

处理

立即用解除开关停用该流程(列表画面1秒即可)
通过实时调试面板·执行历史确认哪个触发器引起暴增
收窄触发器节点的 *_pattern 选项,或紧随其后连线 flow_throttle / flow_debounce
如有需要,用 flow_check_existence_field 或 flow_msg_type_filter 限定消息类型
修改后用测试运行重现 → 确认正常后重新部署

场景 2 — 外部系统故障导致批量失败

症状:HTTP/Kafka/外部数据库等外部 IO 节点的错误同时累积

处理

若影响范围广,批量解除相关流程(列表画面批量选择后批量解除)
确认外部系统恢复
若外部 IO 节点没有 flow_retry 连线,请添加连线
若外部系统响应变慢,调整 timeout_ms
根据运行环境全部重新部署后重新部署

场景 3 — 循环导致的无限循环

症状:单条消息持续经过同一节点,处理时间累积

处理

虽然 max_revisit 保护最多只允许3次再访问后自动阻断,但为运维安全请先解除再检查
在图形中可视化追踪循环(编辑画面中沿连线追踪)
若为有意循环(flow_retry),确认 max_attempts 是否合适
若为非预期循环,请删除连线或以 flow_check_relation 添加分支
修改后重置全部计数 → 重新部署

场景 4 — 怀疑数据损坏(自动更新错误)

症状:自动化导致域数据以非预期方式被更新

处理

立即解除该流程
从外部保管的先前图形 JSON 备份或自动快照中确认先前版本(需管理员协助)
分析图形:确认是否存在非预期的 Update 节点连线·错误的 *_field 路径·脚本转换错误
受影响的域数据以单独流程在后台画面进行更正
修改后的图形通过测试运行验证后重新部署

场景 5 — 系统检查·维护期间暂停

症状:外部系统检查期间希望暂时停止外部 IO 调用

处理

批量解除受影响的流程(列表中多选)
检查结束后重新部署
自带调度器的触发器(计划·外部 MQTT 订阅·外部数据库轮询)请执行一次全部重新部署以确认正常注册

场景 6 — 实时调试加载队列已满

症状:运维诊断页面的加载队列大小接近临界值,丢弃计数增加

处理

减少打开调试模式的流程数和加载量(验证结束的流程请关闭调试模式)
用触发器阶段的 *_pattern 减少后续处理量本身
若系统进入自动恢复模式,负载看门狗会留下一行 DEGRADED/CRITICAL 日志 — 请与管理员共享

所有场景中,优先顺序为 立即解除 → 查明原因 → 修改 → 验证 → 重新部署。也请参考变更流程(变更安全流程)。

安全·敏感信息处理

流程涉及外部 API 调用·邮件/SMS 发送·webhook 接收等敏感输入输出,请遵守以下原则。

令牌·凭据

请勿在节点设置中直接以明文输入 API 令牌·密码。请使用 ${ENV_VAR} 形式的环境变量替换,从运行环境的密钥存储库注入。
为在运行环境中不暴露令牌,请尽可能将 headers JSON 中的令牌位置分离为环境变量。

// 권장
{ "Authorization": "Bearer ${MES_TOKEN}" }

// 비권장
{ "Authorization": "Bearer eyJhbGciOi..." }

保护 webhook 入口点

flow_on_webhook 虽是已认证用户调用的内部入口点,但若要向外部暴露,请务必连线载荷验证过滤器。

// flow_script_filter — 토큰 + 필수 필드 검사
if (data.token !== '${WEBHOOK_TOKEN}') return false;
if (!data.asset_id || !data.cmd_key)   return false;
true

敏感数据掩码

实时调试面板和执行历史会显示消息预览。个人信息(姓名·联系方式·账户)和令牌请在加载前进行掩码。

// 디버그/로그 적재 직전 변환
data.email_masked = data.email
  ? data.email.replace(/(.{2}).+(@.+)/, '$1***$2') : null;
delete data.email;
delete data.token;
msg

外部 IO 的响应本体处理

flow_http_request 的 data.response 会加载完整响应本体。若响应中包含敏感数据,请立即用后续转换节点仅提取所需键并删除原始数据。

// 응답에서 필요한 필드만 보존
data = { id: data.response_obj.id, status: data.response_obj.status };
msg

脚本节点隔离

脚本节点在安全沙箱中执行,文件·网络·任意类访问均被阻断。若需要外部调用,请务必单独连线外部集成节点。

流程指标与报警

流程运行时自动采集的指标,以及查看这些指标的地方·设置报警的方法。

流程单位 KPI

在列表画面(/flow/index)的各行会显示以下 KPI。

指标	含义	异常信号
总执行次数	触发器触发并通过图形一次的次数(成功/失败合计)	相较平时骤减 → 触发器已死 / 骤增 → 暴增
错误次数	图形中任一节点走 `FAILURE` 分支的次数	若占比超过总数5%需要检查
错误率	`에러 / 전체 × 100`	超过一定临界值请设置报警
最后执行	最近一次触发时刻	仅当"5分钟以上无触发"是正常情况时才可视为 OK
平均处理时间	一条消息通过整个图形的平均 ms	在增删外部 IO 节点时变化较大
最近处理时间趋势	5分钟迷你图 — 实时调试面板	出现 spike 时检查外部系统响应

节点单位 KPI

在编辑画面点击节点后,节点右下角显示以下内容。

[Node Name]
처리 999 · 에러 3 · 평균 12ms · 최근 18ms

位置	显示
上方点灯	灰色(待机) / 绿色(处理中) / 红色(错误)
下方标签	`처리 N · 에러 M · 평균 Xms · 최근 Yms`

节点统计4种

计数器	含义
处理	进入该节点并以正常分支输出的消息数
错误	`FAILURE` 分支或异常发生次数
平均处理时间	节点自身处理 ms(含外部 IO)
最近处理时间	最后一条处理的 ms

系统单位指标

流程引擎整体的指标可在系统 → 监控画面确认。

指标在 JMX 域 plantpulse.core.engine 下以以下五个指标暴露。

指标	种类	含义	查看方法
`FLOW_EXECUTOR_QUEUE_SIZE`	Gauge	流程执行池的待处理队列大小	若持续堆积,说明处理速度跟不上进入速度
`FLOW_EXECUTOR_ACTIVE_COUNT`	Gauge	执行池的活动线程数	若贴近池大小则表示饱和
`FLOW_EXECUTOR_DROPPED_COUNT`	Gauge (累积)	因池饱和而丢弃的任务累积数	不为0则说明消息丢失 — 应最先查看的值
`FLOW_DEBUG_QUEUE_SIZE`	Gauge	调试事件加载待处理队列大小	打开调试模式的流程越多,该值越大
`FLOW_EXECUTION_TIME`	Timer	每个流程的处理时间分布	追踪平均·最大值

没有名为 flow.engine.* 的指标

旧文档曾以表格刊载 flow.engine.queue_depth · in_flight · dispatch_lag_ms · exec_p95_ms failed_per_min · script_timeout_per_min,但不存在这些名称的指标。 以上五个即为全部。

(flow.engine.enabled 不是指标,而是 engine.properties 的设置键 — 是关闭整个流程引擎的主开关。)

基于指标的报警注册模式

用高级报警或CEP → 触发器监视流程自身指标的4种模式。

模式 A — 错误率超过临界值

context EVERY_1_MINUTES
SELECT flow_id, count(CASE WHEN status='FAILURE' THEN 1 END) * 100.0 / count(*) AS err_pct
FROM   AssetEvent.win:time(5 min)
WHERE  event_type = 'FLOW_EXEC'
GROUP  BY flow_id
HAVING err_pct > 5

模式 B — 处理队列产生背压

当 flow.engine.queue_depth > 500 持续1分钟以上时报警 — 触发器触发速度超过处理速度的情况。

模式 C — 特定流程的触发中断

SELECT * FROM pattern [
  every a = AssetEvent(event_type='FLOW_EXEC', flow_id='my-flow')
        -> ( timer:interval(15 min)
             and not AssetEvent(event_type='FLOW_EXEC', flow_id=a.flow_id) )
]

若原本正常每分钟触发N件的流程沉默15分钟以上,则报警。

模式 D — 外部 IO 超时暴增

context EVERY_5_MINUTES
SELECT flow_id, node_id, count(*) AS timeout_count
FROM   Log.win:time(5 min)
WHERE  module = 'flow-engine' AND code = 'NODE_TIMEOUT'
GROUP  BY flow_id, node_id
HAVING count(*) > 10

建议的报警输出渠道

报警种类	建议渠道
错误率·触发中断(运维直接)	邮件 + 推送通知
队列背压·超时暴增(系统)	Slack/Teams webhook
自动诊断(参考)	仅诊断日志

⚠️ 用于报警的流程本身绝不要依赖它自己触发的报警指标 — 若报警流程本身宕机,报警本身也不会来。用于报警的流程请通过系统 → 监控的外部健康检查监视。

消息处理语义与背压

流程引擎处理消息时的保证级别与背压(backpressure)动作。

传递保证 — At-Least-Once

流程引擎保证at-least-once 传递。

情况	动作
正常处理	触发一次 → 通过图形一次 → 完成一次
处理中引擎重启	触发器队列中剩余的消息会在下次启动后重新处理
节点单位异常	仅转到 `FAILURE` 分支。消息不会消失
外部 IO 超时	可通过 `FAILURE` 分支 + `flow_retry` 自动重试

可能重复 — 不是恰好一次(exactly-once)。若 flow_create_* 中途中断后重试,同一 ID 可能进入两次,请使用 on_duplicate=skip 或外部系统的幂等键。

幂等键模式

流程调用外部系统时保证幂等性的模式。

// flow_script_transform — 멱등성 키 생성
msg.data.idempotency_key =
  msg.metadata.asset_id + '|' +
  msg.metadata.ts + '|' +
  msg.data.event_type;

之后在外部调用时以请求头附加:

"headers": { "Idempotency-Key": "${data.idempotency_key}" }

处理顺序 — 触发器单位 FIFO

同一 originator 的消息	不同 originator
触发器单位 FIFO(按到达顺序)	并行处理(顺序无关)

即使资产 A 的2件事件几乎同时发生,A 的事件仍按发生顺序处理。资产 A 与资产 B 的事件可能在不同工作者中并行处理。

背压(Backpressure)

处理速度跟不上触发速度时的动作。

情况	动作
队列深度 < 80%	正常 — 新触发立即 enqueue
队列深度 80%~100%	产生诊断 WARN 日志 — 队列继续接收
队列深度 = 100%(已满)	新触发器被丢弃 + 产生诊断 ERROR 日志

队列已满时操作员应做的事

在系统 → 监控确认 flow.engine.queue_depth
在列表画面中识别执行次数暴增的流程
收窄该流程的触发器预过滤(*_pattern)以降低负载
或暂时解除该流程的部署以应急处理

断路器(外部 IO)

外部 IO 节点(HTTP/Kafka/MQTT/邮件)在满足以下条件时会被批量阻断30秒。

条件	临界
最近1分钟内连续失败	≥ 10件
平均响应时间	≥ 10秒

阻断期间进入该节点的所有消息立即分支到 FAILURE。外部系统响应恢复后自动解除阻断。此动作的作用是隔离外部故障,使其不至瘫痪流程引擎本身。

断路器开启的时刻会记录在诊断日志的 code = CIRCUIT_OPENED 中。若外部系统很快恢复,可查看执行历史的 metadata.retry_count,手动重新执行遗漏的消息。

循环防止

流程中如果产生以下循环,可能发生无限循环。

A → B → C → A  (잘못된 결선)

流程引擎通过两道防线阻断循环。

防线	动作
`max_depth=100`	经过100个节点则强制终止
`max_revisit=3`	第4次访问同一节点的瞬间丢弃消息 + 诊断 ERROR

flow_retry 的 SUCCESS 分支 → 回到原节点的循环是有意循环,因 metadata.retry_count 一并增加,在触发 max_revisit 之前会正常终止。

从执行历史中能看到什么

旧文档的"错误代码目录"并不真实存在

NODE_SCRIPT_TIMEOUT · FLOW_TIMEOUT · FLOW_MAX_DEPTH · TRIGGER_QUEUE_FULL WEBHOOK_AUTH_FAIL · DOMAIN_DUP_KEY 等代码曾以表格刊载,但**产品的任何地方都不会产生该字符串。**若以该代码在日志中搜索,永远是0件。

流程引擎实际留下的是以下5种事件类型。

流程执行历史(mm_flow_log,左侧菜单 Automation > 流程执行历史)中加载的事件有五种。

事件类型	何时	加载条件
`FLOW_START`	进入流程时	始终
`FLOW_END`	流程处理结束(成功·超时均相同)	始终
`NODE_IN`	节点接收消息时	仅调试模式时
`NODE_OUT`	节点发送消息时	仅调试模式时
`NODE_ERROR`	节点处理中异常	始终

各行共同拥有的字段。

字段	内容
`level`	`INFO` / `WARN` / `ERROR`
`flow_id` · `flow_node_id` · `node_name` · `node_type`	属于哪个流程的哪个节点
`msg_id`	消息单位追踪用 — 若要衔接单条消息的流程需以该值绑定
`message` · `error_message`	人类可读的说明,异常消息
`duration_ns`	节点执行时间(纳秒) — 仅在 `NODE_OUT` · `NODE_ERROR` 中有意义

平时不会留下节点单位日志

NODE_IN · NODE_OUT 仅在调试模式下加载。平时执行历史中仅显示 FLOW_START · FLOW_END 属正常。若要逐一跟踪节点,请打开该流程的调试模式 — 但日志量会大幅增加。

从症状出发查找

从症状出发而非代码。

症状	首先查看之处
触发器虽已触发但似乎没有节点运行	流程是否已部署(开关ON) · 触发器的 `*_pattern` 是否正在过滤该消息
处理中途中断	一条消息超过30秒(`flow_timeout_ms`)。查看服务器日志的 `FlowExecutor timeout` 警告确认
同一消息反复处理	循环。以 `max_revisit`(3次)自动阻断,但请检查连线
图形过长导致无法结束	超过 `max_depth`(100)。分割为子流程
节点抛出异常	在执行历史中查看 `NODE_ERROR` 行的 `error_message`
外部调用失败	在后续节点中确认该节点的 `data.response_status` · `data.error`

限度值(100 · 3 · 30,000ms)是 ExecutionState 的默认值。

集群·高可用(HA)动作

若平台以集群环境安装,流程引擎按以下规则分布式动作。

节点角色分离

角色	动作
领导者(Leader)	串行处理流程图形变更(保存/部署/解除)的单一节点
工作者(Worker)	并行执行触发器触发·消息处理的普通节点(所有节点都兼任工作者角色)
调度器(Scheduler)	负责 `flow_schedule` cron 评估 — 与领导者相同节点

领导者节点在平台启动时自动选举,若领导者宕机,其他节点之一会自动晋升为领导者。操作员无需直接指定。

消息分发 — 资产单位一致路由

分发键	动作
`originator.id`(资产/标签/订单 ID)	同一资产的消息始终路由到同一工作者 — 保证顺序
外部接入(`flow_on_webhook` 等)	轮询分发

此规则可防止资产 A 的事件在多个工作者中同时处理而导致顺序混乱。结果是,资产单位保证单一工作者处理,资产之间实现并行处理。

领导者故障时的动作 — 故障转移

[Leader 다운 t=0]
        ↓
[다른 노드가 리더 승격 시도 t=0~3s]
        ↓
[새 리더 확정 t=3~5s]  ← cron 스케줄·플로우 배포 변경 재개
        ↓
[기존 워커들은 정상 동작 유지 — 트리거 처리 영향 없음]

阶段	影响
0~3 秒	`flow_schedule` 触发暂停 / 触发器处理不受影响
3~5 秒	确定新领导者,重启计划
5 秒之后	正常

cron 触发根据**misfire 策略立即重新执行一次遗漏的触发**。运行中若 flow_schedule 的评估单位小于5秒,可能出现短暂遗漏。

触发器队列的持久性

项目	动作
触发器队列位置	内存队列 + 永久存储(事务日志)
节点重启时	队列中剩余的未处理消息会在下次启动后重新分发
处理中节点宕机	该消息会被重新处理,但需借助at-least-once 保证在外部系统中使用幂等键

集群部署时操作员检查清单

所有节点的时间同步(NTP) — 使触发时刻在各节点间一致
将外部系统(MQTT/Kafka broker)部署在所有节点均可到达的网络位置
flow_send_email 的 SMTP 设置只需在系统设置中注册一次 — 所有节点共享
工作者池大小(flow.engine.workers)按每个节点的 CPU 核心数调整

单节点模式(开发·小规模)

领导者·工作者·调度器均在同一进程中动作
无故障转移 — 若节点宕机,流程本身会中断
触发器队列以内存 + 磁盘保障,重启后可恢复

端到端追踪

事后追踪一条消息如何通过整个图形的方法。

自动分配相关 ID

流程引擎会为所有触发触发的消息自动分配相关 ID(correlation ID)。

{
  "type": "POST_TELEMETRY",
  ...,
  "metadata": {
    "trace_id":   "tr-a1b2c3d4-e5f6-7890-...",
    "span_id":    "sp-01",
    "parent_id":  null,
    ...
  }
}

字段	含义
`metadata.trace_id`	一次触发全程唯一的 ID — 图形中所有节点通过的消息共享
`metadata.span_id`	每个节点唯一的 ID — 每经过一个节点都会更新
`metadata.parent_id`	前一个节点的 `span_id`

在执行历史画面按 trace_id 搜索

在执行历史画面的 trace_id 输入框中粘贴 ID,会显示该消息经过的所有节点的时间顺序日志。

🔍 trace_id = tr-a1b2c3d4-...

[12:34:56.123] [trg]      flow_on_tag_point      태그=MOTOR.TEMP, value=87
[12:34:56.125] [filter]   flow_script_filter     score>80 → TRUE
[12:34:56.126] [transform] flow_change_originator Tag → Asset
[12:34:56.130] [action]   flow_create_work_order WO-20260513-001 생성
[12:34:56.241] [external] flow_send_email        admin@... 발송 성공

调用子流程时 trace 传播

用 flow_subflow 节点调用其他流程时,相同的 trace_id 会原样延续。即主流程 + 被调用子流程的所有节点日志都可以用同一 trace 搜索。

传播到外部系统

外部 HTTP 调用时会自动附加 X-Trace-Id 请求头。

GET /api/orders HTTP/1.1
Host: erp.example.com
X-Trace-Id: tr-a1b2c3d4-e5f6-...
X-Span-Id:  sp-04

若外部系统接收该请求头并一并记录在日志中,可用同一 ID 匹配两边系统的日志。

在报警·邮件中暴露 trace_id

若在报警本体或邮件模板中包含 ${metadata.trace_id},操作员收到报警后可立即在历史画面中追踪该事件。

제목: [긴급] 모터 과열 — ${metadata.asset_id}
본문:
시각: ${metadata.ts}
값: ${data.value}°C
trace: ${metadata.trace_id}
이력 보기: https://platform.example.com/flow/log?trace_id=${metadata.trace_id}

追踪保留期

数据	保留期
trace_id 与各节点 span 日志	7天(与执行历史相同)
外部加载(长期保管)	使用审计·历史追踪的外部系统镜像

若 trace_id 过长导致消息大小负担,可通过 flow_script_transform 仅暴露最后4位(如 tr-...d4)的简短形式,用于报警·邮件。但搜索时需要完整 ID。

图形模式目录

流程图形中常用的连线模式10种。

模式 1 — Pipeline(简单串行)

[trigger] → [filter] → [transform] → [action]

模式 2 — Fan-out(1对N)

                     ┌─→ [action 1]
[trigger] → [t] ─────┼─→ [action 2]
                     └─→ [action 3]

同一消息的副本会分发到多个节点。各分支独立处理,一个分支失败不影响其他分支。

模式 3 — Fan-in(N对1) — Merge

[trigger A] ──┐
[trigger B] ──┼─→ [flow_merge] → [aggregator] → [action]
[trigger C] ──┘

模式 4 — Scatter-Gather(分散 → 收集)

[trigger] → [split] ─┬─→ [process] ──┐
                     ├─→ [process] ──┼─→ [merge] → [action]
                     └─→ [process] ──┘

模式 5 — Switch(条件分支)

                            ┌─[CRITICAL]→ [긴급 알람]
[trigger] → [flow_switch] ──┼─[WARN]    → [경고 알람]
                            └─[NORMAL]  → [통과]

模式 6 — Retry with Fallback

[risky] ─[FAILURE]─→ [flow_retry] ─[SUCCESS]─→ (다시 risky)
                                  └[EXHAUSTED]→ [fallback action]

模式 7 — Circuit Breaker(利用断路器)

[trigger] → [throttle] → [external_io] ─[SUCCESS]─→ [save]
                                       └[FAILURE]─→ [log only]

模式 8 — Dead Letter Queue (DLQ)

[main flow] ─[FAILURE]─→ [flow_save_attributes] → 별도 자산에 적재
                                                  ↑
                              운영자가 주기적 점검 후 수동 재처리

模式 9 — Sliding Window Aggregation

[trigger] → [flow_throttle 60s] → [transform: 누적] → [action]
                                  (마지막 60건만 유지)

模式 10 — Saga(多阶段事务)

[start] → [step1] ─OK→ [step2] ─OK→ [step3] ─OK→ [complete]
              │            │            │
              └─FAIL→[rollback1]        │
                           │            │
                           ┌─FAIL→[rollback1+2]
                                        │
                                        └─FAIL→[rollback1+2+3]

模式选择指南

需求	建议模式
简单阈值报警	Pipeline (1)
一个事件 → 多个渠道	Fan-out (2)
多个事件 → 一份摘要	Fan-in / Merge (3)
数组批量处理	Scatter-Gather (4)
分支处理	Switch (5)
外部 API 可靠性	Retry (6)
外部系统保护	Circuit (7)
失败消息保管	DLQ (8)
最近N件累积决策	Sliding (9)
多个外部系统一致性	Saga (10)

流程测试最佳实践

安全变更·部署流程的测试策略。

3阶段测试 — 单元 → 集成 → 模拟

阶段	工具	验证对象
① 单元	编辑画面 ▶ 测试运行	单个节点独立动作(脚本式、外部调用响应)
② 集成	同一画面,临时载荷 + 实时调试 ON	整个图形序列·分支
③ 模拟	部署 + 等待实际触发器(预生产环境)	真实消息流·与外部系统的结合

测试运行载荷库

可粘贴到测试运行对话框中的各触发器载荷示例请参考各触发器载荷示例章节。请提前准备运行环境中经常发生的边界情况。

边界情况示例

情况	载荷
null 字段	`{"value": null}` — 脚本是否对 null 安全
空字符串	`{"value": ""}` — 验证是否将空字符串解释为0
负数	`{"value": -1}` — 阈值检查是否按 ± 符号意图工作
巨大数字	`{"value": 1e20}` — 溢出/精度
Unicode	`{"name": "한글-Émoji-🚀"}` — 外部系统编码

变更安全流程

기존 플로우를 [내보내기] (JSON 파일 저장)
새 플로우를 사본으로 만듦 (이름 끝에 `_v2`)
사본의 트리거 패턴을 좁혀 일부 자산만 매칭 (예: TEST-* 사이트만)
신구 동시 배포 — 새 버전 데이터 확인
1주일 안정성 검증 후 신 버전을 전체 트리거 패턴으로 변경
구 버전 해제 + 보관

保管回归测试场景

建议以文本文件保管常用场景,并在变更后务必重新执行。

# regression-tests/alarm-to-workorder.json
{
  "case": "고온 알람 → 워크오더 자동 생성",
  "input": {
    "type": "TAG_ALARM",
    "data": { "alarm_band": "HI_HI", "value": 95.0 }, ...
  },
  "expected": {
    "domain_changes": ["WorkOrder.CREATED"],
    "notifications": ["email:admin@example.com"]
  }
}

外部系统模拟(Mock)模式

在预生产环境模拟外部系统响应:

方法	说明
将 flow_http_request 的 url 改为模拟服务器	将生产 URL 分离为预生产 URL 变量
变更 flow_jdbc_poll 的 datasource	仅将生产 DB → 预生产 DB
将 flow_send_email 的 to_field 改为 fake@example.com	防止误发

重置计数器后进行压力测试

对新版本流程重置计数(全部)
让其正常触发5分钟
在列表画面确认处理/错误计数、平均处理时间
若 p95 处理时间较原来增加20%以上,请分析原因并考虑回滚

性能极限与调优

流程引擎的处理极限与调优提示。

默认限制

项目	默认值	备注
一条消息的节点访问数(`max_depth`)	100	若经过100个节点仍未结束则强制中断
同一节点再访问次数(`max_revisit`)	3	防止循环无限循环
流程处理时间(`flow_timeout_ms`)	30,000ms	一条消息处理超过30秒则强制终止
外部 IO 节点超时	5,000ms(`timeout_ms`)	HTTP/Kafka/MQTT 等
脚本执行超时	500ms(`timeout_ms`)	节点单位
执行历史保留	7天	之后自动过期

常用窗口大小建议

节点	建议窗口	备注
`flow_throttle`	1,000~60,000ms	依据外部系统 API 限度调整
`flow_debounce`	500~5,000ms	仅想传递稳定值时
`flow_merge`	5,000~60,000ms	太短会碎片化,太长会增加延迟
`flow_retry` 退避	起始1,000ms × 2倍	若重试5次即1·2·4·8·16秒

提高吞吐量的方法

触发器预过滤 — 用 *_pattern 仅让必要的消息进入(最有效)
以子流程简化图形 — 主图形只负责分支·路由,重负载处理交给子流程
将外部 IO 异步连线 — 用 flow_delay / flow_throttle 平滑外部 API 负载
仅在验证阶段打开调试模式 — 运维稳定后关闭调试模式
仅连线所需类别 — 不要将 Edge·外部集成等重负载节点连接到所有分支

消息大小

消息的 data / metadata 会被 JSON 序列化并加载到执行历史。对于巨大载荷(数MB以上的响应本体等),请尽量用转换节点提取所需的键。在历史保留·调试显示·子流程调用中,消息大小都与处理成本成正比。

审计·历史追踪

追踪流程变更·执行历史时应确认的位置。

图形变更历史

自动快照 — 每次保存图形都会按版本单位加载。如需恢复到之前状态,请请管理员协助。
变更者/变更时刻 — 列表画面的最终修改列显示最近保存时刻。变更者以系统认证用户为基准记录。
建议记录变更原因 — 建议在流程元信息的描述字段中一并记录变更原因·负责人·相关工单号,以提高可追溯性。

执行事件追踪

消息 ID 追踪 — 用执行历史画面的消息 ID过滤器,可按时间顺序查询单条消息的整个流程(FLOW_START → 所有 NODE_IN/NODE_OUT → FLOW_END)。
CSV 导出 — 用执行历史画面的CSV 下载按钮导出当前查询结果,可传递给外部审计系统。

域变更历史

流程的动作节点(flow_create_* / flow_update_* / flow_delete_*)所执行的域变更均委托给同一域服务,因此都会一并记录在后台画面的按域变更历史中。作业者 ID 以 insert_user_id="flow" 等表示,可与普通用户变更区分。

通过外部加载进行长期保管

若需要超出执行历史保留期(7天)的长期保管,请创建单独流程,将核心事件加载到外部系统(Kafka·外部数据库等)。

[flow_on_tag_alarm]
   │
   ▼
[flow_kafka_publish: topic=audit.alarm.events]

新建流程部署清单

在生产环境部署新流程前应确认的项目。

图形结构

是否精确连接了一个(或有意的多个)触发器节点
所有动作节点的 FAILURE 输出是否已处理(至少 flow_log)
若存在循环,是否为像 flow_retry 等有意的连线,是否处于 max_revisit 保护之内
外部 IO 节点是否已连线 retry_count/retry_delay_ms 或 flow_retry

节点设置

触发器的 *_pattern 是否过窄或过宽
*_field 动态选项路径是否实际存在于消息中
Create 节点的 NOT NULL 字段是否全部填充(除自动补全外)
Update 节点是否确认是有意进行部分更新

验证·测试

用正常载荷通过一次 SUCCESS 分支
用异常载荷(必填字段缺失等)通过 FAILURE 分支
用外部 IO 失败情况通过 Retry → EXHAUSTED 分支(如适用)
实时调试面板中的 INFO/ERROR 显示是否符合预期
/flow/log 画面中是否记录了所有阶段

运维安全

备份(图形 JSON 导出)是否已在外部保管
变更原因·负责人是否记录在流程描述中
若有通知(Email/SMS/Push)连线,收件人是否已验证
是否有部署后5~10分钟的执行历史监控计划

部署策略 — Canary / Blue-Green / A·B 测试

安全上线新流程的3种策略。所有策略均可仅用平台的基本功能(触发器模式·消息 dispatch·实时调试)实现。

策略 1 — Canary(仅部分资产逐步应用)

先将新流程仅应用于部分资产,观察一段时间后再扩展至全部。

[1단계 출시]
  새 플로우 v2 — 트리거 패턴: asset_id LIKE 'LINE-1.%'  (1개 라인만)
  기존 플로우 v1 — 트리거 패턴: asset_id LIKE 'LINE-2.%' OR 'LINE-3.%' OR ...

[2단계 확대 (1주일 후 안정 확인)]
  v2 패턴: 'LINE-1.%' OR 'LINE-2.%'
  v1 패턴: 'LINE-3.%' OR 'LINE-4.%'

[3단계 전체 (2주 후)]
  v2 패턴: '%'   ← 모든 자산
  v1 해제·보관

Canary 进展清单

期间	监控项
第1天	错误率 < 1% / 平均处理时间在原有 ±20% 以内
1周	外部系统集成100%正常 / 报警发生频率适当
2周	累积统计 / 用户反馈 / 决定下一产线扩展

Canary 产线请选择即使运行中也影响较小的产线(检修频率高或仅在夜间运行的产线等)。

策略 2 — Blue-Green(新旧同时运行后立即切换)

[Blue (현재)]                       [Green (새 버전)]
플로우 v1 — 배포됨                  플로우 v2 — 배포 + 격리된 originator
모든 트리거 처리                    metadata.test_mode=true 인 메시지만 처리

[전환 결정 시점]
   v2 트리거 패턴을 v1 과 동일하게 변경 (1초)
   v1 해제 토글 (1초)

Blue-Green 的核心是两个版本同时部署状态下立即切换 — 发现问题时立即重新部署 v1 进行回滚。

v2 隔离连线模式

[trigger 모든 메시지]
   ↓
[flow_script_filter — metadata.test_mode === true]
   ↓ TRUE
[새 로직]

测试用消息通过 /flow/{id}/run API 明确指定 metadata.test_mode=true 发送。

策略 3 — A/B 测试(性能·结果比较)

两个版本接收相同消息后执行不同动作,再比较结果。

[trigger 메시지]
   ↓
[flow_split (메시지 복제)]
   ├ A 경로 → 기존 v1 액션 → [flow_save_attributes target=stats_v1]
   └ B 경로 → 새 v2 액션  → [flow_save_attributes target=stats_v2]

之后在每日统计画面比较 stats_v1 vs stats_v2 累积结果。

A/B 比较自动分析

-- EQL 으로 두 버전 비교 (예: 알람 생성 누적)
context EVERY_1_HOURS
SELECT
  count(CASE WHEN metadata.flow_version='v1' THEN 1 END) AS v1_count,
  count(CASE WHEN metadata.flow_version='v2' THEN 1 END) AS v2_count
FROM AssetAlarm.win:time(1 hour)

策略选择指南

情况	建议策略
首次上线新自动化场景	Canary — 在一个产线验证后扩展
现有逻辑大幅变更(结构改编)	Blue-Green — 可立即回滚
测量两种算法哪个更好	A/B 测试
简单选项值调整	直接变更 — `metadata.audit_diff` 备注后监控1周

回滚流程(通用)

发现问题时立即回滚:

在列表画面中解除新版本的开关
(Canary/A·B 情况)将触发器模式改为0件匹配
监控原版本单独运行5分钟
确认诊断日志中没有 code=FLOW_NOT_DEPLOYED
分析原因 — 在执行历史中用 trace_id 追踪失败消息

上线前检查清单(压缩版)

新建流程部署清单的核心一览:

[ ] 테스트 실행으로 정상·경계·실패 시나리오 모두 통과
[ ] 외부 IO 노드에 timeout_ms / 재시도 정책 설정됨
[ ] 자격 증명은 ${creds.*} 참조 (평문 미포함)
[ ] 트리거 패턴이 의도한 자산만 매칭
[ ] 영향받는 도메인 (자산/태그/주문) 식별됨
[ ] 운영 시간(특히 야간) 영향 검토됨
[ ] 롤백 시점·기준·담당자 결정됨
[ ] [감사 로그](#审计历史追踪) 에 변경 의도 메모 작성됨

节点快速设置参考

汇集运维中常用节点核心设置的速查表。

触发器快速设置

节点	核心选项
`flow_schedule`	`cron` (例:`0 /5 * * ?` = 每5分钟), 或 `interval_ms`
`flow_on_webhook`	无选项 — 从外部以 `POST /webhook/flow/{flow_id}` 触发
`flow_on_mqtt_subscribe`	`broker_url`, `topic`, `client_id`, `username`/`password`
`flow_jdbc_poll`	`dsn`, `sql`, `interval_ms`
`flow_on_*` (域)	`_pattern` (glob — `MOTOR-`, `SITE-?` 等)

转换快速设置

节点	核心选项
`flow_script_transform`	`language` (`JS`/`EQL`), `script`, `timeout_ms` (默认500)
`flow_change_originator`	`entity_type`, `id_field`
`flow_rename_keys`	`mapping` (例:`{"old":"new"}`)
`flow_template`	`template` (`${data.x}` / `${metadata.y}` 替换)
`flow_split`	`path` (数组位置,默认 `data`)
`flow_to_email`	`subject_template`, `body_template`

流程控制快速设置

节点	核心选项
`flow_delay`	`delay_ms`
`flow_throttle`	`max_msgs`, `window_ms`, (分支:SUCCESS/THROTTLED)
`flow_debounce`	`window_ms`
`flow_merge`	`window_ms` (累积于 data.merged 数组)
`flow_subflow`	`target_flow_id`
`flow_retry`	`max_attempts` (默认3), `backoff_ms` (默认1000), `backoff_multiplier` (默认2.0)
`flow_log`	`level` (INFO/WARN/ERROR), `prefix`
`flow_noop`	(无选项)

外部集成快速设置

节点	静态选项	动态选项(`*_field`)
`flow_http_request`	`method`, `url`, `headers`, `body_template`, `timeout_ms`, `retry_count`, `retry_delay_ms`	`url_field`, `method_field`, `body_field`
`flow_kafka_publish`	`bootstrap_servers`, `topic`, `value_template`, `headers`	`topic_field`, `key_field`
`flow_mqtt_publish`	`broker_url`, `topic`, `qos`	`topic_field`
`flow_webhook_callback`	`url`, `method`, `headers`	`url_field`
`flow_send_email`	`to`, `cc`, `subject`, `body`	`to_field`, `cc_field`, `subject_field`, `body_field`
`flow_send_sms`	`to`, `text`	`to_field`, `text_field`
`flow_send_push`	`title`, `body`	`title_field`, `body_field`
`flow_jdbc_query`	`dsn`, `sql`, `params_field`	—

动作 — 集成/存储快速设置

节点	核心选项
`flow_save_tag_point`	`tag_id_field` (默认 `metadata.tag_id`), `value_field`, `timestamp_field`
`flow_save_attributes`	`entity_type_field`, `id_field`, `attributes_field`
`flow_dds_publish`	`type`, `originator_field`, `payload_field`
`flow_publish_asset_event`	`asset_id_field`, `event_type_field`, `severity_field`, `details_field`
`flow_publish_asset_command`	`asset_id_field`, `tag_id`, `cmd_key_field`, `payload_field`

域 CRUD 快速设置

节点	核心选项
`flow_create_*`	按域字段 + `*_field` 动态选项。NOT NULL 字段部分自动补全(自动补全项目请参考域 CRUD表)
`flow_update_*`	部分更新:仅更新输入字段,忽略空值。全量覆盖用 Delete + Create 组合
`flow_delete_*`	`*_id_field`
`flow_start/end/pause/resume_work_order`	`order_id_field` (默认 `data.order_id`)
`flow_abort_work_order`	+ `abort_code_field`, `abort_notes_field`
`flow_update_tag_alarm_band_numeric`	`tag_id_field`, `hi_field` 等(仅更新输入字段)

Edge 快速设置

Edge 节点均使用相同的选项组。

选项	说明
`url`	Edge REST 端点(例:`http://edge.local:60000/opc/server`)
`method`	HTTP 方法(未设置时按节点默认值)
`headers`	JSON 请求头(认证令牌等)
`body_template`	请求本体(未设置时原样发送 `data`)
`timeout_ms`	5000

流程 REST API — 以程序操作流程

将流程在外部自动化工具(Ansible/GitOps/CI 流水线)中以代码管理,或希望外部系统立即执行流程时使用的 REST API。

认证

所有 API 调用需在请求头中附加在安全 → API 认证令牌颁发的令牌。

Authorization: Bearer {api_token}

端点列表

1) 查询流程列表

GET /flow/list

curl -H "Authorization: Bearer ${TOKEN}" \
     https://platform.example.com/flow/list

响应:

{
  "data": [
    { "flow_id": "flow-abc123", "flow_name": "MES 동기화", "deployed": true,
      "node_count": 12, "exec_count": 9430, "error_count": 2, "last_exec_at": 1746247200000 },
    ...
  ]
}

2) 查询单个流程

GET /flow/get/{flow_id}

响应本体中包含图形节点·关系·选项全部。可原样用于备份/版本管理。

3) 创建流程

POST /flow/create
Content-Type: application/json

{
  "flow_name":        "신규 자동화",
  "description":     "외부 알림 → 워크오더 자동 생성",
  "deployed":         false
}

从响应中获取 flow_id 用于后续调用。

4) 修改流程(元信息)

POST /flow/update
Content-Type: application/json

{
  "flow_id":     "flow-abc123",
  "flow_name":   "수정된 이름",
  "description": "..."
}

5) 保存图形(批量替换节点·关系)

POST /flow/{flow_id}/graph
Content-Type: application/json

{
  "nodes":     [ { "node_id": "n1", "type": "flow_on_tag_point", "options": {...}, "x": 100, "y": 100 }, ... ],
  "relations": [ { "from_node_id": "n1", "to_node_id": "n2", "relation": "TRUE" }, ... ]
}

图形保存是事务性的 — 验证失败时整个图形会回滚。

6) 部署 / 解除

变更流程元信息的 deployed 字段会自动部署·解除。

# 배포
curl -X POST -H "Authorization: Bearer ${TOKEN}" -H "Content-Type: application/json" \
     -d '{"flow_id":"flow-abc123","deployed":true}' \
     https://platform.example.com/flow/update

7) 立即执行(手动触发)

POST /flow/{flow_id}/run
Content-Type: application/json

{
  "type":       "WEBHOOK",
  "originator": { "entity_type": "External", "id": "manual-run" },
  "data":       { "test": true }
}

不经过触发器节点的预过滤,立即从第一个节点开始执行。适用于手动验证·调试。

8) Dispatch(绕过触发器触发)

POST /flow/dispatch
Content-Type: application/json

{
  "type":       "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.SPEED" },
  "data":       { "value": 1500 },
  "metadata":   { "tag_id": "MOTOR-001.SPEED", "ts": 1746247200000 }
}

通过正常触发器处理路径接入消息 — 匹配该消息的所有流程会同时触发。

9) 重置统计

POST /flow/{flow_id}/stats/reset-errors        — 에러 카운트만 초기화
POST /flow/{flow_id}/stats/reset-all           — 전체 카운트 초기화

10) Export / Import — 图形备份

# Export — JSON 파일로 다운로드
curl -H "Authorization: Bearer ${TOKEN}" \
     https://platform.example.com/flow/${FLOW_ID}/export \
     -o flow-backup.json

# Import — 같은 JSON 을 다른 환경에 등록
curl -X POST -H "Authorization: Bearer ${TOKEN}" -H "Content-Type: application/json" \
     --data @flow-backup.json \
     https://platform.example.com/flow/import

Import 时若存在冲突 ID,会自动颁发新 ID。外部依赖(凭据·资产 ID 等)请在导入后单独映射。

11) 重新部署所有流程

POST /flow/redeploy

用于大量变更后或节点重启后的批量同步。运行中调用请谨慎 — 会产生临时处理延迟。

12) 自我诊断

POST /flow/selftest

对流程引擎的内部组件(触发器队列·分发器·节点注册表·脚本运行时)进行一次检查并返回状态。结果也会记录在诊断日志中。

13) 查询节点目录

GET /flow/catalog

返回当前注册的所有节点类型·选项模式。用于 UI 动态生成检查器。外部工具自动生成图形时也可参考。

用 Webhook 触发器从外部触发

包含 flow_on_webhook 触发器的流程,可让外部系统直接以以下 URL 触发。

POST /webhook/flow/{flow_id}
Authorization: Bearer {token}    또는    X-API-Key: {token}
Content-Type:  application/json

{
  "order_no":  "PO-001",
  "customer":  "ACME",
  "quantity":  1000
}

响应:

{ "status": "ACCEPTED", "trace_id": "tr-..." }

响应码	含义
200	已 enqueue 到消息队列(实际处理结果是异步的)
401	认证令牌错误/缺失
404	流程 ID 不存在或未部署
413	本体超过256KB
422	触发器节点不是 `flow_on_webhook`
429	超出每分钟调用限度

外部自动化工具集成示例

GitHub Actions — PR 合并时部署流程

- name: 플로우 배포
  run: |
    curl -X POST \
      -H "Authorization: Bearer ${{ secrets.PP_TOKEN }}" \
      -H "Content-Type: application/json" \
      --data @flows/mes-sync.json \
      https://platform.example.com/flow/import

Ansible — 流程批量管理

- name: 모든 플로우 백업
  uri:
    url: "https://platform.example.com/flow/{{ item }}/export"
    headers:
      Authorization: "Bearer {{ pp_token }}"
    dest: "/backups/{{ item }}.json"
  loop: "{{ pp_flow_ids }}"

Jenkins — 每次构建执行 selftest

stage('PlantPulse Flow Selftest') {
  steps {
    sh """
      curl -X POST -H 'Authorization: Bearer ${PP_TOKEN}' \\
           https://platform.example.com/flow/selftest \\
           --fail-with-body
    """
  }
}

API Rate Limit

端点	每分钟限度(按令牌)
`GET /flow/list` · `get` · `catalog`	600
`POST /flow/create` · `update` · `delete`	60
`POST /flow/{id}/run` · `dispatch` · `webhook/{id}`	1,000
`POST /flow/redeploy` · `selftest`	10

超出限度时返回 429 Too Many Requests 响应。

OPC/PLC 工业集成模式

工业现场特有的集成场景集合。结合Edge 节点22种和资产事件发布4种使用。

模式 A — OPC 服务器自动注册

新产线投产时,从 ERP 接收产线信息并自动向 Edge 注册 OPC 服务器。

[flow_on_webhook]
   ↓ (data = {"line_id":"LINE-7","host":"10.0.7.10","port":4840})
[flow_change_originator]
   ↓ (originator → Edge:EDGE-A)
[flow_edge_opc_create]
   ↓ (path=/api/v1/opc, body={"opc_id":"OPC-${data.line_id}","host":"${data.host}","port":${data.port}})
[flow_edge_opc_start]
   ↓ (수집 시작 명령)
[flow_save_attributes]
   ↓ (등록 이력을 자산 메타에 기록)

Edge 节点会自动从 mm_edge 主表 lookup api_key。操作员无需单独进行认证设置。

模式 B — PLC 命令自动发布

资产报警发生时,自动向 PLC 发布停止命令。

[flow_on_asset_alarm]
   ↓ where priority='ERROR' AND alarm_band='TRIP_HI'
[flow_publish_asset_command]
   ↓ (event_type='STOP', payload={"reason":"trip_high","triggered_by":"flow"})
[flow_log] (감사 로그)

flow_publish_asset_command 会立即通过资产的命令主题(asset/{id}/cmd/STOP)传递,Edge 对 PLC 执行 OPC Write。

模式 C — OPC 标签批量注册(CSV/Excel 接入)

新设备安装时一次性注册 CSV 中的100~1000件标签。

[flow_jdbc_poll]            ← 외부 DB 에 적재된 CSV 행
   ↓ (한 행 = 한 메시지)
[flow_script_transform]     ← 태그 정의 가공
   ↓
[flow_create_tag]           ← 도메인 등록 (NOT NULL 자동 보강)
   ↓
[flow_edge_tag_create]      ← 엣지에 동기화
   ↓
[flow_log] (성공 카운트)

注册1,000件时建议用 flow_throttle 将频率平滑至每分钟100件以下。Edge 可能暂时响应变慢。

模式 D — OPC 断开自动恢复

OPC 连接中断后恢复时自动重启。

[flow_on_opc_status]
   ↓ where prev_status='CONNECTED' AND status='DISCONNECTED'
[flow_delay 30s]            ← 잠시 안정화 대기
   ↓
[flow_edge_opc_start]       ← 수집 재시작 시도
   ↓ SUCCESS: 정상
   └ FAILURE: [flow_retry max=3] → EXHAUSTED: [관리자 알림]

模式 E — 班次开始时检查产线状态

在每个班次开始时刻检查站点的所有产线。

[flow_schedule cron='0 0 7,15,23 * * ?']   ← 7시·15시·23시
   ↓
[flow_edge_monitoring]      ← 엣지에서 OPC 상태 전체 조회
   ↓ data.opcs = [...]
[flow_split]                ← 배열을 행별 메시지로
   ↓
[flow_script_filter]        ← status != 'CONNECTED' 만
   ↓
[flow_send_email]           ← 비정상 OPC 만 한 통의 메일에 합쳐 발송

模式 F — 班次变更时自动映射作业者

根据排班变更自动映射工单的作业者。

[flow_on_entity_event type='ENTITY_UPDATED' Calendar]
   ↓
[flow_script_filter (시프트 시작 시각이 지금인 경우만)]
   ↓
[flow_jdbc_query (그 시프트의 작업자 목록 조회)]
   ↓ data.employees=[...]
[flow_split]
   ↓
[flow_update_work_order (작업자 매핑)]

模式 G — 多个 PLC 同时调用(Scatter-Gather)

同时调用多个 PLC 的数据后整合结果。

[flow_on_webhook]
   ↓ data.targets=['PLC-1','PLC-2','PLC-3']
[flow_split]
   ↓ (3개로 분할)
[flow_edge_tag_read]
   ↓ (병렬 호출)
[flow_merge window=5s]
   ↓ data.merged=[...]
[flow_script_transform (data.summary 만들기)]
   ↓
[flow_publish_asset_aggregation]

模式 H — 报警波段自动调整

将预测分析学习得到的 LIMIT_MIN/MAX 值自动应用到报警波段。

[flow_schedule cron='0 0 4 * * MON']    ← 매주 월요일 4시
   ↓
[flow_jdbc_query (forecast 결과 조회)]
   ↓ data.tags=[{tag_id, limit_min, limit_max}]
[flow_split]
   ↓
[flow_update_tag_alarm_band_numeric]    ← lo=limit_min, hi=limit_max
   ↓ SUCCESS
[flow_log] (적용 결과 기록)

模式 I — 停机时间自动分类

按原因分类资产停止事件,准确反映在 OEE 可用率中。

[flow_on_asset_event event_type='SHUTDOWN']
   ↓
[flow_switch]
   ├ CASE 점심: hour_of_day(ts) BETWEEN 12 AND 13 → [flow_create_calendar (계획정지)]
   ├ CASE 시프트 종료: 시프트 종료 시각 ±5분    → [flow_log (정상 종료)]
   ├ CASE 정기점검: 마지막 정비일 +30일 경과     → [flow_create_calendar (정기점검)]
   └ DEFAULT (비계획)                            → [flow_send_email (긴급)]

模式 J — 与外部 ERP 双向同步工单

PlantPulse ↔ 外部 ERP 双向镜像。

입력 1: ERP → 플랜트펄스
[flow_on_webhook] → [flow_create_work_order]

입력 2: 플랜트펄스 → ERP
[flow_on_entity_event type='ENTITY_CREATED' WorkOrder]
  → [flow_script_filter (출처가 ERP 가 아닌 경우만)]
  → [flow_http_request (ERP REST PUT)]

为防止无限循环,请对从外部接入的消息标记 metadata.source='ERP',并在相反方向的流程中过滤该消息。

工业集成时的注意事项

事项	建议
OPC Write 命令应在安全联锁后发布	仅在运转员确认或自动安全规则通过后发布
请勿将 PLC 轮询周期设置得过短	1秒以下会有 PLC CPU 占用率激增的风险
Edge 设备的 Docker 容器资源限度	OPC 采集 + 额外容器时需注意 CPU/内存超过80%
试运行阶段所有命令请用 dry-run 模式	利用 `flow_edge_tag_write` 的 `dry_run=true` 选项
停电应对 — 仅信任永久存储触发器	`flow_on_webhook` 等为易失性,域事件触发器为永久性

外部系统集成 Cookbook

按常调用的外部系统汇总 flow_http_request 设置和载荷示例。

Slack — Incoming Webhook 通知

URL:    https://hooks.slack.com/services/T0000/B0000/{secret}
Method: POST
Headers:
  Content-Type: application/json
Body (body_template):
  {
    "text": "${data.title}",
    "blocks": [
      { "type": "header",  "text": { "type": "plain_text", "text": "${data.title}" } },
      { "type": "section", "text": { "type": "mrkdwn",     "text": "*자산:* ${metadata.asset_id}\n*값:* ${data.value}\n*시각:* ${metadata.ts}" } }
    ]
  }

成功:200 + ok 本体
失败:400(载荷错误) / 404(secret 错误) / 429(超出每分钟调用限度)
建议:每分钟不超过1次连线 flow_throttle

Microsoft Teams — Webhook 通知

URL:    https://{org}.webhook.office.com/webhookb2/{id}/IncomingWebhook/{secret}
Method: POST
Headers:
  Content-Type: application/json
Body:
  {
    "@type":    "MessageCard",
    "@context": "https://schema.org/extensions",
    "themeColor": "FF0000",
    "title": "${data.title}",
    "sections": [{
      "facts": [
        { "name": "자산", "value": "${metadata.asset_id}" },
        { "name": "값",   "value": "${data.value}"       },
        { "name": "시각", "value": "${metadata.ts}"      }
      ]
    }]
  }

将 themeColor 分支为 FF0000(红)/FFA500(橙)/00C853(绿),可视化优先级。

Jira — 自动创建 Issue

URL:    https://{org}.atlassian.net/rest/api/3/issue
Method: POST
Headers:
  Authorization: Basic {base64(email:apitoken)}
  Content-Type:  application/json
Body:
  {
    "fields": {
      "project":    { "key": "OPS" },
      "summary":    "[자동] ${data.title}",
      "description": {
        "type": "doc",
        "version": 1,
        "content": [{
          "type": "paragraph",
          "content": [{ "type": "text", "text": "자산: ${metadata.asset_id}\n값: ${data.value}" }]
        }]
      },
      "issuetype":  { "name": "Bug" },
      "priority":   { "name": "High" }
    }
  }

响应本体 data.response_body.key 中会有类似 OPS-1234 的 issue key,供后续节点使用
失败:400(字段错误) / 401(认证) / 403(项目无权限)

GitHub Issues — 自动创建 Issue

URL:    https://api.github.com/repos/{owner}/{repo}/issues
Method: POST
Headers:
  Authorization: Bearer {pat_token}
  Accept:        application/vnd.github+json
Body:
  {
    "title":  "[자동] ${data.title}",
    "body":   "자산: ${metadata.asset_id}\n값: ${data.value}\n시각: ${metadata.ts}\ntrace: ${metadata.trace_id}",
    "labels": ["automation", "ops"]
  }

ERP (SAP S/4HANA OData) — 发布工单

URL:    https://{host}/sap/opu/odata/sap/API_MAINTNOTIFICATION/MaintenanceNotification
Method: POST
Headers:
  Authorization:   Basic {base64(user:pass)}
  X-CSRF-Token:    fetch          ← 별도 GET 으로 토큰 받기
  Content-Type:    application/json
Body:
  {
    "NotificationType":         "M2",
    "MaintenanceNotificationType": "M2",
    "TechnicalObject":          "${metadata.asset_id}",
    "NotificationText":         "${data.title}",
    "MalfunctionStartDate":     "${data.start_date}",
    "Priority":                 "${data.priority}"
  }

需先以 GET 获取 CSRF 令牌,再用相同会话 cookie 执行 POST。请连接两个 flow_http_request 节点,将第一个节点响应头中的令牌传播到第二个节点的请求头。

MES(外部数据库直接 INSERT) — `flow_jdbc_query`

向外部 MES 系统的工单表直接插入行。

INSERT INTO mes.work_order
  (order_no, asset_id, product_id, qty, status, due_date, created_by)
VALUES
  (:order_no, :asset_id, :product_id, :qty, 'NEW', :due_date, 'flow')

节点选项	值
`datasource_id`	mes_db(系统设置中预先注册)
`query`	上述 SQL
`binds`	`{"order_no":"${data.order_no}","asset_id":"${metadata.asset_id}",...}`

flow_jdbc_query 除 SELECT 外还支持 INSERT/UPDATE/DELETE。事务按节点单位自动 commit/rollback。

Grafana — 自动刷新仪表盘通知

URL:    https://{host}/api/annotations
Method: POST
Headers:
  Authorization: Bearer {api_key}
  Content-Type:  application/json
Body:
  {
    "dashboardUID": "...",
    "panelId":      4,
    "time":         ${metadata.ts},
    "tags":         ["alarm", "${metadata.asset_id}"],
    "text":         "${data.title}"
  }

报警发生时会在图表上自动显示标记 — 事后分析非常有用。

Telegram — 机器人消息

URL:    https://api.telegram.org/bot{token}/sendMessage
Method: POST
Body:
  {
    "chat_id": "-1001234567890",
    "text":    "🚨 *${data.title}*\n자산: `${metadata.asset_id}`\n값: ${data.value}",
    "parse_mode": "Markdown"
  }

REST 认证模式速查表

认证方式	请求头
API Key (请求头)	`X-API-Key: {token}`
API Key (查询字符串)	URL 末尾 `?api_key={token}`
Bearer Token	`Authorization: Bearer {token}`
Basic Auth	`Authorization: Basic {base64(user:pass)}`
OAuth2 (Client Credentials)	颁发令牌 → Bearer 请求头(以单独节点刷新)
HMAC 签名	以脚本计算基于本体/时刻的签名,附加 `X-Signature` 请求头

响应解析模式

flow_http_request 响应后,在后续节点使用 data.response_body:

// 응답에서 ID 추출
data.created_id = data.response_body.id;
data.status     = data.response_body.status;

// 응답 배열에서 첫 행만
data.first_item = (data.response_body.items || [])[0] || null;

// 응답이 문자열이면 JSON 파싱
if (typeof data.response_body === 'string') {
  try { data.response_body = JSON.parse(data.response_body); } catch (e) {}
}
msg

外部系统调用时的安全建议

事项	建议
请勿将令牌以明文留在图形中	在系统设置 → 凭据存储库注册后以 `${creds.slack_webhook}` 形式引用
若响应本体含敏感信息请掩码	用 `flow_script_transform` 移除 PII/令牌后传给下一节点
每个外部系统需单独重试策略	调用频率高的地方单独连线 `flow_retry` + `flow_throttle`
将调用结果保存为审计日志	用 `flow_save_attributes` 向资产添加调用历史元信息

数据转换 Cookbook

可直接移用到 flow_script_transform 节点中的转换模式集合。所有示例均以 JavaScript 为基准,使用 data / metadata 简写别名。

单位转换

// 섭씨 → 화씨
data.temp_f = data.temp_c * 9 / 5 + 32;

// 바 → kPa
data.pressure_kpa = data.pressure_bar * 100;

// rpm → rad/s
data.angular_velocity = data.rpm * 2 * Math.PI / 60;

// kWh → MJ
data.energy_mj = data.energy_kwh * 3.6;

// 바이트 → MB (소수 1자리)
data.size_mb = Math.round(data.bytes / 1024 / 1024 * 10) / 10;

msg

日期·时刻转换

const d = new Date(metadata.ts);

// ISO 8601 — 2026-05-12T12:34:56.789Z
data.iso = d.toISOString();

// 사람용 — 2026-05-12 21:34:56 (KST)
data.local = d.toLocaleString('ko-KR', { hour12: false });

// 날짜만 — 2026-05-12
data.date = d.toISOString().slice(0, 10);

// 시간만 — 21:34:56
data.time = d.toTimeString().slice(0, 8);

// 분 단위로 내림 (스파크라인 키)
data.minute_key = Math.floor(metadata.ts / 60000) * 60000;

// 시간 단위로 내림
data.hour_key = Math.floor(metadata.ts / 3600000) * 3600000;

// 한국 시간 (UTC+9) 직접 더하기 (서버 시간이 UTC 인 경우)
data.kst_hour = new Date(metadata.ts + 9 * 3600000).getUTCHours();

msg

字符串规范化

// 공백 trim + 소문자화
data.normalized = (data.text || '').trim().toLowerCase();

// 한글·영문 외 제거 (특수문자/공백 정리)
data.clean = (data.text || '').replace(/[^가-힣a-zA-Z0-9]/g, '');

// camelCase → snake_case
data.snake = (data.text || '').replace(/([A-Z])/g, '_$1').toLowerCase().replace(/^_/, '');

// 전화번호 정규화 (숫자만)
data.phone_digits = (data.phone || '').replace(/\D/g, '');

// 한국 휴대전화 자동 포맷 (010-1234-5678)
const p = (data.phone || '').replace(/\D/g, '');
data.phone_formatted = p.length === 11 ? p.replace(/(\d{3})(\d{4})(\d{4})/, '$1-$2-$3') : p;

msg

嵌套对象平坦化

// data.location.address.city → data.city
function flatten(obj, prefix, out) {
  out = out || {};
  for (var k in obj) {
    var v = obj[k];
    var key = prefix ? prefix + '_' + k : k;
    if (v && typeof v === 'object' && !Array.isArray(v)) flatten(v, key, out);
    else out[key] = v;
  }
  return out;
}
data = flatten(data);
msg

平坦对象 → 嵌套

// data.user_name + data.user_email → data.user = {...}
function unflatten(obj) {
  var out = {};
  for (var k in obj) {
    var parts = k.split('_');
    var cur = out;
    for (var i = 0; i < parts.length - 1; i++) {
      cur[parts[i]] = cur[parts[i]] || {};
      cur = cur[parts[i]];
    }
    cur[parts[parts.length - 1]] = obj[k];
  }
  return out;
}
data = unflatten(data);
msg

生成 CSV 行

// 외부 시스템에 CSV 한 줄 보낼 때
function csvEscape(v) {
  if (v === null || v === undefined) return '';
  v = String(v);
  return /[,"\n]/.test(v) ? '"' + v.replace(/"/g, '""') + '"' : v;
}
data.csv_row = [
  csvEscape(metadata.ts),
  csvEscape(metadata.asset_id),
  csvEscape(data.value),
  csvEscape(data.quality)
].join(',');
msg

数组聚合

var arr = data.values || [];

// 합·평균·최소·최대
data.sum = arr.reduce(function(a, b) { return a + b; }, 0);
data.avg = arr.length ? data.sum / arr.length : 0;
data.min = arr.length ? Math.min.apply(null, arr) : null;
data.max = arr.length ? Math.max.apply(null, arr) : null;

// 중앙값
var sorted = arr.slice().sort(function(a, b) { return a - b; });
var mid    = Math.floor(sorted.length / 2);
data.median = sorted.length === 0 ? null
            : sorted.length % 2 ? sorted[mid]
            : (sorted[mid - 1] + sorted[mid]) / 2;

msg

条件字段添加(架构演进)

// 알람 우선순위에 따라 색상·아이콘 자동 부여
var p = data.priority || 'INFO';
data.color   = { ERROR: '#d32f2f', WARN: '#f57c00', INFO: '#1976d2' }[p] || '#9e9e9e';
data.icon    = { ERROR: '🔴',       WARN: '🟠',       INFO: '🔵' }[p] || '⚪';
data.urgency = p === 'ERROR' ? 3 : p === 'WARN' ? 2 : 1;
msg

载荷部分掩码

function maskEmail(e) {
  if (!e || e.indexOf('@') < 0) return e;
  var parts = e.split('@');
  return parts[0].slice(0, 2) + '***@' + parts[1];
}
function maskPhone(p) {
  return (p || '').replace(/(\d{3})\d{4}(\d{4})/, '$1-****-$2');
}
data.user_email = maskEmail(data.user_email);
data.user_phone = maskPhone(data.user_phone);
msg

多语言(i18n)模板

// 메시지 본문을 사용자 로케일에 따라 분기
const tpl = {
  'ko-KR': '🚨 ${asset} 온도 ${value}°C 임계 초과',
  'en-US': '🚨 ${asset} temperature ${value}°C exceeds threshold',
  'ja-JP': '🚨 ${asset} 温度 ${value}°C 閾値超過'
};
const locale = metadata.user_locale || 'ko-KR';
const template = tpl[locale] || tpl['ko-KR'];
data.notification = template
  .replace('${asset}', metadata.asset_id)
  .replace('${value}', data.value);
msg

JSON Path 安全查询

// data.deep.nested.field 처럼 깊은 경로를 안전하게 조회
function get(obj, path, dflt) {
  var keys = path.split('.');
  var cur  = obj;
  for (var i = 0; i < keys.length; i++) {
    if (cur === null || cur === undefined) return dflt;
    cur = cur[keys[i]];
  }
  return cur === undefined ? dflt : cur;
}
data.city = get(data, 'location.address.city', 'Unknown');
msg

消息合并(`flow_merge` 后处理)

// flow_merge 가 data.merged 배열로 모은 메시지들을 1건으로 합침
var items = data.merged || [];
data.summary = {
  count:     items.length,
  first_ts:  items[0]?.metadata?.ts,
  last_ts:   items[items.length - 1]?.metadata?.ts,
  assets:    Array.from(new Set(items.map(function(m) { return m.metadata?.asset_id; }))),
  max_value: Math.max.apply(null, items.map(function(m) { return m.data?.value || 0 }))
};
delete data.merged;
msg

可复用脚本集合

可直接移用到 flow_script_filter / flow_script_transform / flow_switch 节点中的脚本库。所有脚本均以 JavaScript 为基准,使用 data / metadata 简写别名。

转换脚本

单位转换·打标签

// 섭씨 → 화씨 + 등급 라벨
data.temp_f = data.temp * 1.8 + 32;
data.grade  = data.temp > 80 ? 'HIGH' : data.temp < 0 ? 'LOW' : 'OK';
msg

时刻·班次·星期元信息补充

const d = new Date(metadata.ts);
const h = d.getHours();
metadata.shift   = (h >= 6 && h < 14) ? 'A' : (h < 22) ? 'B' : 'C';
metadata.weekday = ['SUN','MON','TUE','WED','THU','FRI','SAT'][d.getDay()];
metadata.is_weekend = (d.getDay() === 0 || d.getDay() === 6);
msg

平均值 ± 3σ 报警波段计算

const m = data.mean, s = data.stddev || 1;
data.tag_id = originator.id + '.TEMP';
data.hi     = m + 3*s;  data.lo    = m - 3*s;
data.hi_hi  = m + 4*s;  data.lo_lo = m - 4*s;
data.use_alarm = true;
msg

MES PO → 工单映射

const po = data;
data = {
  master_id: 'WO-MES-' + po.po_no,
  asset_id:  po.line_id || 'UNASSIGNED',
  title:     po.product_name + ' (' + po.qty + ')',
  due_date:  po.delivery_date,
  qty:       po.qty,
  product_id: po.product_code
};
msg

外部载荷规范化(统一各种字段名)

// 외부 시스템에 따라 키 이름이 다른 경우 — 한 줄로 정규화
data.value     = data.value ?? data.val ?? data.v;
data.timestamp = data.timestamp ?? data.ts ?? metadata.ts;
data.tag_id    = data.tag_id ?? data.tagId ?? data.id;
msg

累积阈值违反次数(有状态模式)

// 자산 attribute 와 함께 사용 — 상태는 메시지 자체에 적재
data.consecutive_fail = (data.consecutive_fail ?? 0) + (data.passed ? 0 : 1);
data.alert            = data.consecutive_fail >= 5;
msg

消息统计摘要(Merge 结果加工)

// flow_merge 후 data.merged 배열을 받아 요약
const arr = data.merged || [];
data.count   = arr.length;
data.values  = arr.map(m => m.data.value).filter(v => v != null);
data.mean    = data.values.reduce((a,b)=>a+b, 0) / (data.values.length || 1);
data.max     = Math.max(...data.values);
data.min     = Math.min(...data.values);
delete data.merged;
msg

按时段应用阈值

const h = new Date(metadata.ts).getHours();
const threshold = (h >= 8 && h < 18) ? 90 : 70;   // 주간 90, 야간 70
data.alarm = data.value > threshold;
data.threshold = threshold;
msg

过滤脚本

优先级白名单

['ERROR', 'CRITICAL'].includes(data.priority)

时间窗口(仅工作日)

const h = new Date(metadata.ts).getHours();
h >= 8 && h < 20

资产·标签 ID 模式

/^MOTOR-.*$/.test(originator.id)
// 또는
metadata.tag_id && metadata.tag_id.startsWith('LINE-A.')

阈值 + 稳定性(连续N次)

data.value > 100 && (data.consecutive_count ?? 0) >= 5

营业时间·节假日检查

const d = new Date(metadata.ts);
const h = d.getHours();
const wd = d.getDay();
// 평일 09-18시만 통과
wd >= 1 && wd <= 5 && h >= 9 && h < 18

仅特定站点

['SITE-01', 'SITE-02'].includes(metadata.site_id)

检查所有必填字段是否存在

data.value != null && data.tag_id && metadata.ts

阻断自身发布的消息(双向同步)

metadata.source !== 'flow'

Switch case 脚本

按优先级等级

// case "Critical"
['CRITICAL', 'EMERGENCY'].includes(data.priority)

// case "High"
data.priority === 'ERROR'

// case "Normal"
['WARN', 'INFO'].includes(data.priority)

按资产状态

// case "Running"
data.status === 'RUN'

// case "Stopped"
['STOP', 'IDLE', 'PAUSED'].includes(data.status)

// case "Faulted"
data.status === 'FAULT' || data.error_count > 0

按工单生命周期

// case "Start"
data.event_type === 'START_REQUEST'

// case "End"
data.event_type === 'COMPLETE' && data.qty_done >= data.qty_planned

// case "Abort"
data.event_type === 'CANCEL'

以上脚本均是汇集运行环境中常用模式的集合。可直接连线在图形中使用,只需按域调整阈值·字段名即可。

术语词典

流程引擎术语

术语	含义
entity_type	消息的主体实体种类 — `Asset`, `Tag`, `Site`, `Order`, `Customer`, `Product`, `Employee`, `Calendar` 等
originator	消息所指向的主体实体(`entity_type` + `id`)。例:`Asset/MOTOR-001`
type	消息分类标签。用于识别触发器接收到的事件种类
data	消息本体(载荷)— 主要由转换·动作节点读写
metadata	消息上下文(时刻·站点·班次·标签 ID 等)— 即使转换也会保持到流程末尾
relation	节点输出连线的标签。`SUCCESS`/`FAILURE`/`TRUE`/`FALSE`/`MATCH`/`NO_MATCH`/`DEFAULT`/`THROTTLED`/`EXHAUSTED` (均为大写)
*`_field` 动态选项**	从消息载荷路径(例:`data.tag_id`)读取值而非静态值的输入方式
glob 模式	触发器 `_pattern` 选项的通配符表达 — `` 匹配0字符以上,`?` 匹配恰好1字符
快照	保存图形时自动加载的版本单位备份。事故时用于恢复到之前状态
部分更新(fetch+merge)	Update 节点先查询现有记录后仅合并输入字段的方式。空值会被忽略
SKIPPED	触发器模式未匹配消息不流向后续节点也被排除在计数器之外的处理
EXHAUSTED	`flow_retry` 节点超过最大重试次数时触发的分支
THROTTLED	`flow_throttle` 节点阻断窗口内超出限度的消息时触发的分支

域·缩写词典

为方便在手册中快速参考工厂运行环境中常用的缩写,整理如下。

缩写	含义	在本手册中
MES	Manufacturing Execution System — 工单·生产实绩管理	外部集成对象(HTTP/MQTT/外部数据库)
ERP	Enterprise Resource Planning — 企业资源·计划系统	外部集成对象
SCADA	Supervisory Control and Data Acquisition — 监视·控制	外部集成对象 / `flow_publish_asset_command` 的出口
OPC	Open Platform Communications — 工业通信标准	Edge 类别(`flow_edge_opc_*`)的对象
OEE	Overall Equipment Effectiveness — 可动率 × 性能 × 品质	`flow_on_oee_event` 触发器
RAM	Reliability·Availability·Maintainability	`flow_on_ram_event` 触发器
EMS	Energy Management System	`flow_on_ems_event` 触发器
EQL	域事件规则表达式	`flow_script_filter`/`flow_script_transform` 的 `language: "EQL"` 选项
CEP	复合事件处理 (Complex Event Processing)	基于 EQL 的规则引擎。报警必须仅通过 CEP 路径产生才能保持一致性
PO	Purchase Order — 采购·生产订单	MES 集成时转换为工单的单位
WO	Work Order (工单)	`flow_*_work_order` 节点群
CMMS	Computerized Maintenance Management System	外部集成对象
MTTF/MTTR	Mean Time To Failure / To Repair	RAM 事件的核心指标
HMI	Human–Machine Interface	SCADA 等的运行画面

版本说明

以下是当前手册所涉及主要功能群的引入时间点。若您正在运行更旧的版本,部分功能的动作可能有所不同。

V2026.05 — 域自动化扩展

新增Edge (Edge) 类别11种 — OPC 服务器/标签 CRUD + 监控自动化
工单状态迁移5种 — flow_start/end/pause/resume/abort_work_order
标签报警波段部分更新2种 — 数值型/布尔型报警波段仅更新输入字段
新增 flow_retry 流程控制 — 退避 + 最大尝试次数 + EXHAUSTED 分支
新触发器5种 — flow_on_asset_health_status / _connection_status / _oee_event / _ram_event / _ems_event
Update 节点部分更新(fetch+merge) — 查询现有记录后仅合并输入字段
Relations 大写标准化 — SUCCESS/FAILURE/... 均改为大写,现有图形自动转换
SKIPPED 处理 — 将触发器模式未匹配的消息从计数器中排除
全部重新部署(redeploy)运维工具 — 批量重新加载活动流程
重置全部计数 — 批量重置节点 + 流程单位统计

V2026.03 — 发布稳定性

保存图形时加载自动快照
实时调试面板(检查器下方2秒刷新)+ 节点点灯·duration 显示
import/export(图形 JSON 文件)
引入负载看门狗 — 负载持续时发布一行诊断日志

更早之前

M1 — 引擎·UI 骨架,图形 CRUD,可视化画布
M2 — 域动作节点(资产·标签·工单·生产域等)
M3 — 触发器多样化,过滤/转换/外部集成8种,脚本引擎
M4 — 调试加载·执行历史画面,统计,部署/解除,导入/导出
M5 — 域触发器整合,场景批量验证

本手册所涉及的所有功能和动作均以上述 V2026.05 时点为基准。

目录​

概述​

学习路线图 — 按角色推荐入门顺序​

👶 初次使用者(1小时内创建第一个流程)​

🧑‍🏭 现场操作员(编写自动化场景)​

🛠 系统管理员(运维·调优·故障应对)​

🔌 开发者·集成工程师(外部系统集成)​

🔐 安全负责人(审计·认证)​

📚 快速参考(已熟悉的用户)​

核心概念​

初次开始​

端到端教程​

步骤 1 — 创建新流程​

步骤 2 — 放置触发器节点​

步骤 3 — 阈值过滤​

步骤 4 — 域变更 (标签 → 资产)​

步骤 5 — 发布作业指示​

步骤 6 — 邮件通知(成功分支)​

步骤 7 — 失败分支处理​

步骤 8 — 保存与测试运行​

步骤 9 — 验证与部署​

步骤 10 — 运维监控​

下一步​

画面构成​

列表画面​

顶部工具​

一览表​

批量部署·解除​

全部重新部署​

编辑画面 (可视化画布)​

顶部工具栏​

测试运行使用方法​

左侧 — 节点调色板​

中央 — 画布​

右侧 — 节点设置​

计数重置​

右侧 — 实时调试​

消息结构​

消息类型​

各触发器载荷示例​

flow_on_tag_point — 标签点位接入​

flow_on_tag_alarm — 标签报警发生​

flow_on_asset_data / flow_on_asset_event — 资产事件​

flow_on_asset_health_status / flow_on_asset_connection_status — 周期评估​

flow_on_oee_event / flow_on_ram_event / flow_on_ems_event — 插件事件​

flow_on_opc_status / flow_on_edge_status — OPC/Edge 状态​

flow_on_diagnostic — 系统诊断消息​

flow_on_domain_changed — 域变更事件​

flow_on_entity_event — 实体生命周期​

flow_on_webhook — 外部 HTTP 推送​

flow_on_mqtt_subscribe — MQTT 主题订阅​

flow_jdbc_poll — 外部数据库周期轮询​

flow_schedule — 基于时间触发​

载荷转换注意事项​

节点目录​

触发器 (22种)​

过滤 (6种)​

转换 (8种)​

动作 — 集成·存储 (2种)​

动作 — 资产事件发布 (4种)​

flow_asset_command_invoke — 执行而非发布​

动作 — 域 CRUD (34种)​

标签报警波段部分更新 (2种)​

作业指示(WorkOrder)状态迁移 (5种)​

Edge (Edge, 22种)​

查询3种​

Docker 应用控制 7种​

外部集成 (9种)​

流程控制 (7种)​

Retry 推荐连线模式​

节点选项详细参考​

flow_script_filter — 脚本过滤​

脚本上下文​

示例​

flow_script_transform — 脚本转换​

示例​

flow_switch — 多重分支​

示例​

flow_retry — 自动重试​

metadata 自动补全​

目录

概述

学习路线图 — 按角色推荐入门顺序

👶 初次使用者(1小时内创建第一个流程)

🧑‍🏭 现场操作员(编写自动化场景)

🛠 系统管理员(运维·调优·故障应对)

🔌 开发者·集成工程师(外部系统集成)

🔐 安全负责人(审计·认证)

📚 快速参考(已熟悉的用户)

核心概念

初次开始

端到端教程

步骤 1 — 创建新流程

步骤 2 — 放置触发器节点

步骤 3 — 阈值过滤

步骤 4 — 域变更 (标签 → 资产)

步骤 5 — 发布作业指示

步骤 6 — 邮件通知(成功分支)

步骤 7 — 失败分支处理

步骤 8 — 保存与测试运行

步骤 9 — 验证与部署

步骤 10 — 运维监控

下一步

画面构成

列表画面

顶部工具

一览表

批量部署·解除

全部重新部署

编辑画面 (可视化画布)

顶部工具栏

测试运行使用方法

左侧 — 节点调色板

中央 — 画布

右侧 — 节点设置

计数重置

右侧 — 实时调试

消息结构

消息类型

各触发器载荷示例

`flow_on_tag_point` — 标签点位接入

`flow_on_tag_alarm` — 标签报警发生

`flow_on_asset_data` / `flow_on_asset_event` — 资产事件

`flow_on_asset_health_status` / `flow_on_asset_connection_status` — 周期评估

`flow_on_oee_event` / `flow_on_ram_event` / `flow_on_ems_event` — 插件事件

`flow_on_opc_status` / `flow_on_edge_status` — OPC/Edge 状态

`flow_on_diagnostic` — 系统诊断消息

`flow_on_domain_changed` — 域变更事件

`flow_on_entity_event` — 实体生命周期

`flow_on_webhook` — 外部 HTTP 推送

`flow_on_mqtt_subscribe` — MQTT 主题订阅

`flow_jdbc_poll` — 外部数据库周期轮询

`flow_schedule` — 基于时间触发

载荷转换注意事项

节点目录

触发器 (22种)

过滤 (6种)

转换 (8种)

动作 — 集成·存储 (2种)

动作 — 资产事件发布 (4种)

`flow_asset_command_invoke` — 执行而非发布

动作 — 域 CRUD (34种)

标签报警波段部分更新 (2种)

作业指示(WorkOrder)状态迁移 (5种)

Edge (Edge, 22种)

查询3种

Docker 应用控制 7种

外部集成 (9种)

流程控制 (7种)

Retry 推荐连线模式

节点选项详细参考

`flow_script_filter` — 脚本过滤

脚本上下文

示例

`flow_script_transform` — 脚本转换

示例

`flow_switch` — 多重分支

示例

`flow_retry` — 自动重试

`metadata` 自动补全