★ FEATURED ARTICLE 【免费下载链接】aws-cdk-examplesExample projects using the AWS CDK项目地址https://gitcode.com/gh_mirrors/aw/aws-cdk-examples点击查看免费下载本篇文章以aws-cdk-examples仓库中的 C# 示例csharp/stepfunctions-job-poller为核心讲解如何用 AWS CDKC#/.NET 语言绑定构建一个「提交任务 → 定期轮询 → 判断完成 → 汇报状态」的 Step Functions 状态机。读完本文你将掌握 Step Functions 的Activity、Wait、Choice、Fail等核心状态在 CDK 中的声明方式理解轮询工作流的 JSONPath 数据流转并能复现该示例的完整构建与部署流程。一、示例概览Job Poller 状态机解决什么问题StepfunctionsJobPoller作业轮询器是一个典型的异步任务编排场景外部系统如 AWS Batch接收任务后需要一段时间才能执行完成调用方无法立刻拿到结果因此需要周期性地查询任务状态直到任务成功或失败为止。仓库中的 C# 示例构建了这样一个基本工作流README.md 的描述This example creates a basic Stepfunction workflow that starts a task and then listens at regular intervals for completion, then reports completion and status.即启动一个任务 → 按固定间隔监听其完成情况 → 汇报最终完成状态。这是 AWS Step Functions 官方文档中经典的 Job Poller作业轮询模式常用于取代在 EC2/Lambda 中自行编写轮询循环的胶水代码把等待与重试的复杂度下沉到状态机层面。需要说明的是该示例在仓库中被标记为Experimental实验性README.md 顶部的 Stability Banner它构建在标注为 Experimental 的 Construct Library 之上可能未针对最新破坏性变更更新如果构建失败可以到仓库提交 issue 反馈。二、项目结构与运行前提示例位于仓库的csharp/stepfunctions-job-poller目录核心文件如下文件作用README.md示例说明与构建、部署命令cdk.jsonCDK App 入口配置Program.csCDK App 入口实例化 Stack 并执行SynthStepfunctionsJobPollerStack.cs状态机工作流的核心定义StepfunctionsJobPollerConstruct.cs一个简单的自定义 Construct 示例S3 桶集合StepfunctionsJobPoller.csproj.NET 工程文件与 CDK 依赖声明cdk.json 指定了 CDK Toolkit 如何执行本应用{ app: dotnet run --project src/StepfunctionsJobPoller/StepfunctionsJobPoller.csproj }也就是说cdk命令会通过dotnet run启动src/StepfunctionsJobPoller工程。该工程StepfunctionsJobPoller.csproj面向net8.0依赖两个关键包Amazon.CDK.Lib版本区间[2.262.0,3.0.0)允许滚动升级到 3.x 之前的最新 2.xConstructs版本10.*。运行前提是本地已安装 .NET SDKnet8.0、Node.js 与 AWS CDK CLInpm install -g aws-cdk并已配置好 AWS 凭证。三、构建与部署README 核心命令按照 README.md 的说明需要先进入示例根目录然后依次执行dotnet build src cdk deploy StepfunctionsJobPollerStackdotnet build src编译解决方案 StepfunctionsJobPoller.sln包含Debug/Release等多种配置验证 C# 代码与 CDK 依赖可正常构建cdk deploy StepfunctionsJobPollerStack将名为StepfunctionsJobPollerStack的 Stack 部署到当前 AWS 账号/区域。Stack 名称来自 Program.cs 中的new StepfunctionsJobPollerStack(app, StepfunctionsJobPollerStack)。此外CDK 常规调试命令同样适用cdk synth可在本地生成 CloudFormation 模板输出到cdk.out目录cdk diff可对比已部署状态cdk destroy可清理资源。四、状态机核心源码解析整个工作流定义集中在 StepfunctionsJobPollerStack.cs 中仅约 60 行代码就声明了一个完整的轮询状态机。下面按状态逐一拆解。4.1 定义两个 Activity任务的提交与检查var submitJobActivity new Activity(this, SubmitJob); var checkJobActivity new Activity(this, CheckJob);Activity是 Step Functions 的一种工作单元它不是由 Lambda 直接触发而是由外部 Worker 程序Activity Worker主动向 Step Functions 拉取任务令牌并执行。这里声明了两个 ActivitySubmitJob负责接收并提交一个作业CheckJob负责查询作业的执行状态。在真实系统中这两个 Activity 通常由运行在 EC2、ECS 或本地数据中心的 Worker 实现它们通过 Step Functions API 轮询任务令牌GetActivityTask/SendTaskSuccess/SendTaskFailure。4.2 Submit Job提交作业并保存任务 IDvar submitJob new StepFunctionsInvokeActivity(this, Submit Job, new StepFunctionsInvokeActivityProps { Activity submitJobActivity, ResultPath $.guid });StepFunctionsInvokeActivity表示调用一个 Activity的任务状态对应 CloudFormation 中的Task类型。关键配置Activity指定要调用的 Activity即submitJobActivityResultPath $.guid把 Activity 的返回结果写入输入 JSON 的guid字段。也就是说Submit Job 返回的作业 ID 会被存入$.guid供后续状态读取。关于ResultPath的语义见第六节数据流转。4.3 Wait X Seconds按动态秒数等待var waitX new Wait(this, Wait X Seconds, new WaitProps { Time WaitTime.SecondsPath($.wait_time) });Wait状态让状态机暂停一段时间。此处通过WaitTime.SecondsPath($.wait_time)实现动态等待等待秒数不写死而是从输入数据的$.wait_time字段读取。这意味着调用方可以在每次执行时传入不同的轮询间隔例如 5 秒、30 秒状态机无需改动即可适配不同的任务耗时预期。4.4 Get Job Status查询作业状态var getStatus new StepFunctionsInvokeActivity(this, Get Job Status, new StepFunctionsInvokeActivityProps { Activity checkJobActivity, InputPath $.guid, ResultPath $.status });这一步调用checkJobActivity查询作业状态配置了两个 JSONPathInputPath $.guid只把上一步写入的作业 ID 作为输入传给 Activity WorkerResultPath $.status把查询结果写入$.status字段。经过此状态$.status便承载了当前作业状态如SUCCEEDED/FAILED/ 运行中。4.5 Job Complete?Choice 分支判断var isComplete new Choice(this, Job Complete?);Choice状态根据条件表达式将执行流路由到不同分支。本示例注册了三个分支见 4.7 的 Chain 组装$.status等于FAILED→ 进入Job Failed失败状态$.status等于SUCCEEDED→ 进入Get Final Job Status汇报最终状态其他值如RUNNING、PENDING→ 回退到Wait X Seconds形成轮询循环。4.6 Job Failed失败终止var jobFailed new Fail(this, Job Failed, new FailProps { Cause AWS Batch Job Failed, Error DescribeJob returned FAILED });Fail状态使执行立即终止并标记为失败同时携带两个错误描述字段Error机器可读的错误码DescribeJob returned FAILED取自 AWS BatchDescribeJobsAPI 的语义Cause人类可读的失败原因AWS Batch Job Failed。这两个信息会出现在 Step Functions 控制台的执行详情中便于排查。4.7 Get Final Job Status汇报最终状态var finalStatus new StepFunctionsInvokeActivity(this, Get Final Job Status, new StepFunctionsInvokeActivityProps { Activity checkJobActivity, InputPath $.guid });当作业确认成功后最后一次调用checkJobActivity拉取最终状态细节例如 Batch 的退出码、输出路径等作为状态机的成功输出。4.8 用 Chain 组装整条执行链var chain Chain .Start(submitJob) .Next(waitX) .Next(getStatus) .Next(isComplete .When(Condition.StringEquals($.status, FAILED), jobFailed) .When(Condition.StringEquals($.status, SUCCEEDED), finalStatus) .Otherwise(waitX)); new StateMachine(this, StateMachine, new StateMachineProps { Definition chain, Timeout Duration.Seconds(30) });Chain是 CDK 提供的工作流拼接 API把各状态串成有向图Start(submitJob)起点为 Submit Job.Next(waitX)→.Next(getStatus)线性执行 等待 → 查状态.Next(isComplete...)进入 Choice 分支When(Condition.StringEquals(...), ...)注册两个条件分支Otherwise(waitX)表示默认作业仍未完成回跳Wait X Seconds从而构成循环——这是轮询模式的核心。最后创建StateMachine并设置Timeout Duration.Seconds(30)整个状态机的最长执行时间为 30 秒超出即终止。在真实 Batch 任务场景中这个超时通常需要按任务的最坏执行时长调大例如分钟级。五、轮询循环状态机如何自己等自己理解本示例的关键在于 Choice 的Otherwise(waitX)回跳。整个执行流程为Submit Job提交作业得到$.guidWait X Seconds按$.wait_time指定的秒数休眠Get Job Status查询状态并写入$.statusJob Complete?判断SUCCEEDED→Get Final Job Status→ 成功结束FAILED→Job Failed→ 失败结束其他状态 → 回到第 2 步Wait X Seconds再次休眠后重新查询。因此等待与重试逻辑完全由 Step Functions 托管开发者无需自建定时任务或 SQS 重试队列状态机天然具备等待—查询—再等待的循环能力且每次等待时长可通过$.wait_time动态调整轮询上限则由状态机级Timeout兜底。从源码结构看WaitTime.SecondsPathStepfunctionsJobPollerStack.cs是让该循环具备灵活性的关键它把等待时长与输入数据解耦属于 CDK 中WaitTime的路径Path用法与固定WaitTime.Duration(...)相对。六、状态数据流转ResultPath / InputPath 的配合本示例值得单独强调的是 JSONPath 字段的流转设计均可在 StepfunctionsJobPollerStack.cs 中逐一对应状态输入处理输出处理效果Submit Job默认全部输入ResultPath $.guid输入新增guid字段Wait X Seconds默认WaitTime.SecondsPath($.wait_time)读取wait_time决定休眠秒数Get Job StatusInputPath $.guidResultPath $.status仅用guid调用 Activity结果写入statusJob Complete?默认Condition.StringEquals($.status, ...)读取status路由分支在 CDK 中InputPath控制把数据的哪一部分作为当前状态输入ResultPath控制把状态结果写到输入数据的哪个位置。二者配合使guid作业 ID与status作业状态这两个字段在状态间有序累积最终形成完整执行上下文。这也是 Step Functions 相比普通队列Lambda 组合更可观测、可重放的原因之一。七、多语言实现对照同一模式在仓库中的其他体现job poller是仓库中的经典示例同一模式在多种 CDK 语言绑定中均有实现可对照学习TypeScript 版typescript/stepfunctions-job-poller/index.ts用tasks.LambdaInvoke代替Activity通过 submit.py 与 check_status.py 两个 Python Lambda 模拟提交/查状态并额外用 EventBridge Rulecron(0 18 ? * MON-FRI *)定时触发状态机。其 Choice 分支条件与 C# 版完全一致SUCCEEDED→ 最终状态、FAILED→ Fail、否则回Wait。Java 版StepFunctionsJobPollerStack.java结构与 C# 版几乎一一对应Activity、StepFunctionsInvokeActivity、WaitTime.secondsPath、Chain.start(...).next(...)说明该模式在 CDK 各语言绑定间保持一致的抽象。Go 版go/step-functions/step-function.go同样以 Lambda 任务 WaitChoice组装轮询状态机并输出状态机 ARN。对照可见无论使用ActivityC#/Java还是LambdaInvokeTypeScript/Go提交 → 等待 → 查状态 → 条件回跳的轮询骨架是通用的Activity适合已有外部 Worker 的场景LambdaInvoke适合任务逻辑本身就是 Lambda 的场景。八、部署后的验证与注意事项部署完成后可在 AWS 控制台的 Step Functions 页面找到名为StateMachine逻辑 ID的状态机。由于本示例使用Activity而非 Lambda 直接驱动实际执行需要配套的 Activity Worker 才能跑通——Worker 通过 AWS SDK 调用GetActivityTask拉取任务并回传结果若没有 Worker执行会一直停留在等待 Activity 响应的状态。实践建议本示例的Timeout为 30 秒StepfunctionsJobPollerStack.cs接入真实任务如 AWS Batch时应按任务的最坏执行时长重新设置轮询间隔通过输入数据的$.wait_time动态控制发起执行时传入合理的秒数即可调整探测频率示例标记为 Experimental升级 CDK 依赖后请先运行dotnet build src验证再部署。小结csharp/stepfunctions-job-poller用约 60 行 CDK 代码展示了 Step Functions 最实用的模式之一——作业轮询Activity对接外部 WorkerWait Choice实现动态间隔的循环探测Fail处理失败路径ResultPath/InputPath负责字段流转。配合 TypeScript、Java、Go 的对照实现你可以快速掌握在 AWS CDK 中声明状态机的通用套路并平滑迁移到自己的异步任务编排场景中。赞分享【免费下载链接】aws-cdk-examplesExample projects using the AWS CDK项目地址https://gitcode.com/gh_mirrors/aw/aws-cdk-examples点击查看免费下载相关推荐aws-cdk-examples .NET Examples 实战指南用 C 编写、构建并部署 AWS CDK 应用aws cdk examples .NET Examples 实战指南用 C 编写、构建并部署 AWS CDK 应用 本文基于 csharp/README.maws-cdk-examples使用 .NET CDK 创建 Elastic Beanstalk 应用与环境含 Platform ARN 实战aws cdk examples使用 .NET CDK 创建 Elastic Beanstalk 应用与环境含 Platform ARN 实战 导读 本指使用 Serverless Framework 构建 AWS Step Functions 状态机Ruby 版周末出游编排实战使用 Serverless Framework 构建 AWS Step Functions 状态机Ruby 版周末出游编排实战 导读 本文以仓库中的 aw示例工程上一篇CANN ops-nn 中 aclnnEmbeddingDenseBackward 算子接口详解Embedding 反向梯度的稠密聚合实现与两段式调用实践下一篇使用 kOps Manifest 与 API 自定义和管理 Kubernetes 集群创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
阅读完成 · 觉得有帮助？
