AI 助理
备案控制台
文档
输入文档关键字查找
首页 云工作流 云工作流 CloudFlow（新版） 操作指南 工作流调度 EventBridge事件调度

EventBridge事件调度

更新时间：
产品详情
我的收藏

EventBridge事件调度包括云产品事件调度和自定义事件源事件调度。云产品事件调度支持包括弹性计算、存储服务、数据库、容器、大数据处理、可观测性服务及中间件服务在内的几乎所有阿里云官方事件源，而自定义事件源事件调度支持包括SLS、Kafka、RocketMQ、RabbitMQ等事件源的接入。本文介绍如何创建EventBridge事件调度以及工作流调度的高级功能。

创建EventBridge事件调度

云产品事件调度

云产品事件调度能够实现通过阿里云产品事件调度工作流执行，这些事件主要包括云监控事件、审计事件、云服务器事件、阿里云物联网IoT事件和部分云产品运维事件等云产品事件。本文以云服务器为例，介绍如何在控制台创建阿里云产品事件调度。

功能简介

您在工作流控制台提交创建工作流的请求后，根据工作流调度的配置信息，将自动在阿里云官方事件总线default下创建一个事件规则rule-created-by-fnf-随机串。创建完成后，您可以在工作流查看工作流调度信息，也可以在事件总线EventBridge控制台查看自动创建的事件规则信息。当事件源指定类型的事件投递到事件总线时，将调度该工作流调度关联的工作流执行一次。

注意事项

事件总线EventBridge的云服务专用事件总线default上最多只能创建10个事件规则，超过10个之后将无法再创建云产品事件调度。

前提条件

创建云产品事件调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择云服务器ECS，设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    ecs-schedule

    事件类型

    选择自定义事件类型或选中全部事件类型。如果选择自定义事件类型，您可以选择云服务器的一个或多个事件类型。

    自定义事件类型，ecs:Disk:ConvertToPostpaidCompleted 保留云盘

    事件模式内容

    不支持手动编辑，您在事件类型参数中选择事件类型后，事件模式内容自动填充。关于事件模式的信息，请参见事件模式

    {
 "source": [
 "acs.ecs"
 ],
 "type": [
 "ecs:Disk:ConvertToPostpaidCompleted"
 ]
}

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

HTTP调度

HTTP请求作为事件源通过事件总线与云工作流集成后，通过HTTP调度能够触发关联工作流执行。本文介绍如何在控制台创建HTTP调度。

注意事项

创建的自定义总线以及事件规则数量超过上限后，将无法再创建事件模式的HTTP调度。在单个阿里云账号单个地域维度下，关于创建工作流调度涉及的资源数量的限制，请参见使用限制。

前提条件

创建HTTP调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择HTTP/HTTPS 触发，设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    https-schedule

    请求类型

    选择HTTPS、HTTP或HTTP&HTTPS。

    HTTP

    请求方法

    选择支持的HTTP请求方法。取值说明如下。

    • GET

    • POST

    • PUT

    • DELETE

    • HEAD

    • PATCH

    GET

    安全配置

    选择安全配置的类型。取值说明如下。

    • 无需配置：无需进行安全配置，接收到的所有URL请求均可触发工作流执行。

    • IP网段：输入正确的IP地址或者IP网段。只有使用该IP地址或该IP网段内的IP地址访问的URL请求支持触发工作流执行。最多支持添加5个IP地址或者IP网段。

    • 安全域名：输入安全的域名信息。只有使用该域名访问的URL请求支持触发工作流执行。最多支持添加5个安全域名。

    IP网段：10.45.12.0/24

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

MNS调度

注意事项

  • 作为事件源的轻量消息队列（原 MNS）必须和创建工作流调度的工作流所在的地域相同。

  • 创建的事件流数量超过上限后，将无法再创建事件模式的MNS调度。在单个阿里云账号单个地域维度下，关于创建工作流调度涉及的资源数量的限制，请参见使用限制。

前提条件

创建工作流调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择轻量消息队列（原 MNS），设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    mns-schedule

    队列名称

    选择已创建的轻量消息队列（原 MNS）。

    MyQueue

    Base64 解码

    如果需要将MNS的数据解码后进行投递，请勾选开启 Base64 解码复选框。

    开启 Base64 解码

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

Kafka调度

注意事项

  • 作为事件源的消息队列Kafka版实例必须和创建工作流调度的工作流在相同的地域。

  • 创建的事件流数量超过上限后，将无法再创建Kafka调度。单个阿里云账号单个地域维度下，关于创建工作流调度涉及的资源数量的限制，请参见使用限制。

前提条件

