别再傻傻用S3了！用Python+boto3玩转Cloudflare R2，免费额度真香（附完整代码）

Python开发者必看：Cloudflare R2存储的完整实战指南与S3成本优化方案

如果你正在使用AWS S3存储数据，每月账单是否让你感到压力？或者作为个人开发者，是否在寻找一个既经济实惠又不牺牲性能的云存储方案？Cloudflare R2可能是你一直在等待的解决方案。与S3兼容的API、慷慨的免费额度以及无缝的CDN集成，让R2成为开发者的新宠。

1. 为什么选择Cloudflare R2而非AWS S3？

在云存储领域，AWS S3长期占据主导地位，但成本问题一直困扰着许多开发者。让我们看看R2如何在这些关键指标上脱颖而出：

成本对比（每月）：

提示：R2的免费额度足够个人项目或小型应用使用，而S3从第一GB就开始计费

性能方面，R2与Cloudflare全球CDN网络深度集成，这意味着你的数据可以更快地送达终端用户。而S3需要额外配置CloudFront才能达到类似效果，这又会增加复杂性和成本。

2. 快速搭建你的第一个R2存储桶

2.1 准备工作

开始前，你需要：

• 一个Cloudflare账户（免费注册）
• 验证的支付方式（仅用于身份验证，不产生费用）
• Python 3.6+环境

2.2 创建存储桶

1. 登录Cloudflare仪表板
2. 导航至"R2"部分
3. 点击"创建存储桶"
4. 输入唯一名称（如"my-app-data"）
5. 选择位置（建议"自动"让Cloudflare优化）
6. 点击"创建"

# 安装必要的Python库 pip install boto3

3. 使用Python和boto3操作R2存储

R2最吸引人的特性之一是其与S3 API的兼容性。这意味着你可以继续使用熟悉的boto3库，只需稍作配置调整。

3.1 基础配置

首先，获取你的API凭证：

1. 在R2控制台点击"管理R2 API令牌"
2. 创建新令牌，选择适当权限
3. 记录访问密钥和秘密密钥

import boto3 from botocore.config import Config # R2配置 r2_config = Config( signature_version='s3v4', region_name='auto' # R2的特殊区域标识 ) s3 = boto3.client( 's3', aws_access_key_id='你的访问密钥', aws_secret_access_key='你的秘密密钥', endpoint_url='https://[account-id].r2.cloudflarestorage.com', config=r2_config )

3.2 文件上传的两种方式

方法一：简单上传

def upload_file(bucket_name, file_path, object_name=None): """上传文件到R2存储桶""" if object_name is None: object_name = file_path try: s3.upload_file(file_path, bucket_name, object_name) print(f"文件 {file_path} 上传成功") except Exception as e: print(f"上传失败: {e}") # 使用示例 upload_file('my-app-data', 'local-file.txt', 'remote-file.txt')

方法二：流式上传（适合大文件）

def upload_large_file(bucket_name, file_path, object_name=None, chunk_size=8*1024*1024): """分块上传大文件""" if object_name is None: object_name = file_path try: with open(file_path, 'rb') as data: s3.put_object( Bucket=bucket_name, Key=object_name, Body=data ) print(f"大文件 {file_path} 上传成功") except Exception as e: print(f"上传失败: {e}") # 使用示例 upload_large_file('my-app-data', 'large-video.mp4')

4. 高级操作与最佳实践

4.1 生成预签名URL

分享文件而不暴露凭证的安全方式：

def generate_presigned_url(bucket_name, object_name, expiration=3600): """生成临时访问URL""" try: url = s3.generate_presigned_url( 'get_object', Params={ 'Bucket': bucket_name, 'Key': object_name }, ExpiresIn=expiration ) return url except Exception as e: print(f"生成URL失败: {e}") return None # 生成1小时有效的URL file_url = generate_presigned_url('my-app-data', 'remote-file.txt') print(f"可分享的URL: {file_url}")

4.2 批量操作与监控

列出存储桶内容：

def list_bucket_contents(bucket_name, prefix=''): """列出存储桶中的对象""" try: contents = [] paginator = s3.get_paginator('list_objects_v2') for page in paginator.paginate(Bucket=bucket_name, Prefix=prefix): if 'Contents' in page: for obj in page['Contents']: contents.append(obj['Key']) return contents except Exception as e: print(f"列出内容失败: {e}") return [] # 使用示例 files = list_bucket_contents('my-app-data') print("存储桶中的文件:", files)

批量删除文件：

def delete_multiple_files(bucket_name, file_keys): """批量删除文件""" try: objects = [{'Key': key} for key in file_keys] s3.delete_objects( Bucket=bucket_name, Delete={'Objects': objects} ) print(f"成功删除 {len(file_keys)} 个文件") except Exception as e: print(f"删除失败: {e}") # 使用示例 delete_multiple_files('my-app-data', ['file1.txt', 'file2.txt'])

4.3 与Cloudflare CDN集成

R2与Cloudflare CDN的无缝集成是其一大优势。启用CDN缓存可以显著提高内容分发速度：

