基于 Serverless 工作流高并发批量解冻 OSS 文件

本文涉及的产品
Serverless 应用引擎 SAE,800核*时 1600GiB*时
函数计算FC,每月免费额度15元,12个月
简介:

前言

Serverless 工作流(Serverless Workflow,原函数工作流)是一个用来协调多个分布式任务执行的全托管 Serverless 云服务,致力于简化开发和运行业务流程所需要的任务协调、状态管理以及错误处理等繁琐工作,让用户聚焦业务逻辑开发。用户可以用顺序、分支、并行等方式来编排分布式任务,服务会按照设定好的顺序可靠地协调任务执行,跟踪每个任务的状态转换,并在必要时执行用户定义的重试逻辑,以确保工作流顺利完成。

函数计算 FC 是事件驱动的全托管计算服务,无需采购服务器和运维,只需上传代码就能实现高可用、高并发、弹性伸缩的后端服务。

本文介绍如何使用 Serverless 工作流来高并发大批量的解冻 oss 归档存储文件,使用工作流的优势:

  • 高并发
  • 错误自动重试,高可靠性
  • 每个步骤都有输入输出记录以及实时执行状态,高可观测性

应用中心一键部署

  1. 前往 Serverless 工作流应用中心 创建并部署 OSS Restore 应用。
    image
  2. 部署完成后执行流程 {stackName}-mainRestoreFlow-{suffix},输入:

    {
      "endpoint": "",
      "bucketName": "",
      "prefix": "",
      "marker": "",
      "maxKeys": 100,
      "pollInterval": 10,
      "workers": 10,
      "groupSize": 1
    }

    执行参数说明:

    • endpoint
      OSS endpoint
    • bucketName:
      OSS bucket 名称
    • prefix:
      OSS bucket 文件过滤前缀
    • maxKeys:
      OSS ListObjects 返回的最大文件数量 (这里不要超过 foreach 的并发限制,默认是 100)
    • pollInterval:
      轮询 OSS 文件 restore 状态的时间间隔(秒)
    • groupSize:
      一个任务步骤(对应一个函数)批量处理的文件数
    • workers:
      多个文件在一个函数中处理时,设定的处理线程池大小
    • marker:
      OSS ListObjects 的开始 marker

    详细信息可参考 ListObjects API

流程执行完毕后,会自动解冻所有符合过滤条件的归档存储文件。

原理

解冻文件主流程

主流程 mainRestoreFlow 主要做以下事情:

  1. 任务步骤 listArchiveFiles 调用 FC 函数 listArchiveFilesmarker 开始执行 ListObjects 列举指定 maxKeys 数量,前缀为 prefix 的 OSS 文件。
  2. 调用子流程 restoreFlow 对获取的文件列表进行解冻,并返回下一次列举的起点 maker
  3. 选择步骤 checkEnd 检测是否已完成列举 bucket 中所有的文件,若没有,跳转到 listArchiveFiles 中从下一个 marker 开始继续执行,否则结束。

主流程定义,请参考 流程定义语言

version: v1
type: flow
steps:
- type: pass
  name: init
  outputMappings:
    - source: $input.endpoint
      target: endpoint
    - source: $input.bucketName
      target: bucketName
    - source: $input.prefix
      target: prefix
    - source: $input.maxKeys
      target: maxKeys
    - source: $input.pollInterval
      target: pollInterval
    - source: $input.marker
      target: marker
    - source: $input.workers
      target: workers
    - source: $input.groupSize
      target: groupSize

# List archive files from marker
- type: task
  name: listArchiveFiles
  resourceArn: acs:fc:::services/<serviceName>/functions/listArchiveFiles
  outputMappings:
    - source: $local.bucketName
      target: bucketName
    - source: $local.filesGroup
      target: filesGroup
    - source: $local.marker
      target: marker
    - source: $local.end
      target: end
    - source: $local.empty
      target: empty
    - source: $local.archiveFilesCount
      target: archiveFilesCount

# Check whether file restore success, if not, retry check
- type: choice
  name: checkEmpty
  choices:
    # If list archive files not empty
    - condition: $.empty == "false"
      steps:
        # Invoke subflow restore to restore listed files
        - type: task
          name: invokeRestoreFlow
          resourceArn: acs:fnf:::flow/<restoreFlow>
          pattern: sync
          serviceParams:
            Input: $
  default:
    goto: checkEnd

# Check list files ended
- type: choice
  name: checkEnd
  choices:
    - condition: $.end == "true"
      goto: success
  default:
    goto: listArchiveFiles

# success
- type: pass
  name: success

解冻文件子流程

子流程 restoreFlow 主要做以下事情:

  1. 并行循环步骤 foreach,并行的对主流程中传入的文件列表生成多个解冻任务。
  2. 解冻任务 restoreTask 调用 FC 函数 restore 对文件列表进行解冻。
  3. 所有解冻任务提交完成后,循环执行:

    • 等待步骤 Wait 等待一段时间
    • 任务步骤 GetJobStatus 获取所有文件的解冻状态
    • 选择步骤 CheckJobComplete 判断是否全部解冻完成,若完成执行结束,否则跳转到 wait 步骤继续循环检测。

子流程定义如下:

