事件分类 在业务流程中，有意义的状态变化都可称为事件。自定义好事件后，可基于该事件实现事件发送、事件订阅，对发送的事件数据进行分析处理等功能。低代码中的事件，分为标准事件和自定义事件两种。 标准事件是系统定义好的事件，可直接订阅。标准事件目前有几下几种： Absolute Time Alert：定时器发送的系统事件。 EV_FLOW_STARTED：流程启动，用于记录工作流的启动。 EV_FLOW_TERMINATED：流程终止，用于记录工作流的流程结束。 EV_FLOW_REVOKED：流程撤回，用于记录工作流的撤回。 EV_TASK_STARTED：任务启动，用于记录人工任务启动。 EV_TASK_TERMINATED：任务终止，用于记录人工任务结束。 自定义事件是指用户根据自身业务需求，自行定义的事件。例如，在某个公司招聘场景中，招聘人员A和招聘经理B负责面试，招聘经理B负责决定该应聘者是否被录用，录用后，人事部C来给新员工办理各种手续。需要进行软件设计，一旦应聘者被录用后，不需要招聘经理自己去通知人事部，而是创建一个记录应聘者录用结果状态的事件，通过订阅该事件，系统根据状态结果自动发邮件给人事部，通知人事部是否办理入职手续。

操作步骤 参考登录经典应用设计器中操作，登录经典版应用设计器。 将鼠标放在应用的某个文件夹上（如Logic），单击“+”，选择“事件”。 图1 选择事件 选择“创建新事件”，设置基本信息，单击“添加”。 图2 创建新事件 标签：新建事件的标签名，用于在页面展示。 名称：事件在系统中的唯一标识，取值为“命名空间__标签__e”。 描述：新建事件的描述信息，建议设置为事件的功能。 在事件详情页面的“自定义参数”中，新增自定义参数。 自定义参数用于记录有意义的事件状态变化，可以根据需要自行创建。例如，在本章节中创建的是记录应聘者是否被录用状态变化的事件，需要新增应聘者姓名“EmployName”和招聘结果“Result”两个参数字段。 在自定义参数中，单击“新建”，进入新增参数页面。 标签和名称设置为“EmployName”，字段类型选择“文本”，单击“保存”。 图3 “Cutsom Parameters”页签新增参数 标签：新建参数的标签名，用于在页面中展示。 名称：新增参数在系统中的唯一标识，创建后不可修改。 字段类型：新建参数的字段类型，如数字、文本、日期等。 描述：新建参数的描述信息，建议配置为参数的功能。 是否必填：参数是否必填，勾选后，表示必填字段。 参考上述操作，新增标签和名称为“Result”，字段类型为“布尔”的“Result”参数。 图4 新增Result参数 图5 自定义参数列表 在“基本信息”页签，单击“启用”，启用该事件。 在“订阅”页签，可查看该事件的订阅者信息。 由于事件刚创建，暂没有事件订阅者，下个章节介绍如何订阅该事件。

通过服务编排订阅事件 参考如何开发服务编排中操作，新建一个服务编排。 图1 新建一个服务编排 标签：新建服务编排的标签名，用于在页面展示。 名称：服务编排在系统中的唯一标识，创建后不可修改。 类型：新建服务编排的类型，本示例选择“Event Trigger”。 Autolaunched Flow：自启动服务编排，即在接口调用后，会立即执行服务编排中定义的逻辑。 Event Trigger：事件触发的服务编排，在事件触发时，才会开始执行服务编排中定义的逻辑。 事件：选择如何自定义事件中创建的事件（请确保事件已启用）。 描述：服务编排的描述信息。 在服务编排设计页面左侧，选择“逻辑”，拖拽“决策”图元至画布中。 该图元类似if语句的判断，用于在服务编排流程中创建判断条件，根据判断结果连线其他图元，决定后续执行何种操作。本示例需要设置应聘者被成功录用、应聘者没被录用和其他三个判断分支。 图2 拖拽决策图元至画布中 选中决策图元，单击，设置分支条件参数。 图3 pass分支条件配置 图4 unpass分支条件配置 可编辑的结果，即所有的判断条件。系统已预置默认条件，即当其他所有条件均不满足情况下，进入该分支。 本示例，需要新增如下两个判断条件： Pass：应聘者被录用。应聘通过时，事件中自定义的布尔类型参数“Result”赋值为全局常量“{!$GlobalConstant.True}”，表示结果为真，即应聘通过。 Unpass：应聘者没有被录用。应聘失败时，事件中自定义的布尔类型参数“Result”赋值为全局常量“{!$GlobalConstant.False}”，表示结果为假，即应聘拒绝。 在服务编排设计页面左侧，选择“基本”，拖拽“发送邮件”图元至画布中。 发送邮件是发送邮件的接口。本示例中，需要根据录用结果，发送不同的邮件给人事部。 录用成功，需要邮件通知人事部办理入职。 录用不成功，需要邮件通知人事部告知应聘者应聘失败。 事件超时（默认分支），需要邮件通知招聘经理，看其是否给出招聘结果。 如下图所示，以应聘者被成功录用为例，设置发送邮件参数。 选中发送邮件图元，单击，设置基本信息。 图5 设置邮件发送基本信息 单击，设置邮件发送参数。 图6 配置邮件发送参数 邮件配置：可选择“直接编辑”手动设置邮件信息或“基于模板”选择邮件模板。本示例选择“直接编辑”。 主题：邮件的标题。 内容：邮件内容，需要使用事件变量“EmployResultNotify__e”。 地址：接收人的邮箱地址。 字符串：表示直接输入。 集合：表示可以拖入一个集合变量。 参照5中操作，设置应聘者录用失败和超时邮件发送。 如图7所示，连接所有元素。 图7 连接并设置起始元素 在服务编排设计页面上方，单击，保存服务编排。 保存成功后，单击，启用服务编排。 启用后，当有“EmployResultNotify”事件触发时，即会执行服务编排中定义的逻辑。