创建Kafka工作流调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择消息队列 Kafka 版，设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    kafka-schedule

    Kafka 实例

    选择已创建的消息队列Kafka版实例。

    alikafka_pre-cn-i7m2t7t1****

    Topic

    选择已创建的消息队列Kafka版实例的Topic。

    topic1

    Group ID

    选择已创建的消息队列Kafka版实例的Group ID。

    重要

    请使用独立的Group ID来创建工作流调度，不要与已有的业务混用Group ID，否则会影响已有的消息收发。

    GID_group1

    消费任务并发数

    消费者的并发数量，取值范围为[1,Topic的分区数]。

    2

    消费位点

    选择消息的消费位点，即消息队列Kafka版从事件总线开始拉取消息的位置。取值说明如下。

    • 最早位点：从最早位点开始消费。

    • 最新位点：从最新位点开始消费。

    最新位点

    网络配置

    选择路由消息的网络类型。取值说明如下。

    • 默认网络：默认使用部署Kafka实例时选择的VPC ID和vSwitch ID。

    • 自建公网：需选择另外的专有网络VPC、交换机和安全组。

    默认网络

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

RocketMQ调度

注意事项

  • 作为事件源的消息队列RocketMQ版的实例必须和创建调度的工作流在相同的地域。

  • 创建的事件流数量超过上限后，将无法再创建事件模式的RocketMQ调度。单个阿里云账号单个地域维度下，关于创建工作流调度涉及的资源数量的限制，请参见使用限制。

前提条件

创建RocketMQ调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择消息队列 RocketMQ 版，设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    rocketmq-schedule

    RocketMQ 实例

    选择已创建的消息队列RocketMQ版的实例。

    MQ_INST_164901546557****_BX7****

    Topic

    选择已创建的消息队列RocketMQ版实例的Topic。

    topic1

    Tag

    填写消息过滤标签。只有收到包含此处设置的过滤标签字符串的消息时，才会触发工作流执行。

    tag

    Group ID

    选择已创建的消息队列RocketMQ版实例的Group ID。推荐您选择快速创建，自动创建以GID_FNF_TRIGGER_{uuid}_{timestamp} 命名的Group ID。

    重要

    请使用独立的Group ID来创建工作流调度，不要与已有的业务混用Group ID，否则会影响已有的消息收发。

    GID_group1

    消费位点

    选择消息的消费位点，即消息队列RocketMQ版从事件总线开始拉取消息的位置。取值说明如下。

    • 最新位点：从最新位点开始消费。

    • 最早位点：从最早位点开始消费。

    • 指定时间戳：从指定时间戳开始消费。

    最新位点

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

RabbitMQ调度

注意事项

  • 作为触发源的消息队列RabbitMQ版实例必须和创建调度的工作流在相同的地域。

  • 创建的事件流数量超过上限后，将无法再创建事件模式的RabbitMQ调度。单个阿里云账号单个地域维度下，关于创建工作流调度涉及的资源数量的限制，请参见使用限制。

前提条件

创建RabbitMQ调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择消息队列 RabbitMQ 版，设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    rabbitmq-schedule

    RabbitMQ 实例

    选择已创建的消息队列RabbitMQ版的实例。

    amqp-cn-i7m2l6m2****

    Vhost

    选择已创建的消息队列RabbitMQ版实例的Vhost。

    myhost-1

    Queue

    选择已创建的消息队列RabbitMQ版实例的Queue。

    myqueue-1

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

SLS调度

通过创建日志服务SLS调度（以下简称SLS调度）将日志服务SLS与云工作流连接起来，当有新日志产生时触发工作流执行，对日志进行处理。本文介绍如何在控制台创建SLS调度。

注意事项

  • 作为触发源的日志服务SLS项目必须和创建调度的工作流在相同的地域。

  • 创建的事件流数量超过上限后，将无法再创建事件模式的SLS调度。在单个阿里云账号单个地域维度下，关于创建工作流调度涉及的资源数量的限制，请参见使用限制。

前提条件

创建SLS调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择日志服务 SLS，设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    sls-schedule

    日志项目

    选择创建的日志服务SLS的Project。

    test-Project

    日志库

    选择创建的日志服务SLS的Logstore

    test-LogStore

    起始消费位点

    起始消费位点，可以选择最早或最新位点，也可以从指定时间开始消费。

    最新位点

    角色配置

    授权事件总线EventBridge使用此角色读取SLS日志内容。

    testRole

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

MQTT调度

云消息队列 MQTT 版作为事件源通过事件总线与云工作流调度集成后，通过云消息队列 MQTT 版调度能够触发关联工作流执行，通过云工作流可以对发布到云消息队列 MQTT 版的消息进行自定义处理。本文介绍如何在控制台创建MQTT调度。

注意事项

  • 作为触发源的云消息队列MQTT版实例必须和创建调度的工作流在相同的地域。

  • 创建的事件流数量超过上限后，将无法再创建事件模式的MQTT调度。在单个阿里云账号单个地域维度下，关于创建工作流调度涉及的资源数量的限制，请参见使用限制。

