华为对象云存储OBS开发记录

Posted on 2022-08-24 In Works

对象存储是一种存储结构，它将数据作为对象进行管理，而不是文件或块，每个对象包括数据、元数据和唯一标识符。
在软件开发中，对象存储通常用于存储非结构化数据，如图片、视频、日志文件等。
使用对象存储，可以通过RESTful API从任何地方访问数据，这使得对象存储非常适合用于构建云原生应用，或者需要大规模数据存储和访问的应用。
国内外许多云服务提供商，都提供对象存储服务：
- 国内
- 国外
本文基于OBS Java API，从头开始实现了一个最简单的分段上传大文件的功能。

开发指南

OBS提供了REST（Representational State Transfer）风格API，支持通过HTTP/HTTPS请求调用，调用方法请参见如何调用API。

同时OBS还提供多种编程语言的SDK，SDK的使用方法请参见SDK参考。

安装依赖

首先，需要在项目中添加华为云OBS的Java SDK依赖。如果项目是Maven项目，可以在pom.xml文件中添加以下依赖：

<dependency>
    <groupId>com.moses.obs</groupId>
    <artifactId>spring-boot-start-obs</artifactId>
    <version>1.0.0</version>
</dependency>

鉴权方式

OBS通过用户帐号中的AK和SK进行签名验证，确保通过授权的帐号才能访问指定的OBS资源。

AK：Access Key ID，接入键标识，用户在对象存储服务系统中的接入键标识，一个接入键标识唯一对应一个用户，一个用户可以同时拥有多个接入键标识。对象存储服务系统通过接入键标识识别访问系统的用户。
SK：Secret Access Key，安全接入键，用户在对象存储服务系统中的安全接入键，是用户访问对象存储服务系统的密钥，用户根据安全接入键和请求头域生成鉴权信息。安全接入键和接入键标识一一对应。

分段上传

对于超过5G的大文件采用分段上传方式，分段上传分为如下3个步骤：

步骤1 初始化分段上传任务（ObsClient.initiateMultipartUpload）。
步骤2 逐个或并行上传段（ObsClient.uploadPart）。
步骤3 合并段（ObsClient.completeMultipartUpload）或取消分段上传任务（ObsClient.abortMultipartUpload）。

初始化上传任务

使用分段上传方式传输数据前，必须先通知OBS初始化一个分段上传任务。该操作会返回一个OBS服务端创建的全局唯一标识（Upload ID），用于标识本次分段上传任务，可以根据这个唯一标识来发起相关的操作，如取消分段上传任务、列举分段上传任务、列举已上传的段等。
通过ObsClient.initiateMultipartUpload初始化一个分段上传任务：

// Endpoint以北京四为例，其他地区请按实际情况填写。
String endPoint = "https://obs.cn-north-4.myhuaweicloud.com";
String ak = System.getenv("ACCESS_KEY_ID");
String sk = System.getenv("SECRET_ACCESS_KEY_ID");

// 创建ObsClient实例
ObsClient obsClient = new ObsClient(ak, sk, endPoint);

InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest("bucketname", "objectname");
ObjectMetadata metadata = new ObjectMetadata();
metadata.addUserMetadata("property", "property-value");
metadata.setContentType("text/plain");
request.setMetadata(metadata);
InitiateMultipartUploadResult result = obsClient.initiateMultipartUpload(request);

String uploadId = result.getUploadId();
System.out.println("\t" + uploadId);

上传段

初始化一个分段上传任务之后，可以根据指定的对象名和Upload ID来分段上传数据。
每一个上传的段都有一个标识它的号码——分段号（Part Number，范围是1~10000）。
对于同一个Upload ID，该分段号不但唯一标识这一段数据，也标识了这段数据在整个对象内的相对位置。
如果同一个分段号上传了新的数据，那么OBS上已有的这个段号的数据将被覆盖。除了最后一段以外，其他段的大小范围是100KB-5GB，最后段大小范围是0-5GB。
每个段不需要按顺序上传，甚至可以在不同进程、不同机器上上传，OBS会按照分段号排序组成最终对象。
通过ObsClient.uploadPart上传段：

// Endpoint以北京四为例，其他地区请按实际情况填写。
String endPoint = "https://obs.cn-north-4.myhuaweicloud.com";
String ak = System.getenv("ACCESS_KEY_ID");
String sk = System.getenv("SECRET_ACCESS_KEY_ID");
String uploadId = "upload id from initiateMultipartUpload";
// 创建ObsClient实例
ObsClient obsClient = new ObsClient(ak, sk, endPoint);

List<PartEtag> partEtags = new ArrayList<PartEtag>();
// 上传第一段
UploadPartRequest request = new UploadPartRequest("bucketname", "objectname");
// 设置Upload ID
request.setUploadId(uploadId);
// 设置分段号，范围是1~10000，
request.setPartNumber(1);
// 设置将要上传的大文件
request.setFile(new File("localfile"));

