Java实现阿里云OSS文件上传链路

init → uploading → verifying → available
  │        │             │
  └──────→ failed ←──────┘
uploading → cancelled

create table file_asset (
    id bigint primary key auto_increment,
    digest char(32) not null,
    size_bytes bigint not null,
    original_name varchar(255) not null,
    mime_type varchar(128) not null,
    object_key varchar(512) not null,
    bucket_name varchar(128) not null,
    status varchar(24) not null,
    visibility varchar(16) not null default 'private',
    created_by bigint not null,
    created_at datetime(3) not null,
    updated_at datetime(3) not null,
    unique key uk_asset_digest_size (digest, size_bytes),
    unique key uk_asset_object_key (bucket_name, object_key),
    key idx_asset_status (status)
);
create table file_reference (
    id bigint primary key auto_increment,
    file_id bigint not null,
    biz_type varchar(64) not null,
    biz_id varchar(64) not null,
    created_by bigint not null,
    created_at datetime(3) not null,
    unique key uk_reference (file_id, biz_type, biz_id),
    constraint fk_reference_asset foreign key (file_id) references file_asset(id)
);
create table multipart_upload_session (
    id varchar(64) primary key,
    upload_id varchar(256) not null,
    file_id bigint null,
    object_key varchar(512) not null,
    digest char(32) not null,
    size_bytes bigint not null,
    original_name varchar(255) not null,
    mime_type varchar(128) not null,
    biz_type varchar(64) not null,
    biz_id varchar(64) not null,
    part_size_bytes int not null,
    status varchar(24) not null,
    expires_at datetime(3) not null,
    created_by bigint not null,
    created_at datetime(3) not null,
    updated_at datetime(3) not null,
    unique key uk_multipart_upload_id (upload_id)
);
create table multipart_upload_part (
    session_id varchar(64) not null,
    part_number int not null,
    etag varchar(128) not null,
    size_bytes bigint not null,
    updated_at datetime(3) not null,
    primary key (session_id, part_number),
    constraint fk_part_session foreign key (session_id) references multipart_upload_session(id)
);

<dependency>
  <groupid>com.aliyun.oss</groupid>
  <artifactid>aliyun-sdk-oss</artifactid>
  <version>3.17.4</version>
</dependency>

app:
  oss:
    endpoint: https://xxx
    region: xxx
    bucket: xxx
    preview-domain: https://xxx
    access-key-id: xxx
    access-key-secret: xxx
    folder: files
    preview-expire-seconds: 300
    multipart:
      threshold-bytes: 104857600       # 100 mib，达到后使用分片
      part-size-bytes: 10485760         # 10 mib
      max-parts: 10000
      session-expire-minutes: 1440

package com.example.file.config;

import com.aliyun.oss.oss;
import com.aliyun.oss.ossclientbuilder;
import jakarta.annotation.predestroy;
import lombok.data;
import org.springframework.boot.context.properties.configurationproperties;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;

/** oss 连接和上传策略配置。 */
@data
@configuration
@configurationproperties(prefix = "app.oss")
public class ossproperties {
  private string endpoint;
  private string region;
  private string bucket;
  private string previewdomain;
  private string accesskeyid;
  private string accesskeysecret;
  private string folder = "files";
  private long previewexpireseconds = 300;
  private multipart multipart = new multipart();

  @data
  public static class multipart {
    private long thresholdbytes = 100l * 1024 * 1024;
    private int partsizebytes = 10 * 1024 * 1024;
    private int maxparts = 10_000;
    private long sessionexpireminutes = 24 * 60;
  }
}

@configuration
class ossclientconfiguration {
  private final ossproperties properties;
  private oss client;

  ossclientconfiguration(ossproperties properties) {
    this.properties = properties;
  }

  @bean
  oss ossclient() {
    client = new ossclientbuilder().build(
        properties.getendpoint(),
        properties.getaccesskeyid(),
        properties.getaccesskeysecret());
    return client;
  }

  @predestroy
  void shutdown() {
    if (client != null) {
      client.shutdown();
    }
  }
}

package com.example.file.oss;

