4. 对象存储跨云复制实战：阿里OSS → AWS S3

跨云复制这件事，说起来简单，做起来坑不少。我最早接触这个需求是在一个出海项目里——国内业务数据存在阿里OSS，海外节点必须同步到AWS S3。当时第一反应是「找个工具同步不就行了？」结果发现，数据一致性、延迟、成本，每个维度都有讲究。

今天咱们就手把手走一遍：利用阿里OSS事件通知触发函数计算，把数据实时写入AWS S3。这套方案我用了两年多，线上跑了上百TB数据，还算稳。

4.1 整体架构：事件驱动 + 无服务器

先看架构图，心里有个全局印象。

阿里OSS → AWS S3 跨云复制架构 阿里云 OSS Bucket 源存储桶 事件通知 ObjectCreated:* 函数计算 (FC) Python 3.9 + boto3 + oss2 RAM角色 临时凭证 VPC & NAT 公网出站 AWS S3 Bucket 目标存储桶 IAM Role S3FullAccess HTTPS / 公网 💡 核心流程：OSS上传 → 事件通知 → FC接收 → 下载文件 → 上传S3 → 确认结果 延迟通常在 1-5 秒内，取决于文件大小和网络链路

说白了，就是三个环节：事件触发 → 函数处理 → 跨云写入。阿里OSS一有文件上传，事件通知就推给函数计算，函数计算拿到事件后，下载文件再上传到AWS S3。全程无服务器，不用管机器，按调用量付费。

关键设计点：

  • 事件通知只监听 ObjectCreated:*，避免删除操作触发
  • 函数计算需要公网出站能力（VPC + NAT 或 弹性公网IP）
  • AWS侧用IAM角色生成临时凭证，别写死AccessKey

4.2 阿里OSS事件通知配置

先配OSS的事件通知。我个人习惯在控制台操作，但生产环境建议用Terraform或ROS模板，方便版本管理。

4.2.1 创建事件通知规则

进入OSS控制台 → Bucket列表 → 选择目标Bucket → 进入「数据处理 → 事件通知」。点「创建规则」，参数如下：