version: v1
type: flow
steps:
  - type: foreach
    name: retoreForeach
    iterationMapping:
      collection: $.filesGroup
      item: files
    steps:
      # Invoke restore function
      - type: task
        name: restoreTask
        resourceArn: !Ref OSSRestoreService/restoreFunction
        retry:
          - errors:
              - FC.ResourceThrottled
              - FC.ResourceExhausted
              - FC.InternalServerError
              - FC.Unknown
              - FnF.TaskTimeout
            intervalSeconds: 1
            maxAttempts: 10
            multiplier: 1.5
            maxIntervalSeconds: 10

  # Wait interval for poll files restore status
  - type: wait
    name: Wait
    duration: $.pollInterval

  # Get file restore status
  - type: task
    name: GetJobStatus
    resourceArn: !Ref OSSRestoreService/restoreStatusFunction

  # Check whether file restore success, if not, retry check
  - type: choice
    name: CheckJobComplete
    inputMappings:
      - target: status
        source: $local.status
    choices:
      - condition: $.status == "success"
        goto: JobSucceeded
      - condition: $.status == "running"
        goto: Wait

  - type: succeed
    name: JobSucceeded
    outputMappings:
      - target: filesGroup
        source: $input.filesGroup
      - target: marker
        source: $input.marker

以上应用代码可参考 oss-restore

总结

使用 Serverless 工作流能极大减少重复的流程控制开发,让错误重试更加容易,以及提供实时的执行进度查询。这里提供的解冻 oss 应用,完全可以照搬到其它的需求上,以实现高并发高可靠性的应用。

欢迎加钉钉群 23116481 交流:
image

相关实践学习
基于函数计算一键部署掌上游戏机
本场景介绍如何使用阿里云计算服务命令快速搭建一个掌上游戏机。
建立 Serverless 思维
本课程包括: Serverless 应用引擎的概念, 为开发者带来的实际价值, 以及让您了解常见的 Serverless 架构模式
目录
相关文章
|
21天前
|
存储 运维 Serverless
函数计算产品使用问题之在YAML文件中配置了环境变量,但在PHP代码中无法读取到这些环境变量,是什么原因
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。
|
18天前
|
运维 Serverless 应用服务中间件
Serverless 应用引擎产品使用合集之关于OSS映射目录的大小限制,如何可以跳过
阿里云Serverless 应用引擎(SAE)提供了完整的微服务应用生命周期管理能力,包括应用部署、服务治理、开发运维、资源管理等功能,并通过扩展功能支持多环境管理、API Gateway、事件驱动等高级应用场景,帮助企业快速构建、部署、运维和扩展微服务架构,实现Serverless化的应用部署与运维模式。以下是对SAE产品使用合集的概述,包括应用管理、服务治理、开发运维、资源管理等方面。
Serverless 应用引擎产品使用合集之关于OSS映射目录的大小限制,如何可以跳过
|
18天前
|
运维 监控 Serverless
Serverless 应用引擎产品使用合集之需要上传多个文件,该如何处理
阿里云Serverless 应用引擎(SAE)提供了完整的微服务应用生命周期管理能力,包括应用部署、服务治理、开发运维、资源管理等功能,并通过扩展功能支持多环境管理、API Gateway、事件驱动等高级应用场景,帮助企业快速构建、部署、运维和扩展微服务架构,实现Serverless化的应用部署与运维模式。以下是对SAE产品使用合集的概述,包括应用管理、服务治理、开发运维、资源管理等方面。
|
21天前
|
人工智能 运维 Serverless
函数计算产品使用问题之上传模型文件占用的是什么空间
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。
|
22天前
|
JSON Serverless 对象存储
函数计算产品使用问题之如何创建一个同时具有HTTP触发器和OSS触发器的函数
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。
|
1月前
|
运维 Java Serverless
Serverless 应用引擎产品使用合集之是否提供工具来给OSS配置HTTPS证书
阿里云Serverless 应用引擎(SAE)提供了完整的微服务应用生命周期管理能力,包括应用部署、服务治理、开发运维、资源管理等功能,并通过扩展功能支持多环境管理、API Gateway、事件驱动等高级应用场景,帮助企业快速构建、部署、运维和扩展微服务架构,实现Serverless化的应用部署与运维模式。以下是对SAE产品使用合集的概述,包括应用管理、服务治理、开发运维、资源管理等方面。
|
1月前
|
存储 运维 Serverless
Serverless 应用引擎产品使用合集之如何访问相同地域的OSS
阿里云Serverless 应用引擎(SAE)提供了完整的微服务应用生命周期管理能力,包括应用部署、服务治理、开发运维、资源管理等功能,并通过扩展功能支持多环境管理、API Gateway、事件驱动等高级应用场景,帮助企业快速构建、部署、运维和扩展微服务架构,实现Serverless化的应用部署与运维模式。以下是对SAE产品使用合集的概述,包括应用管理、服务治理、开发运维、资源管理等方面。
|
21天前
|
弹性计算 运维 Serverless
函数计算产品使用问题之想往nas传个文件,该怎么操作
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。
|
22天前
|
JSON 运维 Serverless
函数计算产品使用问题之如何实现数据的读取和修改,而不需要每次都从OSS下载完整的数据
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。
|
22天前
|
运维 网络协议 Serverless
函数计算产品使用问题之怎么将生成的图片保存到oss上
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。

热门文章

最新文章

相关产品

  • 函数计算