import com.example.file.config.ossproperties;
import com.aliyun.oss.httpmethod;
import com.aliyun.oss.oss;
import com.aliyun.oss.model.abortmultipartuploadrequest;
import com.aliyun.oss.model.completemultipartuploadrequest;
import com.aliyun.oss.model.generatepresignedurlrequest;
import com.aliyun.oss.model.initiatemultipartuploadrequest;
import com.aliyun.oss.model.objectmetadata;
import com.aliyun.oss.model.partetag;
import com.aliyun.oss.model.uploadpartrequest;
import com.aliyun.oss.model.uploadpartresult;
import java.io.inputstream;
import java.net.url;
import java.time.duration;
import java.util.arraylist;
import java.util.date;
import java.util.list;
import org.springframework.stereotype.component;

/** 对阿里云 oss sdk 的最小封装，屏蔽 bucket 和 sdk 请求对象。 */
@component
public class ossgateway {
  private final oss client;
  private final ossproperties properties;

  public ossgateway(oss client, ossproperties properties) {
    this.client = client;
    this.properties = properties;
  }

  /** 普通上传。调用方必须提供长度，避免 sdk 退化为不可控的流式行为。 */
  public void putobject(string objectkey, inputstream input, long contentlength, string contenttype) {
    objectmetadata metadata = new objectmetadata();
    metadata.setcontentlength(contentlength);
    metadata.setcontenttype(contenttype);
    client.putobject(properties.getbucket(), objectkey, input, metadata);
  }

  /** 初始化 multipart upload，返回 oss uploadid。 */
  public string initiatemultipart(string objectkey, string contenttype) {
    initiatemultipartuploadrequest request =
        new initiatemultipartuploadrequest(properties.getbucket(), objectkey);
    objectmetadata metadata = new objectmetadata();
    metadata.setcontenttype(contenttype);
    request.setobjectmetadata(metadata);
    return client.initiatemultipartupload(request).getuploadid();
  }

  /** 上传单个分片，etag 是完成合并时的权威凭证。 */
  public partetag uploadpart(
      string objectkey, string uploadid, int partnumber, inputstream input, long partsize) {
    uploadpartrequest request = new uploadpartrequest();
    request.setbucketname(properties.getbucket());
    request.setkey(objectkey);
    request.setuploadid(uploadid);
    request.setpartnumber(partnumber);
    request.setpartsize(partsize);
    request.setinputstream(input);
    uploadpartresult result = client.uploadpart(request);
    return result.getpartetag();
  }

  /** 按 partnumber 升序完成合并。parts 必须来自服务端持久化的 etag。 */
  public void completemultipart(string objectkey, string uploadid, list<partetag> parts) {
    // oss sdk 3.17.4 可能在完成前排序该列表，不能传入 list.of()/stream.tolist() 的不可变结果。
    list<partetag> mutableparts = new arraylist<>(parts);
    client.completemultipartupload(
        new completemultipartuploadrequest(
            properties.getbucket(), objectkey, uploadid, mutableparts));
  }

  /** 取消未完成会话，释放 oss 临时分片。 */
  public void abortmultipart(string objectkey, string uploadid) {
    client.abortmultipartupload(
        new abortmultipartuploadrequest(properties.getbucket(), objectkey, uploadid));
  }

  public boolean exists(string objectkey) {
    return client.doesobjectexist(properties.getbucket(), objectkey);
  }

  public long objectsize(string objectkey) {
    return client.getobjectmetadata(properties.getbucket(), objectkey).getcontentlength();
  }

  /** 为私有对象生成短期 get url；url 不写入 file_asset。 */
  public string presignget(string objectkey, duration validity) {
    if (validity.isnegative() || validity.iszero()) {
      throw new illegalargumentexception("签名有效期必须大于 0");
    }
    date expiration = new date(system.currenttimemillis() + validity.tomillis());
    generatepresignedurlrequest request = new generatepresignedurlrequest(
        properties.getbucket(), objectkey, httpmethod.get);
    request.setexpiration(expiration);
    url url = client.generatepresignedurl(request);
    return url.toexternalform();
  }