配置项 推荐值 说明
规则名称 oss-to-s3-replication 见名知意
事件类型 ObjectCreated:* 覆盖Put、Post、Copy、CompleteMultipartUpload
资源描述 prefix/* 可以指定目录前缀，比如 data/*
接收端 函数计算（FC） 选择已创建的函数

小技巧：如果只同步特定目录，资源描述写 images/* 或 backup/*，能减少无效触发。我在项目中遇到过全桶监听导致函数被频繁调用，后来加了前缀过滤，调用量降了60%。

4.2.2 事件格式说明

事件通知推给FC时，Payload长这样：

{
  "events": [
    {
      "eventName": "ObjectCreated:Put",
      "eventSource": "acs:oss",
      "eventTime": "2024-01-15T10:30:00.000Z",
      "eventVersion": "1.0",
      "oss": {
        "bucket": {
          "name": "my-source-bucket",
          "arn": "acs:oss:cn-hangzhou:123456789:my-source-bucket"
        },
        "object": {
          "key": "data/report-2024-01-15.csv",
          "size": 1048576,
          "eTag": "abc123def456",
          "type": "Normal"
        }
      },
      "region": "cn-hangzhou"
    }
  ]
}

嗯，这里要注意：事件是数组结构，一次可能推送多条。函数里一定要遍历处理，别只取 events[0]。

4.3 函数计算（FC）代码实现

接下来是核心——FC函数怎么写。我用Python 3.9，依赖 oss2 和 boto3。先看完整代码，再拆开讲。

import oss2
import boto3
import json
import os
from urllib.parse import unquote

# 阿里OSS配置
OSS_ENDPOINT = os.environ['OSS_ENDPOINT']
OSS_BUCKET = os.environ['OSS_BUCKET']

# AWS S3配置
AWS_REGION = os.environ['AWS_REGION']
S3_BUCKET = os.environ['S3_BUCKET']
AWS_ROLE_ARN = os.environ['AWS_ROLE_ARN']

def handler(event, context):
    # 解析OSS事件
    events = json.loads(event)
    for evt in events.get('events', []):
        oss_info = evt.get('oss', {})
        bucket_name = oss_info.get('bucket', {}).get('name')
        object_key = unquote(oss_info.get('object', {}).get('key', ''))
        
        if not object_key:
            continue
            
        print(f"Processing: {bucket_name}/{object_key}")
        
        # 1. 从OSS下载文件到临时目录
        auth = oss2.Auth(
            context.credentials.access_key_id,
            context.credentials.access_key_secret
        )
        bucket = oss2.Bucket(auth, OSS_ENDPOINT, bucket_name)
        local_path = f"/tmp/{object_key.replace('/', '_')}"
        
        # 确保目录存在
        os.makedirs(os.path.dirname(local_path), exist_ok=True)
        
        bucket.get_object_to_file(object_key, local_path)
        print(f"Downloaded to {local_path}")
        
        # 2. 获取AWS临时凭证（STS）
        sts_client = boto3.client(
            'sts',
            region_name=AWS_REGION,
            aws_access_key_id=context.credentials.access_key_id,
            aws_secret_access_key=context.credentials.access_key_secret,
            aws_session_=context.credentials.security_
        )
        
        assumed_role = sts_client.assume_role(
            RoleArn=AWS_ROLE_ARN,
            RoleSessionName='oss-to-s3-session'
        )
        
        creds = assumed_role['Credentials']
        
        # 3. 上传到S3
        s3_client = boto3.client(
            's3',
            region_name=AWS_REGION,
            aws_access_key_id=creds['AccessKeyId'],
            aws_secret_access_key=creds['SecretAccessKey'],
            aws_session_=creds['SessionToken']
        )
        
        # 保持相同的对象键
        s3_client.upload_file(local_path, S3_BUCKET, object_key)
        print(f"Uploaded to s3://{S3_BUCKET}/{object_key}")
        
        # 4. 清理临时文件
        os.remove(local_path)
    
    return {
        'statusCode': 200,
        'body': json.dumps('Sync completed')
    }

4.3.1 代码要点拆解

这段代码有几个地方值得细说：

  • 临时凭证获取：FC运行时会自动注入 context.credentials，包含临时AccessKey。别自己配AK/SK，不安全。
  • AWS STS AssumeRole：用阿里云的临时凭证去调用AWS STS，换取S3的写权限。这样两边都不用存长期密钥。
  • 文件路径处理：/tmp 目录是FC的临时存储，最大512MB。大文件要注意，我遇到过2GB文件把磁盘写满的情况。
  • 对象键保持一致：源和目标用相同的Key，方便后续管理和对比。

⚠️ 曾经踩过的坑：有一次我忘了对 object_key 做URL解码，结果文件名里带中文的文件全部上传失败。阿里OSS事件通知里的Key是URL编码的，必须用 unquote 解码。

4.4 AWS侧配置：IAM角色与S3权限

AWS这边要创建一个IAM角色，让阿里云FC能通过STS扮演它。配置步骤：

  1. 创建IAM角色：选择「Another AWS account」，填入阿里云账号ID（可以在FC的RAM角色里找到）。
  2. 附加权限策略：至少需要 s3:PutObject 和 s3:PutObjectAcl。如果要做清单对比，再加 s3:ListBucket。
  3. 设置信任关系：信任策略里指定阿里云RAM角色的ARN，限制只有那个角色能AssumeRole。

信任策略示例：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::123456789012:root"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "your-external-id"
        }
      }
    }
  ]
}

安全建议：一定要用 ExternalId 做条件限制。ExternalId 是阿里云RAM角色的唯一标识，加了之后就算别人知道你的Role ARN也扮演不了。

4.5 网络成本优化：公网流量怎么省

跨云复制最大的成本往往不是函数计算调用费，而是公网出站流量。阿里OSS到AWS S3走公网，每GB大概0.5-0.8元，量大了真扛不住。

我分享几个实战中验证过的优化手段：

优化手段 节省幅度 适用场景
OSS内网传输到同区域ECS，再由ECS转发 30%-50% 已有ECS实例，且OSS和ECS同区域
使用阿里云CDN加速上传到S3 15%-25% 文件较大（>100MB），且对延迟不敏感
压缩后再传输 40%-70% 文本类文件（JSON、CSV、日志）
增量同步 + 跳过未变更文件 50%-80% 文件频繁更新但变化量小

你想想看，如果每天同步1TB数据，走公网直传一个月要花1.5万左右。用ECS中转加压缩，能压到5000以内。这笔账算下来，优化投入的研发成本两周就回本了。

4.6 异常处理与监控

线上跑起来之后，最怕悄无声息地失败。我建议至少做三件事：

  • FC错误日志：开启阿里云日志服务（SLS），把FC的stdout和stderr都收集起来。搜索关键字 ERROR 或 Failed 就能定位问题。
  • 死信队列（DLQ）：事件通知可以配置投递失败后写入MNS队列。这样即使FC挂了，事件也不会丢。
  • S3事件通知反向校验：在S3侧也开启事件通知，对比两边的事件数量。如果差值超过阈值就告警。

我曾经遇到过一个隐蔽的问题：FC函数执行超时（默认60秒），大文件下载+上传耗时超过限制，导致事件被丢弃。后来把FC超时时间改成300秒，同时加了断点续传逻辑，问题才解决。

4.7 性能与成本实测数据

最后给一组我在生产环境实测的数据，供你参考：

文件大小 平均耗时 FC调用次数/天 月成本（估算）
100KB以下 0.8-1.2秒 50万 约300元
1MB-10MB 2-5秒 10万 约500元
100MB-500MB 15-40秒 1万 约800元
1GB以上 60-180秒 1000 约600元

注意，这里只算了FC的计算费用，没算OSS的流量费。流量费取决于你的优化手段，差距很大。

← 上一章 📖 返回目录 下一章 →