结果验证 创建一个服务编排“fire_alarm”，创建后启用该服务编排，实现当kafka中出现高温报警事件时，AstroZero需要接收该事件，再向第三方系统发送该告警邮件。 鼠标放在应用下的Logic文件夹上，单击“+”，选择“服务编排”。 选择“创建一个新的服务编排”，设置标签和名称为“fire_alarm”，单击“添加”。 在服务编排设计页面左侧，拖拽“逻辑”下的“等待”图元至画布中。 单击，在“可编辑的事件”中，单击“新增”，新增“event0”分支。 配置该分支事件为高温报警事件，当系统中出现该事件时，会执行该分支。 图9 配置等待图元 拖拽“基本”下的“发送邮件”图元至画布中。 单击，设置发送邮件参数。 配置高温报警邮件的主题和内容，地址填入第三方系统的邮件接收地址。 图10 配置发送邮件图元 直接编辑：手动设置邮件信息。 基于模板：基于同一账号或基线中已有的邮件模板，设置邮件主题和内容，方便用户增加效率。 模板：选择“基于模板”时，需要选择邮件模板。 主题：设置邮件标题。 选中“直接编辑”时，需要配置邮件标题。 选中“基于模板”时，该参数配置区域置灰。 内容：设置邮件内容。 选中“直接编辑”时，需要配置邮件内容。 选中“基于模板”时，该参数配置区域置灰。 自定义变量：选中“基于模板”时，需要配置模板中的参数。在“值”中，直接输入参数取值或者从全局上下文拖拽变量。 地址：接收人的邮箱地址。 字符串：直接输入分号分隔的字符串或从全局上下文拖拽变量。 集合：从全局上下文拖拽集合变量。 抄送：抄送人员的邮箱地址，直接输入分号分隔的字符串或者从全局上下文拖拽变量。 密送：密送发送人员的邮箱地址，直接输入分号分隔的字符串或者从全局上下文拖拽变量。 按照下图连接所有图元，等待图元和发送邮件图元之间选择“event0”分支。 图11 连接所有图元 单击页面上方的，保存服务编排。 保存成功后，单击，启用服务编排。 创建一个服务编排“send_fire_event”，将温度监测事件“temperature”中位置参数“position”赋值为仓储公司某区域，温度“temperature”赋值为70度，启用该Flow，用于模拟事件发送。 鼠标放在应用下的Logic文件夹上，单击“+”，选择“服务编排”。 选择“创建一个新的服务编排”，设置标签和名称为“send_fire_event”，单击“添加”。 在服务编排编辑器页面左侧，拖拽“基本”下的“发送事件”图元至画布中。 单击，设置发送事件参数。 图12 配置发送事件图元 事件：选择待发送的事件。 分区字段：从事件中选择一个自定义参数用作Kafka的分区字段，则根据该字段值进行路由，相同的值将路由到同一个Kafka分区。如果不指定，则默认随机路由到不同Kafka分区，从而提升并发处理性能。 发送到外部：系统内部有配置与AstroZero相连Kafka的固定Topic“__BINGO_SYS_TOPIC”。若不勾选“发送到外部”，表示将事件数据发到默认的Topic上。若勾选“发送到外部”，则表示将事件数据发送到与AstroZero相连Kafka的其他Topic上。 延迟到事务结束（提交或回滚）后才发送：是否延迟到该服务编排事务结束后，才发送事件。 主题：勾选“发送到外部”时，才显示该参数，表示非系统配置的默认Topic。 目标/源：配置事件数据。 本示例中，“目标”请从下拉框中选择事件的自定义参数“temperature”和“position”，在“源”中给目标赋值。 按照下图连接所有图元。 图13 连接图元 单击页面上方的，保存服务编排。 保存成功后，单击，启用服务编排。 在“事件流”列表页面，单击该事件流程所在行的，运行该事件流程。 单击服务编排“send_fire_event”编辑器上方的，执行服务编排。 不用输入任何输入参数，直接单击“运行”。 参考上一步，执行服务编排“fire_alarm”。 不用输入任何输入参数，单击“运行”。 打开第三方系统邮箱，成功收到高温警报邮件，验证成功。