  /**
   * 生成无有效期的公开访问地址。
   *
   * <p>该方法只拼接配置好的 https 域名和对象键，不生成签名参数，也不会改变对象 acl。
   * 只有 bucket/object 已明确允许匿名读取时才能调用。
   */
  public string publicurl(string objectkey) {
    if (objectkey == null || objectkey.isblank()) {
      throw new illegalargumentexception("对象键不能为空");
    }
    if (properties.getpreviewdomain() == null || properties.getpreviewdomain().isblank()) {
      throw new illegalstateexception("未配置公开访问域名");
    }
    string folder = properties.getfolder().replaceall("^/+|/+$", "");
    if (!objectkey.startswith(folder + "/")) {
      throw new illegalargumentexception("对象键不在受管目录下");
    }
    java.net.uri domain = java.net.uri.create(properties.getpreviewdomain().trim());
    if (!"https".equalsignorecase(domain.getscheme()) || domain.gethost() == null) {
      throw new illegalargumentexception("公开访问域名必须是 https url");
    }
    return properties.getpreviewdomain().trim().replaceall("/+$", "") + "/" + objectkey;
  }
}

/** 返回带过期时间的预览结果，便于前端提前刷新。 */
public previewurlresult generatepreviewurl(long fileid, long operatorid) {
  fileasset asset = assetrepository.findavailablebyid(fileid)
      .orelsethrow(() -> new fileuploadexception("file_not_found", "文件不存在"));
  permissionservice.assertreadable(asset, operatorid);

  duration validity = duration.ofseconds(properties.getpreviewexpireseconds());
  instant expiresat = instant.now().plus(validity);
  string url = ossgateway.presignget(asset.objectkey(), validity);
  return new previewurlresult(fileid, url, expiresat);
}

public record previewurlresult(long fileid, string url, instant expiresat) {}

public string generatepreviewurl(string objectkey, long requestedseconds) {
  long minseconds = 30;
  long maxseconds = 3600;
  long seconds = math.max(minseconds, math.min(requestedseconds, maxseconds));
  return ossgateway.presignget(objectkey, duration.ofseconds(seconds));
}

/** 公开对象链接：不签名、不设置过期时间。 */
public string generatepublicurl(long fileid, long operatorid) {
  fileasset asset = assetrepository.findavailablebyid(fileid)
      .orelsethrow(() -> new fileuploadexception("file_not_found", "文件不存在"));
  permissionservice.assertreadable(asset, operatorid);
  if (!visibility.public.name().equals(asset.visibility())) {
    throw new fileuploadexception("file_not_public", "该文件不是公开对象");
  }
  return ossgateway.publicurl(asset.objectkey());
}

@getmapping("/{fileid}/public-url")
public publicurlresponse publicurl(
    @pathvariable long fileid, authentication authentication) {
  string url = filepresentationservice.generatepublicurl(
      fileid, currentuserid(authentication));
  return new publicurlresponse(fileid, url);
}

public record publicurlresponse(long fileid, string url) {}

package com.example.file.service;

import com.example.file.config.ossproperties;
import java.nio.charset.standardcharsets;
import java.security.messagedigest;
import java.util.hexformat;
import java.util.uuid;
import org.springframework.stereotype.component;

/** 生成不可由用户控制目录的 oss 对象键。 */
@component
public class objectkeyfactory {
  private final ossproperties properties;

  public objectkeyfactory(ossproperties properties) {
    this.properties = properties;
  }

  public string create(string digest, string originalname) {
    string extension = extensionof(originalname);
    string shard = digest.substring(0, 2);
    return "%s/%s/%s/%s%s".formatted(
        trimslash(properties.getfolder()), shard, digest, uuid.randomuuid(), extension);
  }

  private string extensionof(string name) {
    int slash = math.max(name.lastindexof('/'), name.lastindexof('\\'));
    string base = slash >= 0 ? name.substring(slash + 1) : name;
    int dot = base.lastindexof('.');
    if (dot <= 0 || dot == base.length() - 1) {
      return "";
    }
    string extension = base.substring(dot).tolowercase();
    return extension.matches("\\.[a-z0-9]{1,10}") ? extension : "";
  }

  private string trimslash(string value) {
    return value == null ? "files" : value.replaceall("^/+|/+$", "");
  }
}

