内网带宽跑不满百 Gbps？OSS 三维调优实测帮你拉到位

export OSS_ACCESS_KEY_ID=<ALIBABA_CLOUD_ACCESS_KEY_ID>
export OSS_ACCESS_KEY_SECRET=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>

package main
import (
    "context"
    "flag"
    "fmt"
    "io"
    "log"
    "time"
    "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
    "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)
// 定义全局变量，用于存储命令行参数
var (
    region      string // OSS 存储桶所在的区域
    endpoint    string // OSS 服务的访问域名
    bucketName  string // OSS 存储桶的名称
    objectName  string // OSS 对象的名称
    chunkSize   int64  // 分块大小（字节）
    prefetchNum int    // 预取数量
)
// 初始化函数，用于解析命令行参数
func init() {
    flag.StringVar(&region, "region", "", "The region in which the bucket is located.")
    flag.StringVar(&endpoint, "endpoint", "", "The domain names that other services can use to access OSS.")
    flag.StringVar(&bucketName, "bucket", "", "The `name` of the bucket.")
    flag.StringVar(&objectName, "object", "", "The `name` of the object.")
    flag.Int64Var(&chunkSize, "chunk-size", 0, "The chunk size, in bytes")
    flag.IntVar(&prefetchNum, "prefetch-num", 0, "The prefetch number")
}
func main() {
    // 解析命令行参数
    flag.Parse()
    // 检查必要的参数是否提供
    if len(bucketName) == 0 {
        flag.PrintDefaults()
        log.Fatalf("invalid parameters, bucket name required")
    }
    if len(region) == 0 {
        flag.PrintDefaults()
        log.Fatalf("invalid parameters, region required")
    }
    // 配置 OSS 客户端
    cfg := oss.LoadDefaultConfig().
        WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()). // 使用环境变量中的凭证
        WithRegion(region) // 设置区域
    // 如果提供了自定义 endpoint，则设置
    if len(endpoint) > 0 {
        cfg.WithEndpoint(endpoint)
    }
    // 创建 OSS 客户端
    client := oss.NewClient(cfg)
    // 打开 OSS 文件
    f, err := client.OpenFile(context.TODO(), bucketName, objectName, func(oo *oss.OpenOptions) {
        oo.EnablePrefetch = true      // 启用预取
        oo.ChunkSize = chunkSize      // 设置分块大小
        oo.PrefetchNum = prefetchNum  // 设置预取数量
        oo.PrefetchThreshold = int64(0) // 设置预取阈值
    })
    if err != nil {
        log.Fatalf("open fail, err:%v", err)
    }
    // 记录开始时间
    startTick := time.Now().UnixNano() / 1000 / 1000
    // 将文件内容读取并丢弃（用于测试读取速度）
    written, err := io.Copy(io.Discard, f)
    // 记录结束时间
    endTick := time.Now().UnixNano() / 1000 / 1000
    if err != nil {
        log.Fatalf("copy fail, err:%v", err)
    }
    // 计算读取速度（MiB/s）
    speed := float64(written/1024/1024) / (float64(endTick-startTick) / 1000)
    // 输出平均读取速度
    fmt.Printf("average speed:%.2f(MiB/s)\n", speed)
}

go run down_object.go -bucket yourbucket -endpoint oss-cn-hangzhou-internal.aliyuncs.com  -object 100GB.file -region cn-hangzhou -chunk-size 419430400 -prefetch-num 256

time ossutil --parallel=256 --part-size=468435456 --endpoint=oss-cn-hangzhou-internal.aliyuncs.com cp oss://cache-test/EcsTest/100GB.file /dev/shm/100GB.file

--region cn-hangzhou
[root@xxx]# time ossutil --parallel=256 --part-size=468435456 --endpoint=cn-hangzhou-internal.oss-data-acc.aliyuncs.com cp oss://yacai-oss-cache-test/EcsTest/100GB.file /dev/shm/100GB.file
Total num: 1, size: 102,400,000,000. Dealt num: 0, OK size: 468,435,456, Progre
Total num: 1, size: 102,400,000,000. Dealt num: 0, OK size: 47,311,981,056, Pro
Succeed: Total num: 1, size: 102,400,000,000. OK num: 1(download 1 objects).
average speed 2942359000(byte/s)
34.802357(s) elapsed
real    0m34.808s
user    1m28.397s
sys     50m4.987s

ossfs cache-test /mnt/cache-test -ourl=http://oss-cn-hangzhou-internal.aliyuncs.com  -oparallel_count=128 -omultipart_size=32