前提条件 请联系运维，开启事件流License权限。开启后，才可新建、查看并管理事件流。 已创建并启用温度监测事件“temperature”，该事件中自定义参数为文本类型“position”和数字类型“temperature”，分别表示位置和温度，具体操作步骤请参考如何自定义事件。 已创建并启用高温报警事件“fire_alarm”，该事件中自定义参数为文本类型“positionalarm”和数字类型“temperaturealarm”，分别表示高温报警位置和温度，具体操作步骤请参考如何自定义事件。

操作步骤 参考登录经典应用设计器中操作，登录经典版应用设计器。 将鼠标放在应用的某个文件夹上（如Logic），单击“+”，选择“事件流”。 开启事件流License权限后，才会有新建事件流的入口。 单击“新建”，进入“事件流”页面。 在“事件流”页面右侧，单击“设置”，进行数据配置。 图2 设置页面 窗口长度：事件流处理的数据采集时间间隔，单位为分钟。建议时间间隔不要配置过小，否则系统频繁的建表和执行SQL语句，会影响系统性能。例如：配置为“1”表示spark streaming每隔1分钟去kafka获取事件数据。 在“事件流”页面右侧，单击“图元”，拖拽“输入源”中Kafka图元至左侧画布区域，并配置基本信息，单击“保存”。 输入源中Kafka图标表示数据来自Kafka。 图3 数据来源Kafka基本信息页面 标签：新建数据源Kafka的标签名，用于在界面展示。 名称：新建数据源Kafka的名称，系统自动生成。 事件：需要订阅的事件。例如，配置为温度监测事件“temperature”。 从外部接收：是否从外部Kafka接收事件。 关闭该开关，从与AstroZero相连的Kafka接收事件。 打开该开关，从外部Kafka接收事件。本示例中关闭该开关，使用与AstroZero相连的Kafka。 Kafka集群地址：数据源Kafka的集群地址。多个节点服务器地址可用“,”间隔，格式为“Kafka节点1的ip:节点1端口号,Kafka节点2的ip:节点2端口号,...”。 获取Kafka的IP和端口号方法：登录每台Kafka节点服务器，查看“${KAFKA_HOME}/config”目录下，“server.properties”中“listeners”的取值。 订阅主题：获取的事件数据来源，即来自于kafka集群的哪个Topic。 从“图元”中，拖拽“投影”图元至左侧画布区域数据源Kafka下方，并在弹出的“添加投影器”页面配置基本信息，单击“保存”。 图4 “添加投影器”页面 标签：新建投影的标签名，用于在界面展示。 名称：新建投影的名称，系统自动生成。 输入字段：订阅事件的所有事件参数。选择所需的参数，单击，选中的参数会出现在“输出字段”中。 输出字段：从订阅事件的事件参数进行筛选后，用到的事件参数。 从“图元”中，拖拽“过滤”图元至左侧画布区域投影下方，并在弹出的“添加过滤器”页面配置基本信息和过滤条件，单击“保存”。 图5 设置过滤条件 标签：新建过滤的标签名，用于在界面展示。 名称：新建过滤的名称，系统自动生成。 条件设置：设置过滤条件。例如，图中的配置表示提取温度大于60度的事件。 字段：上一步筛选后的事件参数。 比较符：操作符，直接在下拉框中选择。 值：字段值。 （可选）从“图元”页面中，拖拽“分组”图元至左侧画布区域过滤器下方，并在弹出的“添加分组器”页面配置基本信息和分组条件，单击“保存”。 图6 “添加分组器”页面配置 标签：新建分组的标签名，用于在界面展示。 名称：新建分组的名称，系统自动生成。 分组设置：设置分组条件。例如，图中配置表示按照相同位置的大于60度的温度取温度平均值。 在“图元”页面右侧，拖拽输出源中的Kafka图元至左侧画布区域最下方，设置spark streaming输出事件和输出事件数据的接收方，单击“保存”。 可以拖拽多个Sink中的Kafka图元，设置多个事件接收方。 图7 数据接收方配置 标签：数据接收方Kafka的标签名，用于在界面展示。 名称：数据接收方Kafka的名称，系统自动生成。 事件：spark streaming输出的事件。本示例此处配置为高温报警事件“fire_alarm”。温度监测事件“temperature”中的“position”，对应高温报警事件“fire_alarm”中的“positionalarm”。温度监测事件的“temperature”，对应高温报警事件“fire_alarm”中的“temperaturealarm”。 发送到外部：是否发送到外部Kafka。若关闭该开关，表示发送到与AstroZero相连的Kafka。若打开该开关，表示发送到外部Kafka，需要配置“Kafka集群地址”和“订阅主题”。本示例中关闭该开关，使用与AstroZero相连的Kafka。 Kafka集群地址：接收事件的Kafka集群地址，例如10.136.14.56:9092。 多个节点服务器地址可用“,”间隔，格式为“Kafka节点1的ip:节点1端口号,Kafka节点2的ip:节点2端口号,...”。 获取Kafka的IP和端口号方法：登录每台Kafka节点服务器，查看“${KAFKA_HOME}/config”目录下，“server.properties”中“listeners”的取值。 接收事件的kafka集群，不需要一定配置为和AstroZero相连。 订阅主题：接收事件数据的kafka Topic，例如__BINGO_PROD_SYS_TOPIC。 事件属性/属性、值：配置输出事件与接收事件中，已定义参数的对应关系。 单击页面右上方保存按钮，设置“事件流”基本参数。 图8 该“事件流”基本参数页面 标签：新建事件流的标签名，用于在界面展示。 名称：新建事件流的名称，系统自动生成。 描述：新建事件流的描述信息。 单击页面上方的，启用事件流。