@service
public class fileassetservice {
  private final fileassetrepository assetrepository;
  private final filereferencerepository referencerepository;
  private final ossgateway ossgateway;
  private final objectkeyfactory keyfactory;

  public fileassetservice(
      fileassetrepository assetrepository,
      filereferencerepository referencerepository,
      ossgateway ossgateway,
      objectkeyfactory keyfactory) {
    this.assetrepository = assetrepository;
    this.referencerepository = referencerepository;
    this.ossgateway = ossgateway;
    this.keyfactory = keyfactory;
  }

  /**
   * 尝试秒传。命中条件是摘要、大小、权限、资产状态和 oss 对象同时满足。
   */
  @transactional
  public instantuploadresult tryinstantupload(instantuploadcommand command, long operatorid) {
    fileasset asset = assetrepository.findavailablebydigestandsize(
        command.digest(), command.sizebytes(), command.orgid()).orelse(null);
    if (asset == null || !ossgateway.exists(asset.objectkey())) {
      return instantuploadresult.miss();
    }
    referencerepository.insertifabsent(
        new filereference(asset.id(), command.biztype(), command.bizid(), operatorid));
    return instantuploadresult.hit(asset.id());
  }

  /** 普通上传完成后的登记；对象校验失败时不创建可用资产。 */
  @transactional
  public fileasset registeruploadedobject(
      uploadmetadata metadata, long operatorid) {
    if (!ossgateway.exists(metadata.objectkey())
        || ossgateway.objectsize(metadata.objectkey()) != metadata.sizebytes()) {
      throw new fileuploadexception("object_verify_failed", "oss 对象不存在或大小不匹配");
    }
    fileasset asset = assetrepository.insertavailable(
        new fileassetdraft(
            metadata.digest(), metadata.sizebytes(), metadata.originalname(),
            metadata.mimetype(), metadata.objectkey(), operatorid));
    referencerepository.insertifabsent(
        new filereference(asset.id(), metadata.biztype(), metadata.bizid(), operatorid));
    return asset;
  }
}

@restcontroller
@requestmapping("/api/files")
public class fileuploadcontroller {
  private final fileuploadapplicationservice applicationservice;

  public fileuploadcontroller(fileuploadapplicationservice applicationservice) {
    this.applicationservice = applicationservice;
  }

  @postmapping(value = "/upload", consumes = mediatype.multipart_form_data_value)
  public uploadresponse upload(
      @requestpart("file") multipartfile file,
      @requestparam string digest,
      @requestparam string biztype,
      @requestparam string bizid,
      authentication authentication) throws ioexception {
    return applicationservice.uploadsmallfile(
        new smalluploadcommand(
            file.getoriginalfilename(), file.getcontenttype(), file.getsize(),
            digest, biztype, bizid, file.getinputstream()),
        currentuserid(authentication));
  }

  private long currentuserid(authentication authentication) {
    return long.parselong(authentication.getname());
  }
}

@service
public class fileuploadapplicationservice {
  private final ossgateway ossgateway;
  private final objectkeyfactory keyfactory;
  private final fileassetservice assetservice;
  private final ossproperties properties;

  @transactional
  public uploadresponse uploadsmallfile(smalluploadcommand command, long operatorid)
      throws ioexception {
    validate(command);
    instantuploadresult instant = assetservice.tryinstantupload(
        new instantuploadcommand(command.digest(), command.sizebytes(),
            command.biztype(), command.bizid(), currentorgid()), operatorid);
    if (instant.hit()) {
      return uploadresponse.instant(instant.fileid());
    }

    string objectkey = keyfactory.create(command.digest(), command.originalname());
    try (inputstream input = command.input()) {
      ossgateway.putobject(objectkey, input, command.sizebytes(), safecontenttype(command.mimetype()));
    }
    fileasset asset = assetservice.registeruploadedobject(
        new uploadmetadata(
            command.digest(), command.sizebytes(), command.originalname(),
            safecontenttype(command.mimetype()), objectkey,
            command.biztype(), command.bizid()), operatorid);
    return uploadresponse.uploaded(asset.id());
  }