前提条件

创建MQTT调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择消息队列 MQTT版，设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    mqtt-schedule

    MQTT 实例

    选择已创建的MQTT实例。

    mqtt-xxx

    MQTT Topic

    选择已创建的MQTT实例的Topic。

    testTopic

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

DTS调度

数据传输服务 DTS（Data Transmission Service）作为事件源通过事件总线与云工作流集成后，通过DTS调度能够触发关联工作流的执行。本文介绍如何在流控制台创建DTS调度。

注意事项

  • 作为触发源的DTS数据订阅任务必须和创建调度的工作流在相同的地域。

  • 创建的事件流数量超过上限后，将无法再创建事件模式的DTS调度。在单个阿里云账号单个地域维度下，关于创建工作流调度涉及的资源数量的限制，请参见使用限制。

前提条件

创建DTS调度

  1. 登录云工作流控制台，然后在顶部菜单栏，选择地域。

  2. 在左侧导航栏，选择工作流列表，然后在工作流列表页面，单击目标工作流。

  3. 在工作流详情页面，单击工作流调度页签，然后单击创建工作流调度。

  4. 在创建工作流调度面板，调度类型选择数据传输服务 DTS，设置相关配置项，然后单击确定。基础配置项说明如下所示。

    配置项

    操作

    本文示例

    名称

    填写自定义的工作流调度名称。

    dts-schedule

    数据订阅任务

    已创建的数据订阅任务名称。

    dtsqntc2***

    消费组

    已创建的用于消费订阅任务的消费组名称。

    test

    账号

    创建消费组时设置的账号。

    test

    密码

    创建消费组时设置的密码。

    *******

    消费位点

    期望消费第一条数据的时间戳，消费位点必须在订阅实例的数据范围之内。

    2022-06-21 00:00:00

    关于推送配置、重试和死信等高级配置项说明，请参见工作流调度高级功能。创建完成后，您可以在工作流详情页面的工作流调度页签，根据界面提示查看、编辑、删除或禁用/启用工作流。

工作流调度高级功能

推送格式

用于指定Event参数中每个数据元素的格式。

  • CloudEvents：以通用格式描述事件数据的规范，旨在简化不同服务和平台间的事件声明和传输。

  • RawData：只投递CloudEvents中数据字段的内容，不包含CloudEvents格式中的其它元数据信息。

批量推送

开启批量推送后，需设置批量推送条数和推量推送间隔。

  • 批量推送条数：一次调用函数发送的最大批量消息条数，当积压的消息数量到达设定值时才会发送请求。取值范围为[1,10000]。

  • 批量推送间隔：调用函数的时间间隔，系统每到间隔时间点会将消息聚合后发给云工作流。取值范围为[0,15]，单位为秒。0秒表示无需等待，实时投递。

批量推送案例：

  • 案例一：

    您设置的批量推送条数为10条，每条消息大小为1 KB，推量推送间隔为15s。在10s内消息条数累积10条，则立即触发推送，无需等到15s再推送。

  • 案例二：

    您设置的批量推送条数为32条，每条消息大小为1 KB，推量推送间隔为15s。在15s内消息积累10条，则立即触发推送，无需等待消息积累到32条后再推送。

  • 案例三：

    您设置的批量推送条数为20条，每条消息大小为2 KB，推量推送间隔为15s。在10s内消息积累40条（消息累积大小40*2 KB=80 KB），达到工作流Input大小限制数，则立即触发推送，第一批聚合32条消息推送到云工作流，第二批聚合8条消息推送到云工作流。

重试策略

事件推送失败时，可按配置的重试策略进行重试。重试策略选项如下。

  • 退避重试：重试3次，每次重试的时间间隔为介于10s~20s的随机值。

  • 指数衰减重试：默认重试策略。重试176次，每次重试的时间间隔按照指数递增至512s，总计重试时间为24小时，即重试时间间隔为1s、2s、4s、8s、16s、32s、64s、128s、256s、512s……512s（共计167次间隔512s）。

容错策略

当发生错误时是否选择容错。

  • 允许容错

    请求失败且重试失败后，跳过此请求，继续处理下一条请求。

  • 禁止容错

    请求失败且重试失败后，消费任务阻塞。

死信队列

仅当开启允许容错时，可配置死信队列。

  • 如果启用死信队列，未被处理或超过重试次数的消息会被投递到目标服务中。云工作流支持的目标服务包括轻量消息队列（原 MNS）、消息队列RocketMQ版、消息队列Kafka版和事件总线EventBridge，您可以根据需求选择不同的队列类型。

  • 如果未启用死信队列，超过重试次数的消息会被丢弃。

上一篇：定时调度下一篇：工作流集成
文档推荐