背景信息 首先，您需要理解以下概念： kafka：是由Apache开发的一个开源流处理平台，是分布式发布、订阅消息系统。kafka主要用于处理活跃的流数据，一般kafka部署为集群模式。 Topic：用来区分kafka集群中，不同类型消息的主题。例如，应用程序A订阅了Topic1，应用程序B订阅了Topic2而没有订阅Topic1，那么发送到Topic1中的数据将只能被应用程序A读到，而不会被应用程序B读到。 spark streaming：大规模流式数据处理工具，第三方工具。

异常处理 当数据源类型选择的是ROMA Connect，配置完数据接入后，MQS平台收不到消息，业务设置的EventTrigger没有拉起Flow。现象、原因分析以及解决方法如下： 问题现象：ROMA Connect消息轨迹是灰色的，可能是MQS服务地址配置错误或者没有配置。解决方法为：检查iotgateway容器的“roma-server”参数是否配置，若没有配置，请配置该参数或联系运维添加正确的“roma-server”地址（端口为“9776”），确认网络正常以及MQS服务器正常，兼容客户端版本。 问题现象：ROMA Connect消息轨迹是红色的，则可能是MQS的source定义错误。解决方法为：在ROMA Connect平台查看应用ID和对应的密钥是否一致，查看该应用是否订阅了主题。配置错误时会导致鉴权失败。 问题现象：ROMA Connect消息轨迹是绿色的， 表示MQS接收成功，并成功发送到Kafka，AppCube也收到了对应的消息，但是可能没成功拉起Flow。解决方法为：查看iotgateway的“flume.log”日志，确认MQS是否接收成功，通过“kafka-console-cumer”确认消息是否转换成功，通过AppCube的日志跟踪功能或者AppCube后台日志确认（日志关键字为“OnEvent”）平台是否收到事件。如果没有对应的服务编排或者脚本日志，可以确定Flow未拉起，联系华为工程师解决。