  private void validate(smalluploadcommand command) {
    if (command.sizebytes() <= 0 || command.sizebytes() > properties.getmultipart().getthresholdbytes()) {
      throw new fileuploadexception("file_size_invalid", "文件大小不符合普通上传范围");
    }
    if (!command.digest().matches("[0-9a-fa-f]{32}")) {
      throw new fileuploadexception("digest_invalid", "md5 格式错误");
    }
  }
}

@postmapping("/multipart/init")
public multipartinitresponse init(
    @requestbody multipartinitrequest request, authentication authentication) {
  return multipartservice.init(request, currentuserid(authentication));
}

@service
public class multipartservice {
  private final ossgateway ossgateway;
  private final multipartsessionrepository sessionrepository;
  private final multipartpartrepository partrepository;
  private final fileassetservice assetservice;
  private final objectkeyfactory keyfactory;
  private final ossproperties properties;

  @transactional
  public multipartinitresponse init(multipartinitrequest request, long operatorid) {
    validatemultipart(request);
    instantuploadresult instant = assetservice.tryinstantupload(
        new instantuploadcommand(request.digest(), request.sizebytes(),
            request.biztype(), request.bizid(), request.orgid()), operatorid);
    if (instant.hit()) {
      return multipartinitresponse.instant(instant.fileid());
    }

    string objectkey = keyfactory.create(request.digest(), request.originalname());
    string uploadid = ossgateway.initiatemultipart(objectkey, request.mimetype());
    string sessionid = uuid.randomuuid().tostring();
    multipartuploadsession session = new multipartuploadsession(
        sessionid, uploadid, objectkey, request.digest(), request.sizebytes(),
        request.originalname(), request.mimetype(), request.biztype(), request.bizid(),
        properties.getmultipart().getpartsizebytes(), operatorid,
        instant.now().plus(properties.getmultipart().getsessionexpireminutes(), chronounit.minutes));
    sessionrepository.insert(session);
    return multipartinitresponse.created(
        sessionid, objectkey, session.partsizebytes(), list.of());
  }

  private void validatemultipart(multipartinitrequest request) {
    long partcount = (request.sizebytes() + properties.getmultipart().getpartsizebytes() - 1)
        / properties.getmultipart().getpartsizebytes();
    if (partcount > properties.getmultipart().getmaxparts()) {
      throw new fileuploadexception("part_count_exceeded", "分片数量超过 oss 限制");
    }
  }
}

@postmapping(value = "/multipart/{sessionid}/parts/{partnumber}",
    consumes = mediatype.multipart_form_data_value)
public partuploadresponse uploadpart(
    @pathvariable string sessionid,
    @pathvariable int partnumber,
    @requestpart("part") multipartfile part,
    authentication authentication) throws ioexception {
  return multipartservice.uploadpart(
      sessionid, partnumber, part, currentuserid(authentication));
}

@transactional
public partuploadresponse uploadpart(
    string sessionid, int partnumber, multipartfile part, long operatorid) throws ioexception {
  multipartuploadsession session = sessionrepository.lockowned(sessionid, operatorid)
      .orelsethrow(() -> new fileuploadexception("session_not_found", "上传会话不存在或无权限"));
  checkpartnumberandsize(session, partnumber, part.getsize());

  try (inputstream input = part.getinputstream()) {
    partetag etag = ossgateway.uploadpart(
        session.objectkey(), session.uploadid(), partnumber, input, part.getsize());
    partrepository.upsert(new multipartpart(
        sessionid, partnumber, etag.getetag(), part.getsize(), instant.now()));
    return new partuploadresponse(partnumber, etag.getetag());
  }
}

@getmapping("/multipart/{sessionid}/parts")
public list<partuploadresponse> listparts(
    @pathvariable string sessionid, authentication authentication) {
  sessionrepository.assertowned(sessionid, currentuserid(authentication));
  return partrepository.findall(sessionid).stream()
      .sorted(comparator.comparingint(multipartpart::partnumber))
      .map(part -> new partuploadresponse(part.partnumber(), part.etag()))
      .tolist();
}

@postmapping("/multipart/{sessionid}/complete")
public uploadresponse complete(
    @pathvariable string sessionid, authentication authentication) {
  return multipartservice.complete(sessionid, currentuserid(authentication));
}

