本文介绍了任务步骤和其相关使用示例。

介绍

任务(Task)步骤定义了函数计算服务的函数调用信息,执行任务步骤会调用相应的函数。

任务步骤包含以下属性:

  • type(必需):task,表示该步骤是任务步骤。
  • name(必需):步骤名称。
  • resourceArn(必需):资源标识、当前支持函数、消息队列、 Serverless 工作流流程,例如 acs:fc:cn-shanghai:123456789:services/fnf_test/functions/hello。详情请参见服务集成
  • pattern(可选):跟集成服务的执行模式,不同的集成服务支持不同的执行模式。默认值为 requestResponse,枚举值如下:
    • requestResponse:同步等待任务执行结束后继续该步骤。
    • sync:异步提交任务后,等待任务执行结束,获得任务返回结果后继续该步骤。
    • waitForCallback:异步提交任务后(如完成函数调用后)当前步骤暂停,直到收到关于该任务的回调或该任务超时。
  • timeoutSeconds(可选):任务超时时间。如果任务执行超过指定的超时时间,则任务步骤执行超时。
  • end(可选):当前步骤结束后是否继续执行其后定义的步骤。
  • inputMappings(可选):输入映射。任务步骤的输入会被用作函数调用的 event(InvokeFunction)。
  • outputMappings(可选):输出映射。其中的 $local 是函数调用的返回结果,其格式必须是 JSON 对象类型。
    说明 如果未指定输出映射,本步骤默认将 $local 作为输出。
  • retry:定义了一组重试策略,其中每一个重试策略包含如下属性:
    • errors:定义了一个或者多个错误,详情请参见下面表格。
    • intervalSeconds:重试的初始间隔时间,最大值是 86400 秒,默认值是 1 秒。
    • maxIntervalSeconds:重试的最长间隔时间,最大值和默认值是 86400 秒。
    • maxAttempts:最多重试次数,默认值是 3 次。
    • multiplier:后一次重试比前一次重试间隔时间的倍数,默认值是 2。
  • catch:定义了一组捕获策略,其中每一个捕获策略包含如下属性:
    • errors:定义了一个或者多个错误,详情请参见下面表格。
    • goto:跳转目的步骤名称。
      说明 该目的步骤只能是和当前任务步骤并列的一个步骤。
表 1. 错误定义
函数执行状态 FC 响应 HTTP Code FC 响应 Serverless 工作流步骤失败(用于重试和捕获 是否需要重试
没有执行 429 ResourceExhausted FC.ResourceExhausted
没有执行 4xx,非 429 ServiceNotFound,FunctionNotFound,InvalidArgument 等 FC.ServiceNotFound,FC.FunctionNotFound,FC.InvalidArgument 等
不确定 500 InternalServerError FC.InternalServerError
没有执行 503 ResourceThrottled FC.ResourceThrottled
函数执行成功,返回错误。 200 用户自定义错误,包含 errorType。 errorType 根据业务决定
函数执行异常,返回错误。 200 无 errorType FC.Unknown
函数执行成功,返回非 JSON 对象。 200 无 errorType FC.InvalidOutput

示例

  • 简单任务步骤

    下面的示例流程包含一个任务步骤。

    • 如果流程的输入是 {"name": "function flow"},输出是 {"hello": "function flow"}
    • 如果流程没有指定输入或者输入不包含 name key,则任务步骤执行失败,从而导致整个流程执行失败。
    • 流程定义
      version: v1
      type: flow
      steps:
        - type: task
          name: hello
          resourceArn: acs:fc:{region}:{accountID}:services/fnf_test/functions/hello           
      说明 您需要将 resourceArn 替换成您账号下对应函数的资源标识。函数代码如下:
    • 函数定义
      import json
      
      class MyError(Exception):
        pass
      
      def handle(event, context):
        evt = json.loads(event)
        if "name" in evt:
          return {
            "hello": evt["name"]
          }
        else:
          raise MyError("My unhandled exception")          
  • 错误重试

    下面的示例会重试 MyError。如果流程没有指定输入或者输入不包含 name key,Serverless 工作流服务会根据重试策略多次重试失败任务调用。

    • 在第一次发生错误后等待 3 秒,然后再次调用函数。
    • 在第二次发生错误后等待 6 秒(intervalSeconds x multiplier),然后再次调用函数。
    • 在第三次发生错误后等待 12 秒(intervalSeconds x multiplier x multiplier),然后再次调用函数。
    • 经过三次重试后,函数仍然返回错误,重试次数超过 maxAttempts,任务步骤失败,流程以失败结束。
    version: v1
    type: flow
    steps:
      - type: task
        name: hello
        resourceArn: acs:fc:{region}:{accountID}:services/fnf_test/functions/hello
        retry:
          - errors:
            - MyError
            intervalSeconds: 3
            maxAttempts: 3
            multiplier: 2            
  • 错误捕获

    下面的示例会对捕获 MyError,然后跳转到 final 步骤。由于错误被成功捕获,最后流程以成功结束。

    version: v1
    type: flow
    steps:
      - type: task
        name: hello
        resourceArn: acs:fc:{region}:{accountID}:services/fnf_test/functions/hello
        catch:
          - errors:
            - MyError
            goto: final
      - type: pass
        name: pass1
      - type: pass
        name: final