// 设置分段大小
request.setPartSize(5 * 1024 * 1024L);
UploadPartResult result = obsClient.uploadPart(request);
partEtags.add(new PartEtag(result.getEtag(), result.getPartNumber()));

// 上传第二段
request = new UploadPartRequest("bucketname", "objectname");
// 设置Upload ID
request.setUploadId(uploadId);
// 设置分段号
request.setPartNumber(2);
// 设置将要上传的大文件
request.setFile(new File("localfile"));
// 设置第二段的段偏移量
request.setOffset(5 * 1024 * 1024L);
// 设置分段大小
request.setPartSize(5 * 1024 * 1024L);
result = obsClient.uploadPart(request);
partEtags.add(new PartEtag(result.getEtag(), result.getPartNumber()));

合并段

所有分段上传完成后，需要调用合并段接口来在OBS服务端生成最终对象。
在执行该操作时，需要提供所有有效的分段列表（包括分段号和分段ETag值）；OBS收到提交的分段列表后，会逐一验证每个段的有效性。当所有段验证通过后，OBS将把这些分段组合成最终的对象。
通过ObsClient.completeMultipartUpload合并段：

// Endpoint以北京四为例，其他地区请按实际情况填写。
String endPoint = "https://obs.cn-north-4.myhuaweicloud.com";
String ak = System.getenv("ACCESS_KEY_ID");
String sk = System.getenv("SECRET_ACCESS_KEY_ID");
String uploadId = "upload id from initiateMultipartUpload";
// 创建ObsClient实例
ObsClient obsClient = new ObsClient(ak, sk, endPoint);

List<PartEtag> partEtags = new ArrayList<PartEtag>();
// 第一段
PartEtag part1 = new PartEtag();
part1.setPartNumber(1);
part1.seteTag("etag1");
partEtags.add(part1);

// 第二段
PartEtag part2 = new PartEtag();
part2.setPartNumber(2);
part2.setEtag("etag2");
partEtags.add(part2);

CompleteMultipartUploadRequest request = new CompleteMultipartUploadRequest("bucketname", "objectname", uploadId, partEtags);

obsClient.completeMultipartUpload(request);

完整代码

import com.obs.services.ObsClient;
import com.obs.services.model.InitiateMultipartUploadRequest;
import com.obs.services.model.InitiateMultipartUploadResult;
import com.obs.services.model.UploadPartRequest;
import com.obs.services.model.UploadPartResult;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class MultipartUploadSample {
    private static final String endPoint = "your-endpoint";
    private static final String accessKey = "your-access-key";
    private static final String secretKey = "your-secret-key";
    private static final String bucketName = "your-bucket-name";
    private static final String objectKey = "your-object-key";
    private static final String filePath = "your-file-path";
    private static final int partSize = 5 * 1024 * 1024;

    public static void main(String[] args) {
        ObsClient obsClient = new ObsClient(accessKey, secretKey, endPoint);
        try {
            // Step 1: Initialize.
            InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucketName, objectKey);
            InitiateMultipartUploadResult result = obsClient.initiateMultipartUpload(request);

            // Step 2: Upload parts.
            File file = new File(filePath);
            long fileLength = file.length();
            long partCount = fileLength % partSize == 0 ? fileLength / partSize : fileLength / partSize + 1;

            if (partCount > 10000) {
                throw new RuntimeException("Total parts count should not exceed 10000");
            } else {
                System.out.println("Total parts count " + partCount + "\n");
            }

            List<UploadPartResult> parts = new ArrayList<>();
            for (int i = 0; i < partCount; i++) {
                long offset = i * partSize;
                long currPartSize = (i + 1 == partCount) ? fileLength - offset : partSize;
                uploadPart(obsClient, bucketName, objectKey, result.getUploadId(), file, offset, currPartSize, i + 1, parts);
            }

            // Step 3: Complete.
            obsClient.completeMultipartUpload(bucketName, objectKey, result.getUploadId(), parts);
            System.out.println("Multipart upload success. \n");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            obsClient.close();
        }
    }

    private static void uploadPart(
            ObsClient obsClient,
            String bucketName,
            String objectKey,
            String uploadId,
            File file,
            long offset,
            long partSize,
            int partNumber,
            List<UploadPartResult> parts) throws IOException {
        FileInputStream fis = new FileInputStream(file);
        try {
            fis.skip(offset);
            UploadPartRequest uploadPartRequest = new UploadPartRequest();
            uploadPartRequest.setBucketName(bucketName);
            uploadPartRequest.setObjectKey(objectKey);
            uploadPartRequest.setUploadId(uploadId);
            uploadPartRequest.setFile(fis);
            uploadPartRequest.setPartSize(partSize);
            uploadPartRequest.setPartNumber(partNumber);
            UploadPartResult uploadPartResult = obsClient.uploadPart(uploadPartRequest);
            parts.add(uploadPartResult);
            System.out.println("Part#" + partNumber + " done\n");
        } finally {
            fis.close();
        }
    }
}