操作步骤 在开发环境修改数据接入的任意配置数据（包括所有图元的配置信息），在打包时，可配置是否覆盖新环境中的同名数据接入配置。具体配置可参考应用打包发布。 鼠标放在App下的Logic文件夹旁会出现加号，单击加号，选择“数据接入”。 单击“新建”，进入“数据接入”创建页面。 从页面右侧“图元”的输入源中拖拽OneNET图标至左侧画布区域，并配置外部数据源基本信息，单击“保存”。 Source中不同的图标表示不同的第三方数据源，支持以下几种数据源。 OneNET 图1 OneNET配置信息页面 图2 OneNET元数据定义页面 表1 数据来源OneNET基本信息页面参数说明 参数名称 参数说明 标签 该数据源OneNET的显示标签名。长度不超过64个字节。 名称 该数据源OneNET的名称，单击参数值后系统可根据“标签”取值自动生成。 协议 通信协议。 Token 数据签名，即令牌，由OneNET提供。 报文加密 对传输过来的报文内容是否进行加密，勾选表示加密。如果设置为加密的，则需配置“Aeskey”。 Aeskey 解密密钥，由OneNET提供。 元数据 源数据的元数据定义。如果为空的{}，您只能传送原始数据，无法使用拦截器对数据进行处理。 定义好的元数据将会显示在右侧区域中。 单击“格式化Json”，可对元数据进行Json格式化处理。 元数据定义示例如下： { "notifyType": "STRING", "requestId": "STRING", "timestamp": "STRING", "eventTime": "datetime", "deviceId": "STRING", "gatewayId": "STRING", "deviceService": { "name": "STRING" }, "service": { "serviceType": "STRING", "serviceId": "STRING", "body": "OBJECT", "data": { "battery_low": "number", "light": "STRING" } }} 数据接入对接中移物联OneNET平台的认证，是按照中移物联给出的对接标准，需根据配置的Token（令牌）、报文中的随机数、报文内容拼接后获取MD5摘要，并和报文中的摘要做比对，如果一致则认为报文是正常的报文。目前中移物联只支持MD5算法获取数据摘要。参见OneNET平台开发文档。 同时，对于报文内容来说，OneNET平台支持对报文内容进行加密后上报。AppCube要使用跟OneNET平台一致的加密算法进行解密，获取的加密向量要跟OneNET平台一致。因此，使用了Aeskey（解密密钥）的前16位作为加密向量。 设备接入 IoTDA（该产品之前名为“Ocean Connect”） 图3 设备接入 IoTDA配置信息页面 表2 数据来源设备接入 IoTDA基本信息页面参数说明 参数名称 参数说明 标签 该数据源设备接入 IoTDA的显示标签名。长度不超过64个字节。 名称 该数据源设备接入 IoTDA的名称，单击参数值后系统可根据“标签”取值自动生成。 数据格式 设备上报数据的格式。 数据传输的协议支持http或者https，由IoTDA决定。 Access Key 单击“生成AK/SK”，该值会自动生成。请将生成的Access Key保存到本地，开发人员通过该值在设备侧生成鉴权属性“authorization”。 Secret Key 单击“生成AK/SK”，该值会自动生成。请将生成的Secret Key保存到本地，开发人员通过该值在设备侧生成鉴权属性“authorization”。 元数据 源数据的元数据定义。如果为空的{}，您只能传送原始数据，无法使用拦截器对数据进行处理。 定义好的元数据将会显示在右侧区域中。 单击“格式化Json”，可对元数据进行Json格式化处理。 元数据定义示例如下： { "notifyType": "STRING", "requestId": "STRING", "timestamp": "STRING", "eventTime": "datetime", "deviceId": "STRING", "gatewayId": "STRING", "deviceService": { "name": "STRING" }, "service": { "serviceType": "STRING", "serviceId": "STRING", "body": "OBJECT", "data": { "battery_low": "number", "light": "STRING" } }} 上报的合法数据示例如下： { "notifyType": "deviceDataChanged", "requestId": "45678", "timestamp":"1900012929922992", "eventTime":"20151212T121212Z", "deviceId":"8b3979fc-b072-433b-b3f6-673072e1bc04", "gatewayId":"XXX", "deviceService":{"name":"buttery"}, "service": { "serviceType": "", "serviceId": "", "data":{ "battery_low":444, "light":"99", "authorization":"Bingo bDTKFSGi:ZwGnN+bb*****************************TjFRMA=" } }} 其中“authorization”值根据AK/SK、上报URL和Body数据生成。 π平台 图4 π平台配置信息页面 表3 数据来源π平台基本信息页面参数说明 参数名称 参数说明 标签 该数据源的显示标签名。长度不超过64个字节。 名称 该数据源的名称，单击参数值后系统可根据“标签”取值自动生成。 字段 请填写输入源的数据字段，不同的字段以逗号进行分隔，例如“id,name,age”。 协议 数据传输的协议。 Access Key 单击“生成AK/SK”，该值会自动生成。请将生成的Access Key保存到本地，并将该值填写到对应的 物联网平台 中。 Secret Key 单击“生成AK/SK”，该值会自动生成。请将生成的Secret Key保存到本地，并将该值填写到对应的物联网平台中。 ROMA Connect 图5 ROMA Connect配置信息页面 表4 数据来源ROMA Connect基本信息页面参数说明 参数名称 参数说明 标签 该数据源的显示标签名。长度不超过64个字节。 名称 该数据源的名称，单击参数值后系统可根据“标签”取值自动生成。 APP ID MQS平台应用ID。应用ID能够引用一个本平台定义的系统参数，引用位置不限制。示例：{!roma_app_prefix}other_roma_app_id。 密钥 MQS平台应用请求密钥。 主题 MQS主题。主题能够引用一个本平台定义的系统参数，引用位置不限制。示例：{!roma_topic_prefix}other_roma_topic。 标签 设置订阅消息的标签，可以指定消费某一类型的消息。默认*表示消费所有类型的消息，可以写多个，用“||”隔开，例如“tag1 || tag2 || tag3”。 通道加密 数据是否加密。若勾选，表示MQS数据传输采用TLS协议。 消费位置 若勾选，停止数据接入任务，后续重新运行数据接入时，停止和重新运行期间ROMA Connect发来的消息会被完全丢弃。若停止与运行数据接入之间时间间隔在十分钟之内，消息可能不会丢失。 注意：存在数据丢失风险，请谨慎勾选。 默认不勾选。 元数据 源数据的元数据定义。如果为空的{}，您只能传送原始数据，无法使用拦截器对数据进行处理。 定义好的元数据将会显示在右侧区域中。 单击“格式化Json”，可对元数据进行Json格式化处理。 元数据定义示例如下： { "notifyType": "STRING", "requestId": "STRING", "timestamp": "STRING", "eventTime": "datetime", "deviceId": "STRING", "gatewayId": "STRING", "deviceService": { "name": "STRING" }, "service": { "serviceType": "STRING", "serviceId": "STRING", "body": "OBJECT", "data": { "battery_low": "number", "light": "STRING" } }} Kafka 图6 添加Kafka 表5 数据来源Kafka基本信息页面参数说明 参数名称 参数说明 标签 该数据源Kafka的显示标签名。长度不超过64个字节。 名称 该数据源Kafka的名称，单击参数值后系统可根据“标签”取值自动生成。 连接外部Kafka AppCube支持从平台内部Kafka或者外部Kafka采集数据，从内部Kafka采集的数据不能再输出到内部Kafka中。勾选该项，表示从外部Kafka采集数据，则需要配置“Kafka集群地址”和“订阅主题”；默认不勾选，表示从平台内部Kafka采集数据。 Kafka集群地址 当勾选“连接外部Kafka”时，该参数才会出现。表示Kafka集群地址，多个节点服务器地址可用“,”间隔，格式为“Kafka节点1的ip:节点1端口号,Kafka节点2的ip:节点2端口号,...”。 获取Kafka的IP和端口号方法：可以通过登录每台Kafka节点服务器，查看“${KAFKA_HOME}/config”目录下“server.properties”中“listeners”取值获取。 例如：10.10.10.1:9091,10.10.10.2:9092 订阅主题 当勾选“连接外部Kafka”时，该参数才会出现。表示采集数据所属的Kafka主题。 主题是消息存储和发布的类名，通过主题您可以分类传输和处理事件。 开启SASL_SSL 当勾选“连接外部Kafka”时，该参数才会出现。开启SASL_SSL后，访问外部Kafka需要提供用户名和密码进行认证，且数据会加密传输。需要根据外部Kafka服务端的实际情况配置。 默认不勾选。 用户名 当勾选“开启SASL_SSL”，该参数才会出现。表示访问外部Kafka需要的用户名。 密码 当勾选“开启SASL_SSL”，该参数才会出现。表示访问外部Kafka需要的密码。 元数据 当勾选“连接外部Kafka”时，会出现“元数据定义”页签，选择该页签，配置源数据的元数据定义。如果为空的{}，您只能传送原始数据，无法使用拦截器对数据进行处理。 定义好的元数据将会显示在右侧区域中。 单击“格式化Json”，可对元数据进行Json格式化处理。 元数据定义示例如下： { "notifyType": "STRING", "requestId": "STRING", "timestamp": "STRING", "eventTime": "datetime", "deviceId": "STRING", "gatewayId": "STRING", "deviceService": { "name": "STRING" }, "service": { "serviceType": "STRING", "serviceId": "STRING", "body": "OBJECT", "data": { "battery_low": "number", "light": "STRING" } }} 事件 当不勾选“连接外部Kafka”时，该参数才会出现。表示从平台内部Kafka采集数据时，数据所关联的具体事件。 5G消息 图7 5G消息配置信息 图8 5G消息元数据定义 表6 数据来源5G消息配置参数说明 参数名称 参数说明 标签 该数据源的显示标签名。长度不超过64个字节。 名称 该数据源的名称，单击参数值后系统可根据“标签”取值自动生成。 协议 数据传输的协议。这里所选的协议必须与在5G消息云服务平台选择的传输协议一致。 数据格式 设备上报数据的格式。 APP ID 设备ID。这里填写的APP ID必须与在5G消息云服务平台填写的保持一致。 支持引用变量。变量在应用开发界面左下角“设置”里“系统参数”页签进行设置。 密钥 5G网关请求密钥。 需要引用变量时，请勾选“引用变量”。变量在应用开发界面左下角“设置”里“系统参数”页签进行设置。 元数据类型 该参数在“元数据定义”页签。为元数据类型，说明如下： 上行消息：5G消息云服务平台向业务应用转发的终端用户上行消息。 状态报告通知消息：业务应用通过5G消息云服务平台向用户发送消息后，5G消息云服务平台向业务应用发送的结果状态报告（即回执消息）。 媒体审核通知：业务应用向5G消息云服务平台上传多媒体内容文件后，文件会送运营商的MAAP平台审核，审核后，5G消息云服务平台向业务应用通报媒体文件审核结果。 消息撤回通知：业务应用发送5G消息给使用5G消息云服务的用户，希望撤回消息。业务应用发起撤回动作到5G消息云服务平台，5G消息云服务平台发送到终端用户，撤销成功后返回5G消息云服务平台，5G消息云服务平台发送消息撤回状态通知给业务应用。 在页面右侧拖拽拦截器中适当的图标，对源数据进行数据拦截。配置后单击“保存”。 拦截器中包含以下图标。 投影：数据选择器。可从源数据中选择部分数据进行发送。 图9 投影配置 表7 投影配置参数说明表 参数名称 参数说明 标签 该投影的显示标签名。长度不超过64个字节。 名称 该投影的名称，单击参数值后系统可根据“标签”取值自动生成。 字段选择器 左侧为外部数据源传入的全部元数据，勾选所需数据字段，单击将所选数据移到右侧输出字段中。 赋值：数据赋值器。可将缓存中的对象数据、传入的外部数据源数据、某个常量数据或者数据取表达式值赋值给某参数。 图10 赋值器配置 表8 赋值器配置参数说明表 参数名称 参数说明 标签 该赋值器的显示标签名。长度不超过64个字节。 名称 该赋值器的名称，单击参数值后系统可根据“标签”取值自动生成。 新增字段 被赋值的参数。 源 源数据。支持以下几种类型。 Cache：缓存中的对象数据。需要提前在设置中单击“新增缓存”进行添加。 reference：传入的外部数据源元数据。 formula：数据取表达式值。格式为“XXX(数据源参数)”。 constant：某常量。格式为“常量值”。 在设置中单击“新增缓存”添加缓存对象数据时，设置如下图所示。 图11 “新增缓存”配置 表9 “新增缓存”参数说明 参数名称 参数说明 标签 该Cache的显示标签名。长度不超过64个字节。 名称 该Cache的名称，单击参数值后系统可根据“标签”取值自动生成。 对象 映射的对象名。 高级配置 高级配置，开关打开后需要配置如下参数。 缓存上限：最大缓存记录数。 过期时间：缓存超时时间。 默认不打开，采用系统默认配置。 字段选择器 左侧为该对象的全部字段，勾选所需数据字段，单击将所选字段移到右侧输出字段中。 键 与外部元数据关联的对象字段。 选择源字段 传入的外部数据源元数据中与该对象关联的数据字段。 过滤：数据过滤器。您可通过配置一定的过滤条件对外部元数据进行数据过滤。 图12 过滤器配置 表10 过滤器配置参数说明表 参数名称 参数说明 标签 该过滤器的显示标签名。长度不超过64个字节。 名称 该过滤器的名称，单击参数值后系统可根据“标签”取值自动生成。 条件设置 过滤条件配置。 单击“新增”可添加多个过滤条件。 在页面右侧拖拽通道中“内存通道”图标至左侧画布区域Onenet下方，并配置数据通道的基本信息，单击“保存”。 通道中“内存通道”表示普通的数据通道，当数据较多时可选择“文件通道”图标进行拖拽。 图13 内存通道配置 表11 内存通道配置参数说明 参数名称 参数说明 标签 该内存通道的显示标签名。长度不超过64个字节。 名称 该内存通道的名称，单击参数值后系统可根据“标签”取值自动生成。 高级配置 高级配置，开关打开后需要配置如下参数： 通道容量：通道数据最大条数。 最大传输量：发到单个输出源的数据最大条数。 默认不打开，采用系统默认配置。 在页面右侧拖拽Sink输出源中的图标至左侧画布区域最下方，设置输出事件和输出事件数据的接收方。 当接收方是Kafka时，配置如下。 您可拖拽多个Sink中的Kafka图标，设置多个事件接收方。 图14 数据接收方配置页签 表12 数据接收方配置参数说明 参数名称 参数说明 标签 数据接收方Kafka的显示标签名。长度不超过64个字节。 名称 数据接收方Kafka的名称，单击参数值后系统可根据“标签”取值自动生成。 事件 输出的事件。 发送到外部 是否发送到外部Kafka。若打开该开关，需要配置“Kafka集群地址”和“订阅主题”。 Kafka集群地址：接收事件的kafka集群地址，多个节点服务器地址可用“,”间隔，格式为“Kafka节点1的ip:节点1端口号,Kafka节点2的ip:节点2端口号,...”。 获取Kafka的IP和端口号方法：可以通过登录每台Kafka节点服务器，查看“${KAFKA_HOME}/config”目录下“server.properties”中“listeners”取值获取。 例如：10.136.14.56:9092 说明：接收事件的kafka集群不需要一定配置为和AppCube相连。 订阅主题：接收事件数据的kafka Topic。 例如：__BINGO_PROD_SYS_TOPIC 是否开启SASL_SSL：当发送到外部Kafka时，该参数才会出现。表示是否开启SASL_SSL认证，开启后，访问外部Kafka需要提供用户名和密码进行认证，且数据会加密传输。需要根据外部Kafka服务端的实际情况配置。默认勾选。 默认不打开该开关，表示输出到与AppCube相连的Kafka。 事件属性/作为分区Key 是否将事件中的参数字段作为Kafka的消息头，用于事件分发时发送到不同的Kafka分区。建议选择有区分度的关键字段。 事件属性/属性、值 配置传输的数据与事件中已定义参数的对应关系。 单击“新增行”可添加多行。 当接收方是ROMA Connect，配置如下。 图15 数据接收方ROMA Connect配置页签 表13 ROMA Connect基本信息页面参数说明 参数名称 参数说明 标签 该接收方的显示标签名。长度不超过64个字节。 名称 该接收方的名称，单击参数值后系统可根据“标签”取值自动生成。 APP ID MQS平台应用ID。应用ID能够引用一个本平台定义的系统参数，引用位置不限制。示例：{!roma_app_prefix}other_roma_app_id。 密钥 MQS平台应用请求密钥。 主题 MQS主题。主题能够引用一个本平台定义的系统参数，引用位置不限制。示例：{!roma_topic_prefix}other_roma_topic。 标签 设置订阅消息的标签，可以指定消费某一类型的消息。默认*表示消费所有类型的消息，可以写多个，用“||”隔开，例如“tag1 || tag2 || tag3”。 加密传输 数据是否加密。 单击页面上方保存按钮，设置该“数据接入”基本参数。单击“保存”。 表14 “数据接入”参数说明 参数名称 参数说明 标签 该“数据接入”的显示标签名。长度不超过64个字节。 名称 该“数据接入”的名称，单击参数值后系统可根据“标签”取值自动生成。 描述 描述信息。 单击页面上方图标启用。 单击页面上方图标运行该“数据接入”。 Kafka接收到数据，表示设置成功。 在数据接入编辑页面上方单击停止图标，可停止运行。 在运行成功的提示框中复制数据推送地址，并进行保存。 也可在该数据接入运行后，单击页面上方复制数据推送地址。 在第三方外部数据源侧配置AppCube提供的数据推送地址，具体操作请参考第三方文档。 可选，可执行该步骤进行调试或故障定位。在数据接入编辑页面上方单击下载图标，下载日志，查看日志进行定位。 当数据源或者输出源配置信息有误时，可通过下载日志进行定位。 可选，进行消息跟踪。在数据接入编辑页面右侧选择“消息跟踪”页签，单击“启动”，可启动消息跟踪。 可实时查看上报的报文处理情况。 图16 消息跟踪 消息跟踪在一个周期（默认30分钟）后自动关闭。