1. 在R2控制台选择你的存储桶
2. 点击"设置"选项卡
3. 启用"通过Cloudflare CDN提供内容"
4. 配置缓存规则（默认7天）

启用后，你的文件将通过Cloudflare的全球网络分发，自动获得：

• 边缘缓存
• DDoS保护
• 智能路由
• 压缩优化

5. 从S3迁移到R2的平滑过渡方案

如果你已经有S3存储的数据，迁移到R2可以分阶段进行：

5.1 双写策略过渡期

def upload_to_both(s3_client, r2_client, file_path, bucket_name, object_name=None): """同时写入S3和R2""" if object_name is None: object_name = file_path # 上传到S3 try: s3_client.upload_file(file_path, bucket_name, object_name) print(f"文件 {file_path} 上传到S3成功") except Exception as e: print(f"S3上传失败: {e}") # 上传到R2 try: r2_client.upload_file(file_path, bucket_name, object_name) print(f"文件 {file_path} 上传到R2成功") except Exception as e: print(f"R2上传失败: {e}") # 配置两个客户端 s3_client = boto3.client('s3') # 默认AWS配置 r2_client = ... # 之前的R2配置 # 使用示例 upload_to_both(s3_client, r2_client, 'data.json', 'my-bucket')

5.2 数据同步工具

对于大量现有数据，可以使用以下工具自动迁移：

• rclone：命令行工具，支持S3到R2的同步
• AWS DataSync：AWS官方迁移服务
• 自定义脚本：基于boto3的多线程迁移程序

# 使用rclone同步示例 rclone copy s3://source-bucket r2://target-bucket --progress

5.3 迁移后验证

确保所有数据完整迁移：

def verify_migration(s3_client, r2_client, bucket_name): """验证S3和R2内容一致""" s3_objects = set(obj['Key'] for obj in s3_client.list_objects_v2(Bucket=bucket_name).get('Contents', [])) r2_objects = set(obj['Key'] for obj in r2_client.list_objects_v2(Bucket=bucket_name).get('Contents', [])) only_in_s3 = s3_objects - r2_objects only_in_r2 = r2_objects - s3_objects if not only_in_s3 and not only_in_r2: print("验证通过：所有文件已成功迁移") return True else: if only_in_s3: print(f"以下文件仅在S3中存在: {only_in_s3}") if only_in_r2: print(f"以下文件仅在R2中存在: {only_in_r2}") return False # 使用示例 verify_migration(s3_client, r2_client, 'my-bucket')

6. 性能优化与成本控制技巧

6.1 存储分层策略

虽然R2没有像S3那样的存储类别，但你可以实现自己的"冷热"数据分离：

• 热数据：频繁访问，保留在R2
• 冷数据：不常访问，可考虑压缩后存储或迁移到更经济的方案

def archive_cold_data(source_bucket, target_bucket, older_than_days=30): """归档超过指定天数的冷数据""" from datetime import datetime, timedelta cutoff = datetime.now() - timedelta(days=older_than_days) for obj in s3.list_objects_v2(Bucket=source_bucket).get('Contents', []): if obj['LastModified'].replace(tzinfo=None) < cutoff: # 复制到归档存储桶 s3.copy_object( Bucket=target_bucket, Key=obj['Key'], CopySource={'Bucket': source_bucket, 'Key': obj['Key']} ) # 删除原文件 s3.delete_object(Bucket=source_bucket, Key=obj['Key']) print(f"已归档: {obj['Key']}") # 使用示例 archive_cold_data('my-app-data', 'my-archive-data')

6.2 请求优化

即使R2的请求费用较低，优化API调用仍能提升性能：

• 批量操作：合并多个小文件
• 缓存结果：减少重复列表请求
• 适当分页：处理大量对象时

def batch_upload(bucket_name, local_folder): """批量上传文件夹内容""" from os import listdir from os.path import isfile, join files = [f for f in listdir(local_folder) if isfile(join(local_folder, f))] for file in files: file_path = join(local_folder, file) s3.upload_file(file_path, bucket_name, file) print(f"已上传: {file}") # 使用示例 batch_upload('my-app-data', './data-files')

6.3 监控与告警

设置基本监控以避免意外费用：

def check_storage_usage(bucket_name, warning_threshold_gb=9): """检查存储使用量是否接近免费额度""" total_size = 0 paginator = s3.get_paginator('list_objects_v2') for page in paginator.paginate(Bucket=bucket_name): if 'Contents' in page: for obj in page['Contents']: total_size += obj['Size'] total_gb = total_size / (1024 ** 3) print(f"当前存储使用量: {total_gb:.2f} GB") if total_gb > warning_threshold_gb: print(f"警告: 存储使用量接近10GB免费额度!") return total_gb # 使用示例 check_storage_usage('my-app-data')

在实际项目中，我发现R2的API响应速度在亚太地区尤其出色，这得益于Cloudflare的全球网络布局。对于需要频繁读写的应用，R2相比S3能提供更一致的性能表现，特别是在处理大量小文件时。