派生自 projectDept/qhighschool
src/main/java/com/qxueyou/scc/teach/res/service/impl/HweiYunOBSServiceImpl.java
@@ -7,10 +7,12 @@ import com.obs.services.model.*; import com.qxueyou.scc.base.model.FileMeta; import com.qxueyou.scc.base.model.Result; import com.qxueyou.scc.base.util.CollectionUtils; import com.qxueyou.scc.base.util.QFileUtils; import com.qxueyou.scc.base.util.UUIDUtils; import com.qxueyou.scc.base.service.ICacheService; import com.qxueyou.scc.base.service.impl.CommonAppService; import com.qxueyou.scc.base.util.*; import com.qxueyou.scc.config.HweiOBSConfig; import com.qxueyou.scc.teach.res.model.Res; import com.qxueyou.scc.teach.res.model.ResDir; import com.qxueyou.scc.teach.res.model.ResFile; import com.qxueyou.scc.teach.res.service.HweiYunOBSService; import com.qxueyou.scc.teach.res.service.IFileService; @@ -21,6 +23,7 @@ import org.apache.logging.log4j.Logger; import org.aspectj.util.FileUtil; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.HashOperations; import org.springframework.stereotype.Service; import org.springframework.web.multipart.MultipartFile; @@ -42,7 +45,7 @@ */ @Slf4j @Service public class HweiYunOBSServiceImpl implements HweiYunOBSService { public class HweiYunOBSServiceImpl extends CommonAppService implements HweiYunOBSService { private final Logger log = LogManager.getLogger(HweiYunOBSServiceImpl.class); @Autowired @@ -50,6 +53,13 @@ @Autowired FileService fileService; /** * redis 模板 */ @Autowired ICacheService cache; public final static String FILE_TYPE_VIDEO = "video"; @@ -315,4 +325,212 @@ } return null; } /** * 文件分片上传实现逻辑，各分片是多线程上传 * * @param input 输入流 * @param uniqueId 资源唯一id(前端传入) * @param chunkNumber 分片编号 * @param chunkSize 分片大小 * @param totalChunk 分片总数 * @param fileName 文件名 * @return */ public Result uploadChunk(InputStream input, String uniqueId, int chunkNumber, long chunkSize, int totalChunk, String fileName) { HashOperations<Object, Object, Object> _cacheMap = cache.template().opsForHash(); ObsClient obsClient = new ObsClient(hweiOBSConfig.getAccessKey(),hweiOBSConfig.getSecurityKey(),hweiOBSConfig.getEndPoint()); //给初始化预留1.2秒钟的时间 try { if (chunkNumber < 4) { Thread.sleep(1200l); } } catch (InterruptedException e) { log.error(e.getMessage(), e); } //获取分片上传id(阿里云OSS初始化分片上传任务返回,和uniqueId及key一一对应),若获取不到，则10秒内循环十次直至获取到，否则返回错误 String uploadId = getUploadId(uniqueId); if (uploadId == null) { return new Result(false); } String key = (String) _cacheMap.get(uniqueId + ClientUtils.getUserId(), "key"); boolean exists = true; //上传片段 if (!_cacheMap.hasKey(uploadId, String.valueOf(chunkNumber))) { //开始上传 UploadPartRequest uploadPartRequest = new UploadPartRequest(hweiOBSConfig.getBucketName(), key); uploadPartRequest.setUploadId(uploadId); // 设置分段号，范围是1~10000 uploadPartRequest.setPartNumber(chunkNumber); // 设置分段大小 uploadPartRequest.setPartSize(chunkSize); uploadPartRequest.setInput(input); UploadPartResult partResult = obsClient.uploadPart(uploadPartRequest); exists = false; _cacheMap.put(uploadId, String.valueOf(chunkNumber), partResult.getEtag()); } //全部片段上传完成,调用client complete方法完成分片上传任务，并插入数据库 if (_cacheMap.size(uploadId) != null && _cacheMap.size(uploadId) >= totalChunk && !_cacheMap.hasKey(uniqueId + ClientUtils.getUserId(), "finish")) { List<PartEtag> tags = new ArrayList<>(5); _cacheMap.entries(uploadId).forEach((k, v) -> { tags.add(new PartEtag((String) v, Integer.parseInt((String) k))); }); CompleteMultipartUploadResult completeMultipartUploadResult = obsClient.completeMultipartUpload(new CompleteMultipartUploadRequest(hweiOBSConfig.getBucketName(), key, uploadId, tags)); System.out.println(completeMultipartUploadResult.getObjectUrl()); ObjectMetadata metadata = obsClient.getObjectMetadata(hweiOBSConfig.getBucketName(), key); // obsClient.setObjectAcl(hweiOBSConfig.getBucketName(), key, AccessControlList.REST_CANNED_PUBLIC_READ); //标记完成 _cacheMap.put(uniqueId + ClientUtils.getUserId(), "finish", "true"); // //更新数据库信息 updateFile((String) _cacheMap.get(uniqueId + ClientUtils.getUserId(), "fileId"), metadata.getContentLength(), metadata.getContentMd5(), key, fileName); //销毁缓存 cache.template().expire(uniqueId + ClientUtils.getUserId(), 1, TimeUnit.MINUTES); } return new Result(true, "hit", exists, "fileId", _cacheMap.get(uniqueId + ClientUtils.getUserId(), "fileId"), "path", _cacheMap.get(uniqueId + ClientUtils.getUserId(), "path")); } @Override public Result initUploadChunk(String uniqueId, String fileName, String md5) { String path = getDestPath(fileName); String key = chopPath(path); //清缓存 cache.template().delete(uniqueId + ClientUtils.getUserId()); InitiateMultipartUploadRequest initiateMultipartUploadRequest = new InitiateMultipartUploadRequest(hweiOBSConfig.getBucketName(), key); ObjectMetadata objectMetadata = new ObjectMetadata(); objectMetadata.setObjectStorageClass(StorageClassEnum.STANDARD); initiateMultipartUploadRequest.setMetadata(new ObjectMetadata()); //调用OSS SDK 接口返回uploadId ObsClient obsClient = new ObsClient(hweiOBSConfig.getAccessKey(),hweiOBSConfig.getSecurityKey(),hweiOBSConfig.getEndPoint()); String uploadId = obsClient.initiateMultipartUpload(initiateMultipartUploadRequest).getUploadId(); //为配合分片上传，后台先新建文件记录 ResFile file = newFileToDB(fileName, path, getFileType(fileName), 0l, md5); //添加到缓存 cache.template().opsForHash().put(uniqueId + ClientUtils.getUserId(), "uploadId", uploadId); cache.template().opsForHash().put(uniqueId + ClientUtils.getUserId(), "key", key); cache.template().opsForHash().put(uniqueId + ClientUtils.getUserId(), "path", path); cache.template().opsForHash().put(uniqueId + ClientUtils.getUserId(), "fileId", file.getFileId()); //防止产生过多缓存垃圾 cache.template().expire(uniqueId + ClientUtils.getUserId(), 1, TimeUnit.DAYS); ResFile file1 = new ResFile(); file1.setUpdateId(uploadId); file1.setFileId(file.getFileId()); file1.setPath(file.getPath()); return new Result(true,"cs",file1); } /** * 更新文件信息到数据库 * * @param fileId * @param fileLength * @param md5 * @param path * @param fileName * @return */ private ResFile updateFile(String fileId, long fileLength, String md5, String path, String fileName) { ResFile file = new ResFile(); file.setFileId(fileId); file.setSize(fileLength); file.setMd5Hash(md5); file.setPath(path); file.setFileName(fileName); cache.template().opsForList().rightPush("BaseOssServiceUpdate", file); return file; } /** * 插入文件信息到数据库 * * @param name * @param relativePath 相对路径 * @param type 文件类型，非文件格式 * @return */ private ResFile newFileToDB(String name, String relativePath, String type, Long fileSize, String md5) { ResFile file = new ResFile(); TraceUtils.setCreateTrace(file); file.setFileName(name); file.setFileFormat(QFileUtils.getFileFormat(name)); file.setFileType(type); file.setMd5Hash(md5); file.setPath(relativePath); file.setSize(fileSize == null ? 0 : fileSize); file.setDeleteFlag(false); save(file); return file; } /** * 如果路径以 / 或 \ 开头，需要截取 * * @param destPath * @return */ public String chopPath(String destPath) { if (destPath.startsWith("/") || destPath.startsWith("\\")) { return destPath.substring(1); } return destPath; } /** * 获取分片上传id(阿里云OSS初始化分片上传任务返回,和uniqueId及key一一对应),若获取不到，则10秒内循环十次直至获取到，否则返回错误 * * @param uniqueId * @return */ private String getUploadId(String uniqueId) { HashOperations<Object, Object, Object> _cacheMap = cache.template().opsForHash(); String uploadId = (String) _cacheMap.get(uniqueId + ClientUtils.getUserId(), "uploadId"); int tryCount = 0; while (uploadId == null && tryCount < 10) { try { Thread.sleep(1000l); uploadId = (String) _cacheMap.get(uniqueId + ClientUtils.getUserId(), "uploadId"); tryCount++; } catch (InterruptedException e) { log.error(e, e); } } return uploadId; } }
