如何使用AWS CDK在Step Functions中多次复用Task State

报错原因 #

AWS CDK 中 Step Functions 的每个状态实例（比如LambdaInvoke生成的Task对象），直接对应状态机定义里的单个唯一状态节点：

• 普通线性Task状态仅允许配置1个next跳转目标
• 重复引用同一个sfn_step_2、sfn_step_3实例编排1->2->3->2->3流程时，第一次给sfn_step_2配置next为sfn_step_3后，第二次尝试从sfn_step_3跳转回同一个sfn_step_2实例，相当于给已经配置过next的状态重复绑定跳转规则，会直接触发CDK校验，抛出State '2nd-One' already has a next state错误。

注意：多个独立的Task状态节点可以调用同一个Lambda函数，不会重复创建Lambda资源，只会在状态机里多生成一个调用同个Lambda的状态节点，无额外冗余成本。

修复方案 #

方案1：为每一次执行创建独立Task实例 #

哪怕两次执行调用的逻辑完全一致，只要是流程中不同位置的节点，就创建独立的Task实例即可，示例代码如下：

# 第一步节点
sfn_step_1 = tasks.LambdaInvoke(self, "1st-one",
                                        lambda_function=first_lambda,
                                        output_path="$.Payload",
                                        payload=stepfunctions.TaskInput.from_object({
                                            "payload.$": "$",
                                        })
                                        )
# 第一轮执行的step2、step3
sfn_step_2_first = tasks.LambdaInvoke(self, "2nd-One-First",
                                        lambda_function=second_lambda,
                                        output_path="$.Payload",
                                        payload=stepfunctions.TaskInput.from_object({
                                            "payload.$": "$",
                                        })
                                        )
sfn_step_3_first = tasks.LambdaInvoke(self, "3rd-one-First",
                                        lambda_function=third_lambda,
                                        output_path="$.Payload",
                                        payload=stepfunctions.TaskInput.from_object({
                                            "payload.$": "$",
                                        })
                                        )
# 第二轮执行的step2、step3，和第一轮调用同一个Lambda即可
sfn_step_2_second = tasks.LambdaInvoke(self, "2nd-One-Second",
                                        lambda_function=second_lambda,
                                        output_path="$.Payload",
                                        payload=stepfunctions.TaskInput.from_object({
                                            "payload.$": "$",
                                        })
                                        )
sfn_step_3_second = tasks.LambdaInvoke(self, "3rd-one-Second",
                                        lambda_function=third_lambda,
                                        output_path="$.Payload",
                                        payload=stepfunctions.TaskInput.from_object({
                                            "payload.$": "$",
                                        })
                                        )

# 链式编排不会触发校验错误
state_machine_definition = sfn_step_1.next(sfn_step_2_first)\
    .next(sfn_step_3_first)\
    .next(sfn_step_2_second)\
    .next(sfn_step_3_second)

方案2：封装工厂函数减少重复代码 #

如果重复的Task配置完全一致，可以封装简单的工厂函数批量生成Task实例，避免重复编写配置代码：

def create_lambda_invoke_task(scope, task_id, target_lambda):
    return tasks.LambdaInvoke(scope, task_id,
                                lambda_function=target_lambda,
                                output_path="$.Payload",
                                payload=stepfunctions.TaskInput.from_object({
                                    "payload.$": "$",
                                })
                            )

# 批量生成所有流程节点
sfn_step_1 = create_lambda_invoke_task(self, "1st-one", first_lambda)
sfn_step_2_first = create_lambda_invoke_task(self, "2nd-One-First", second_lambda)
sfn_step_3_first = create_lambda_invoke_task(self, "3rd-one-First", third_lambda)
sfn_step_2_second = create_lambda_invoke_task(self, "2nd-One-Second", second_lambda)
sfn_step_3_second = create_lambda_invoke_task(self, "3rd-one-Second", third_lambda)

# 正常编排流程即可
state_machine_definition = sfn_step_1.next(sfn_step_2_first)\
    .next(sfn_step_3_first)\
    .next(sfn_step_2_second)\
    .next(sfn_step_3_second)

补充：如果是动态次数的循环逻辑，不需要硬编码多个Task节点，可以使用Map状态或者Choice+执行计数器的方式实现循环跳转；固定次数的顺序重复执行，使用上述独立实例方案最直接，合成无报错，运行逻辑也完全符合预期。

内容的提问来源于stack exchange，提问作者back-new