ossfs cache-test /mnt/cache-test -ourl=http://oss-cn-hangzhou-internal.aliyuncs.com  -odirect_read -odirect_read_prefetch_chunks=128 -odirect_read_chunk_size=32

import oss2
import time
import os
import threading
from io import BytesIO
# 配置参数（通过环境变量获取）
OSS_CONFIG = {
    "bucket_endpoint": os.environ.get('OSS_BUCKET_ENDPOINT', 'oss-cn-hangzhou-internal.aliyuncs.com'),  # 默认Endpoint示例
    "bucket_name": os.environ.get('OSS_BUCKET_NAME', 'bucket_name'),  #Bucket 名称
    "access_key_id": os.environ['ACCESS_KEY_ID'],         # RAM用户ACCESS_KEY_ID
    "access_key_secret": os.environ['ACCESS_KEY_SECRET']  # RAM用户ACCESS_KEY_SECRET
}
# 初始化 OSS Bucket 对象
def __bucket__():
    auth = oss2.Auth(OSS_CONFIG["access_key_id"], OSS_CONFIG["access_key_secret"])
    return oss2.Bucket(
        auth, 
        OSS_CONFIG["bucket_endpoint"], 
        OSS_CONFIG["bucket_name"], 
        enable_crc=False
    )
# 获取对象大小
def __get_object_size(object_name):
    simplifiedmeta = __bucket__().get_object_meta(object_name)
    return int(simplifiedmeta.headers['Content-Length'])
# 获取远程模型最后修改时间
def get_remote_model_mmtime(model_name):
    return __bucket__().head_object(model_name).last_modified
# 列出远程模型文件
def list_remote_models(ext_filter=('.ckpt',)):  # 添加默认扩展名过滤
    dir_prefix = ""
    output = []
    for obj in oss2.ObjectIteratorV2(
        __bucket__(),
        prefix=dir_prefix,
        delimiter='/',
        start_after=dir_prefix,
        fetch_owner=False
    ):
        if not obj.is_prefix():
            _, ext = os.path.splitext(obj.key)
            if ext.lower() in ext_filter:
                output.append(obj.key)
    return output
# 分段下载线程函数
def __range_get(object_name, buffer, offset, start, end, read_chunk_size, progress_callback, total_bytes):
    chunk_size = int(read_chunk_size)
    with __bucket__().get_object(object_name, byte_range=(start, end)) as object_stream:
        s = start
        while True:
            chunk = object_stream.read(chunk_size)
            if not chunk:
                break
            buffer.seek(s - offset)
            buffer.write(chunk)
            s += len(chunk)
            # 计算已下载字节数并调用进度回调
            if progress_callback:
                progress_callback(s - start, total_bytes)
# 读取远程模型（增加进度回调可选参数）
def read_remote_model(
    checkpoint_file, 
    start=0, 
    size=-1, 
    read_chunk_size=2*1024*1024,  # 2MB
    part_size=256*1024*1024,      # 256MB
    progress_callback=None        # 进度回调
):
    time_start = time.time()
    buffer = BytesIO()
    obj_size = __get_object_size(checkpoint_file)
    end = (obj_size if size == -1 else start + size) - 1
    s = start
    tasks = []
    # 进度计算
    total_bytes = end - start + 1
    downloaded_bytes = 0
    while s <= end:
        current_end = min(s + part_size - 1, end)
        task = threading.Thread(
            target=__range_get,
            args=(checkpoint_file, buffer, start, s, current_end, read_chunk_size, progress_callback, total_bytes)
        )
        tasks.append(task)
        task.start()
        s += part_size
    for task in tasks:
        task.join()
    time_end = time.time()
    # 显示总耗时
    print(f"Downloaded {checkpoint_file} in {time_end - time_start:.2f} seconds.")
     # 计算并打印下载文件的大小（单位：GB）
    file_size_gb = obj_size / (1024 * 1024 * 1024)
    print(f"Total downloaded file size: {file_size_gb:.2f} GB")
    buffer.seek(0)
    return buffer
# 进度回调函数
def show_progress(downloaded, total):
    progress = (downloaded / total) * 100
    print(f"Progress: {progress:.2f}%", end="\r")
# 调用示例
if __name__ == "__main__":
    # 调用 list_remote_models 方法列出远程模型文件
    models = list_remote_models()
    print("Remote models:", models)
    if models:
        # 选择第一个模型文件进行下载
        first_model = models[0]
        buffer = read_remote_model(first_model, progress_callback=show_progress)
        print(f"\nDownloaded {first_model} to buffer.")