@transactional
public uploadresponse complete(string sessionid, long operatorid) {
  multipartuploadsession session = sessionrepository.lockowned(sessionid, operatorid)
      .orelsethrow(() -> new fileuploadexception("session_not_found", "上传会话不存在或无权限"));
  if (session.isexpired()) {
    throw new fileuploadexception("session_expired", "上传会话已过期");
  }
  if (session.isavailable()) {
    return uploadresponse.uploaded(session.fileid());
  }

  list<multipartpart> persistedparts = partrepository.findall(sessionid).stream()
      .sorted(comparator.comparingint(multipartpart::partnumber))
      .tolist();
  assertallpartspresent(session, persistedparts);
  list<partetag> etags = persistedparts.stream()
      .map(part -> new partetag(part.partnumber(), part.etag()))
      .tolist();
  ossgateway.completemultipart(session.objectkey(), session.uploadid(), etags);

  if (!ossgateway.exists(session.objectkey())
      || ossgateway.objectsize(session.objectkey()) != session.sizebytes()) {
    sessionrepository.markfailed(sessionid, "object_verify_failed");
    throw new fileuploadexception("object_verify_failed", "合并后对象校验失败");
  }
  fileasset asset = assetservice.registeruploadedobject(
      new uploadmetadata(
          session.digest(), session.sizebytes(), session.originalname(),
          session.mimetype(), session.objectkey(), session.biztype(), session.bizid()),
      operatorid);
  sessionrepository.markavailable(sessionid, asset.id());
  return uploadresponse.uploaded(asset.id());
}

@deletemapping("/multipart/{sessionid}")
@responsestatus(httpstatus.no_content)
public void cancel(@pathvariable string sessionid, authentication authentication) {
  multipartservice.cancel(sessionid, currentuserid(authentication));
}

@transactional
public void cancel(string sessionid, long operatorid) {
  multipartuploadsession session = sessionrepository.lockowned(sessionid, operatorid)
      .orelsethrow(() -> new fileuploadexception("session_not_found", "上传会话不存在或无权限"));
  if (!session.isavailable() && !session.iscancelled()) {
    ossgateway.abortmultipart(session.objectkey(), session.uploadid());
    sessionrepository.markcancelled(sessionid);
  }
}

@scheduled(fixeddelaystring = "${app.oss.multipart.cleanup-delay-ms:3600000}")
public void cleanupexpiredsessions() {
  for (multipartuploadsession session : sessionrepository.findexpireduploading(instant.now())) {
    try {
      ossgateway.abortmultipart(session.objectkey(), session.uploadid());
      sessionrepository.markcancelled(session.id());
    } catch (runtimeexception ex) {
      log.warn("oss multipart cleanup failed, sessionid={}", session.id(), ex);
    }
  }
}

@getmapping("/{fileid}/preview-url")
public previewurlresponse previewurl(
    @pathvariable long fileid, authentication authentication) {
  fileasset asset = assetrepository.findavailablebyid(fileid)
      .orelsethrow(() -> new fileuploadexception("file_not_found", "文件不存在"));
  permissionservice.assertreadable(asset, currentuserid(authentication));
  duration validity = duration.ofseconds(properties.getpreviewexpireseconds());
  string url = ossgateway.presignget(asset.objectkey(), validity);
  return new previewurlresponse(
      fileid, url, instant.now().plus(validity), properties.getpreviewexpireseconds());
}

@test
void instantuploadcreatesreferencewithoutuploadingagain() { /* verify putobject never called */ }

@test
void multipartresumeuploadsonlymissingparts() { /* persisted part 1 is reused */ }

@test
void completerejectsmissingpart() { /* expect part_missing */ }

@test
void completepassesmutablepartlisttoosssdk() { /* sdk sorting must not throw unsupportedoperationexception */ }

@test
void duplicatecompletereturnsexistingasset() { /* idempotent */ }

@test
void expiredpreviewurlisrejectedbyoss() { /* requires isolated real oss or contract test */ }

mvn -pl ai-image-backend -am -dskiptests=false test
git diff --check -- docs/java实现阿里云oss文件上传链路与断点续传.md