OSS文件分片上传
依赖
<!-- https://mvnrepository.com/artifact/com.aliyun.oss/aliyun-sdk-oss -->
<dependency>
<groupId>com.aliyun.oss</groupId>
<artifactId>aliyun-sdk-oss</artifactId>
<version>3.8.1</version>
</dependency>
基础参数dto
/**
* @author WJL
*/
@Data
@Builder
public class OssParamDTO {
private String endpoint;
private String accessKeyId;
private String accessKeySecret;
private String bucketName;
private String folder;
/**
* objectName = folder + fileName
*/
private String objectName;
/**
* 上传线程
*/
private Integer task;
/**
* 每个线程处理大小 分片大小
*/
private Integer number;
}
具体上传方法
小文件上传
public static PutObjectResult uploadFile(OssParamDTO ossParamDTO, InputStream inputStream){
// 创建OSSClient实例。
OSS ossClient = new OSSClientBuilder().build(ossParamDTO.getEndpoint(), ossParamDTO.getAccessKeyId(), ossParamDTO.getAccessKeySecret());
PutObjectResult putObjectResult = null;
// 上传文件流。
try {
putObjectResult = ossClient.putObject(ossParamDTO.getBucketName(), ossParamDTO.getObjectName(), inputStream);
//权限设置
ossClient.setBucketAcl(ossParamDTO.getBucketName(), CannedAccessControlList.PublicRead);
} catch (Exception e) {
e.printStackTrace();
}finally {
// 关闭OSSClient。
ossClient.shutdown();
}
return putObjectResult;
}
大文件上传，分片oss自己处理
处理逻辑：前段轮训查询数据库某个字段，当该字段被回调接口更新时结束轮训，上传完成
public static void uploadBigFile(OssParamDTO ossParamDTO,String path, File file,Long fileId) throws Throwable {
System.out.println("上传时间："+System.currentTimeMillis());
// Endpoint以杭州为例，其它Region请按实际情况填写。
String endpoint = ossParamDTO.getEndpoint();
// 阿里云主账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM账号进行API访问或日常运维，请登录 https://ram.console.aliyun.com 创建RAM账号。
String accessKeyId = ossParamDTO.getAccessKeyId();
String accessKeySecret = ossParamDTO.getAccessKeySecret();
// 创建OSSClient实例。
OSS ossClient = new OSSClientBuilder().build(endpoint, accessKeyId, accessKeySecret);
ObjectMetadata meta = new ObjectMetadata();
// 指定上传的内容类型。
meta.setContentType("text/plain");
// 通过UploadFileRequest设置多个参数。
UploadFileRequest uploadFileRequest = new UploadFileRequest(ossParamDTO.getBucketName(),ossParamDTO.getObjectName());
// 指定上传的本地文件。
uploadFileRequest.setUploadFile(path);
// 指定上传并发线程数，默认为1。
uploadFileRequest.setTaskNum(ossParamDTO.getTask());
// 指定上传的分片大小，范围为100KB~5GB，默认为文件大小/10000。
uploadFileRequest.setPartSize(ossParamDTO.getNumber() * 1024 * 1024);
uploadFileRequest.setObjectMetadata(meta);
// 设置上传成功回调，参数为Callback类型。
Callback callback = new Callback();
callback.setCalbackBodyType(Callback.CalbackBodyType.URL);
//回调参数 --- 同同步到数据库
callback.setCallbackBody("fileId="+fileId+"&fileName=${object}&uploadStatus=1");
//回调接口（自己服务器接口，可供外网访问）
callback.setCallbackUrl("http://3m8wv2.natappfree.cc/web/common/callBack");
uploadFileRequest.setCallback(callback);
// 断点续传上传。
ossClient.uploadFile(uploadFileRequest);
//权限设置
ossClient.setBucketAcl(ossParamDTO.getBucketName(), CannedAccessControlList.PublicRead);
// 关闭OSSClient。
ossClient.shutdown();
}
大文件本地分片，多线程执行分片上传，再合并碎片
分片上传代码
PartETag getUploadPartETag(String objectName, String bucketName, String uploadId,
InputStream instream, Long curPartSize,Integer partNum,
OSS ossClient, CountDownLatch countDownLatch){
long before = System.currentTimeMillis();
UploadPartRequest uploadPartRequest = null;
try {
log.debug("分片文件上传线程： {}",Thread.currentThread().getName());
uploadPartRequest = new UploadPartRequest();
uploadPartRequest.setBucketName(bucketName);
uploadPartRequest.setKey(objectName);
uploadPartRequest.setUploadId(uploadId);
uploadPartRequest.setInputStream(instream);
// 设置分片大小。除了最后一个分片没有大小限制，其他的分片最小为100KB。
uploadPartRequest.setPartSize(curPartSize);
// 设置分片号。每一个上传的分片都有一个分片号，取值范围是1~10000，如果超出这个范围，OSS将返回InvalidArgument的错误码。
uploadPartRequest.setPartNumber(partNum);
// 每个分片不需要按顺序上传，甚至可以在不同客户端上传，OSS会按照分片号排序组成完整的文件。
UploadPartResult uploadPartResult = ossClient.uploadPart(uploadPartRequest);
// 每次上传分片之后，OSS的返回结果会包含一个PartETag。PartETag将被保存到partETags中。
log.debug("getPartETag ::{}" ,uploadPartResult.getPartETag().getETag());
return uploadPartResult.getPartETag();
}finally {
countDownLatch.countDown();
log.debug("线程： {} 执行完毕, 等待线程数 ：{}, 消耗时间： {}",
Thread.currentThread().getName(),countDownLatch.getCount(),
((System.currentTimeMillis()-before)/1000)+"s");
}
}
外部分片代码
@Qualifier("taskExecutor")
@Autowired
ThreadPoolTaskExecutor taskExecutor;
/**
* 上传
* @param ossParamDTO
* @param multipartFile
* @return
*/
public CompleteMultipartUploadResult uploadBigFileForProd(OssParamDTO ossParamDTO, MultipartFile multipartFile){
Long before = System.currentTimeMillis();
// Endpoint以杭州为例，其它Region请按实际情况填写。
String endpoint = ossParamDTO.getEndpoint();
// 阿里云主账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM账号进行API访问或日常运维，请登录 https://ram.console.aliyun.com 创建RAM账号。
String accessKeyId = ossParamDTO.getAccessKeyId();
String accessKeySecret = ossParamDTO.getAccessKeySecret();
String bucketName = ossParamDTO.getBucketName();
// <yourObjectName>表示上传文件到OSS时需要指定包含文件后缀在内的完整路径，例如abc/efg/123.jpg。
String objectName = ossParamDTO.getObjectName();
// 创建OSSClient实例。
OSS ossClient = new OSSClientBuilder().build(endpoint, accessKeyId, accessKeySecret);
// 创建InitiateMultipartUploadRequest对象。
InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucketName, objectName);
// 初始化分片。
InitiateMultipartUploadResult upresult = ossClient.initiateMultipartUpload(request);
// 返回uploadId，它是分片上传事件的唯一标识，您可以根据这个ID来发起相关的操作，如取消分片上传、查询分片上传等。
String uploadId = upresult.getUploadId();
// partETags是PartETag的集合。PartETag由分片的ETag和分片号组成。
List<PartETag> partETags = new ArrayList<>();
// 计算文件有多少个分片 15MB
final long partSize = 2 * 1024 * 1024L;
long fileLength = multipartFile.getSize();
int partCount = (int) (fileLength / partSize);
if (fileLength % partSize != 0) {
partCount++;
}
// 遍历分片上传。
log.info("分片数量 {}",partCount);
List<Future<PartETag>> futureList = Collections.synchronizedList(new ArrayList());
CountDownLatch countDownLatch = new CountDownLatch(partCount);
for (int i = 0; i < partCount; i++) {
long startPos = i * partSize;
long curPartSize = (i + 1 == partCount) ? (fileLength - startPos) : partSize;
InputStream instream = null;
try {
instream = multipartFile.getInputStream();
} catch (IOException e) {
e.printStackTrace();
}
// 跳过已经上传的分片。
try {
instream.skip(startPos);
} catch (IOException e) {
e.printStackTrace();
}
int finalI = i;
InputStream finalInstream = instream;
Future<PartETag> partETagFuture = taskExecutor.submit(() ->
fileServiceExtAsync.getUploadPartETag(objectName, bucketName, uploadId, finalInstream, curPartSize, finalI + 1, ossClient, countDownLatch));
futureList.add(partETagFuture);
}
try {
countDownLatch.await();
for (Future<PartETag> tagFuture : futureList) {
partETags.add(tagFuture.get());
}
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
// 创建CompleteMultipartUploadRequest对象。
List<PartETag> collect = partETags.stream().sorted(Comparator.comparing(PartETag::getPartNumber)).collect(Collectors.toList());
// 在执行完成分片上传操作时，需要提供所有有效的partETags。OSS收到提交的partETags后，会逐一验证每个分片的有效性。当所有的数据分片验证通过后，OSS将把这些分片组合成一个完整的文件。
log.debug("文件开始合并");
CompleteMultipartUploadRequest completeMultipartUploadRequest =
new CompleteMultipartUploadRequest(bucketName, objectName, uploadId, collect);
// 如果需要在完成文件上传的同时设置文件访问权限，请参考以下示例代码。
completeMultipartUploadRequest.setObjectACL(CannedAccessControlList.PublicRead);
// 完成上传。
CompleteMultipartUploadResult completeMultipartUploadResult = ossClient.completeMultipartUpload(completeMultipartUploadRequest);
// 关闭OSSClient。
ossClient.shutdown();
log.debug("消耗总时间： {}",((System.currentTimeMillis()-before)/1000)+"s");
return completeMultipartUploadResult;
}
网友评论