Skip to content

Y
yd-oss
Commit c4f2b8e5 by zhangxingmin
Options

push

parent 4348c14c
Showing with 91 additions and 80 deletions
yd-oss-api/src/main/java/com/yd/oss/api/service/impl/ApiChunkedUploadContextServiceImpl.java
...@@ -14,11 +14,13 @@ import com.yd.oss.service.model.ChunkedUploadContext; ...@@ -14,11 +14,13 @@ import com.yd.oss.service.model.ChunkedUploadContext;
import com.yd.oss.service.model.OssProvider; import com.yd.oss.service.model.OssProvider;
import com.yd.oss.service.service.IOssProviderService; import com.yd.oss.service.service.IOssProviderService;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock; import org.redisson.api.RLock;
import org.redisson.api.RedissonClient; import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile; import org.springframework.web.multipart.MultipartFile;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.*; import java.util.*;
...@@ -46,137 +48,135 @@ public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadConte ...@@ -46,137 +48,135 @@ public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadConte
private static final String LOCK_KEY_PREFIX = "chunked:upload:lock:"; private static final String LOCK_KEY_PREFIX = "chunked:upload:lock:";
/** /**
* 上传单个分片（使用分布式锁保证上下文唯一性） * 上传单个分片（使用分布式锁保护上下文初始化 + Map存储ETag）
* @param taskId 录制任务ID
* @param chunkIndex 分片索引（从0开始）
* @param chunk 分片文件
* @param projectBizId 项目业务ID（为空时使用默认OSS）
* @param source 来源标识（如 'recording'）
*/ */
@Override @Override
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
public void uploadChunk(String taskId, Integer chunkIndex, MultipartFile chunk, public void uploadChunk(String taskId, Integer chunkIndex, MultipartFile chunk,
String projectBizId, String source) { String projectBizId, String source) {
// ---------- 前置准备 ----------
log.info("【uploadChunk】开始处理, taskId={}, chunkIndex={}, projectBizId={}, source={}, fileSize={}", log.info("【uploadChunk】开始处理, taskId={}, chunkIndex={}, projectBizId={}, source={}, fileSize={}",
taskId, chunkIndex, projectBizId, source, chunk != null ? chunk.getSize() : 0); taskId, chunkIndex, projectBizId, source, chunk != null ? chunk.getSize() : 0);
// 获取 OSS 服务商（不依赖上下文，提前获取） // 获取OSS服务商
OssProvider provider = ossProviderService.getProviderByProjectId(projectBizId); OssProvider provider = ossProviderService.getProviderByProjectId(projectBizId);
if (provider == null) { if (provider == null) {
log.error("【uploadChunk】未找到OSS服务商, projectBizId={}", projectBizId); log.error("【uploadChunk】未找到OSS服务商, projectBizId={}", projectBizId);
throw new BusinessException("未找到对应的OSS服务商配置"); throw new BusinessException("未找到对应的OSS服务商配置");
} }
log.info("【uploadChunk】获取到OSS服务商: name={}, bucket={}", provider.getName(), provider.getBucketName());
// 创建 OSS 客户端（提前创建，供后续使用）
OSS ossClient = ossClientFactory.createOssClient(provider); OSS ossClient = ossClientFactory.createOssClient(provider);
log.info("【uploadChunk】OSS客户端创建成功");
// ---------- 分布式锁（保护整个方法，确保同一taskId操作串行） ---------- // 获取或创建上下文（仅初始化时加锁）
ChunkedUploadContext context = getOrCreateContext(taskId, provider, source, ossClient);
// 上传分片并更新ETag（使用Map存储，避免PartETag反序列化问题）
try {
int partNumber = chunkIndex + 1;
log.info("【uploadChunk】开始上传分片, taskId={}, partNumber={}, chunkSize={}",
taskId, partNumber, chunk.getSize());
UploadPartRequest uploadPartRequest = new UploadPartRequest();
uploadPartRequest.setBucketName(provider.getBucketName());
uploadPartRequest.setKey(context.getObjectKey());
uploadPartRequest.setUploadId(context.getUploadId());
uploadPartRequest.setPartNumber(partNumber);
uploadPartRequest.setInputStream(chunk.getInputStream());
uploadPartRequest.setPartSize(chunk.getSize());
UploadPartResult uploadResult = ossClient.uploadPart(uploadPartRequest);
// 读取现有ETag列表（存储为List<Map>）
String json = context.getPartEtagsJson();
List<Map<String, Object>> partInfoList;
if (StringUtils.isBlank(json) || "[]".equals(json)) {
partInfoList = new ArrayList<>();
} else {
partInfoList = objectMapper.readValue(json, new TypeReference<List<Map<String, Object>>>() {});
}
// 添加新分片信息
Map<String, Object> newPart = new HashMap<>();
newPart.put("partNumber", partNumber);
newPart.put("etag", uploadResult.getETag());
partInfoList.add(newPart);
// 按partNumber排序
partInfoList.sort(Comparator.comparingInt(m -> (Integer) m.get("partNumber")));
// 序列化并更新数据库
context.setPartEtagsJson(objectMapper.writeValueAsString(partInfoList));
context.setUpdateTime(LocalDateTime.now());
contextMapper.updateById(context);
log.info("【uploadChunk】分片上传成功, taskId={}, partNumber={}, ETag={}, 当前总分片数={}",
taskId, partNumber, uploadResult.getETag(), partInfoList.size());
} catch (Exception e) {
log.error("【uploadChunk】分片上传或更新ETag失败, taskId={}, chunkIndex={}", taskId, chunkIndex, e);
throw new BusinessException("分片上传失败: " + e.getMessage());
}
}
/**
* 获取或创建上传上下文（使用分布式锁保证唯一性）
*/
private ChunkedUploadContext getOrCreateContext(String taskId, OssProvider provider,
String source, OSS ossClient) {
String lockKey = LOCK_KEY_PREFIX + taskId; String lockKey = LOCK_KEY_PREFIX + taskId;
RLock lock = redissonClient.getLock(lockKey); RLock lock = redissonClient.getLock(lockKey);
boolean locked = false; boolean locked = false;
try { try {
// 尝试获取锁，等待5秒，锁持有时间30秒（足够处理上传+DB更新） locked = lock.tryLock(5, 10, TimeUnit.SECONDS);
locked = lock.tryLock(5, 30, TimeUnit.SECONDS);
if (!locked) { if (!locked) {
log.error("【uploadChunk】获取分布式锁失败, taskId={}", taskId);
throw new BusinessException("系统繁忙，请稍后重试"); throw new BusinessException("系统繁忙，请稍后重试");
} }
log.debug("【uploadChunk】获取锁成功, taskId={}", taskId);
// ---------- 1. 查询或创建上传上下文（原子操作） ----------
ChunkedUploadContext context = contextMapper.selectOne( ChunkedUploadContext context = contextMapper.selectOne(
new LambdaQueryWrapper<ChunkedUploadContext>() new LambdaQueryWrapper<ChunkedUploadContext>()
.eq(ChunkedUploadContext::getTaskId, taskId) .eq(ChunkedUploadContext::getTaskId, taskId)
.last(" limit 1 ")
); );
if (context == null) { if (context == null) {
log.info("【uploadChunk】上下文不存在，将初始化新的分片上传, taskId={}", taskId); log.info("【getOrCreateContext】上下文不存在，初始化新上传, taskId={}", taskId);
context = new ChunkedUploadContext(); context = new ChunkedUploadContext();
context.setTaskId(taskId); context.setTaskId(taskId);
// 生成 OSS 对象 Key
String objectKey = String.format("sharding/" + source + "/%tY/%tm/%s_%d.webm", String objectKey = String.format("sharding/" + source + "/%tY/%tm/%s_%d.webm",
new Date(), new Date(), taskId, System.currentTimeMillis()); new Date(), new Date(), taskId, System.currentTimeMillis());
context.setObjectKey(objectKey); context.setObjectKey(objectKey);
log.info("【uploadChunk】生成 objectKey: {}", objectKey);
// 调用 OSS 初始化分片上传
InitiateMultipartUploadRequest initRequest = InitiateMultipartUploadRequest initRequest =
new InitiateMultipartUploadRequest(provider.getBucketName(), objectKey); new InitiateMultipartUploadRequest(provider.getBucketName(), objectKey);
InitiateMultipartUploadResult initResult = ossClient.initiateMultipartUpload(initRequest); InitiateMultipartUploadResult initResult = ossClient.initiateMultipartUpload(initRequest);
context.setUploadId(initResult.getUploadId()); context.setUploadId(initResult.getUploadId());
context.setPartEtagsJson("[]"); context.setPartEtagsJson("[]"); // 初始为空数组
context.setStatus(1); // 上传中 context.setStatus(1); // 上传中
// 插入数据库（此时其他线程被锁阻挡，不会重复插入）
int insertResult = contextMapper.insert(context); int insertResult = contextMapper.insert(context);
log.info("【uploadChunk】初始化分片上传并入库, taskId={}, uploadId={}, objectKey={}, insertResult={}", log.info("【getOrCreateContext】初始化分片上传并入库, taskId={}, uploadId={}, objectKey={}, insertResult={}",
taskId, context.getUploadId(), objectKey, insertResult); taskId, context.getUploadId(), objectKey, insertResult);
} else { } else {
log.info("【uploadChunk】找到已存在的上下文, taskId={}, uploadId={}, status={}, partCount={}", log.info("【getOrCreateContext】找到已存在的上下文, taskId={}, uploadId={}, status={}",
taskId, context.getUploadId(), context.getStatus(), taskId, context.getUploadId(), context.getStatus());
context.getPartEtagsJson() != null ? context.getPartEtagsJson().length() : 0);
} }
return context;
// ---------- 2. 上传当前分片 ----------
int partNumber = chunkIndex + 1;
log.info("【uploadChunk】开始上传分片, taskId={}, partNumber={}, chunkSize={}",
taskId, partNumber, chunk.getSize());
UploadPartRequest uploadPartRequest = new UploadPartRequest();
uploadPartRequest.setBucketName(provider.getBucketName());
uploadPartRequest.setKey(context.getObjectKey());
uploadPartRequest.setUploadId(context.getUploadId());
uploadPartRequest.setPartNumber(partNumber);
uploadPartRequest.setInputStream(chunk.getInputStream());
uploadPartRequest.setPartSize(chunk.getSize());
UploadPartResult uploadResult = ossClient.uploadPart(uploadPartRequest);
// ---------- 3. 更新 ETag 列表（在锁保护下，不会并发覆盖） ----------
PartETag partETag = new PartETag(uploadResult.getPartNumber(), uploadResult.getETag());
String json = context.getPartEtagsJson();
List<PartETag> partETags = objectMapper.readValue(json, new TypeReference<List<PartETag>>() {});
partETags.add(partETag);
partETags.sort(Comparator.comparingInt(PartETag::getPartNumber));
context.setPartEtagsJson(objectMapper.writeValueAsString(partETags));
context.setUpdateTime(LocalDateTime.now());
contextMapper.updateById(context);
log.info("【uploadChunk】分片上传成功, taskId={}, partNumber={}, ETag={}, 当前总分片数={}",
taskId, partNumber, uploadResult.getETag(), partETags.size());
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
log.error("【uploadChunk】获取锁被中断, taskId={}", taskId, e); log.error("【getOrCreateContext】获取锁被中断, taskId={}", taskId, e);
throw new BusinessException("系统中断，请重试"); throw new BusinessException("系统中断，请重试");
} catch (Exception e) {
log.error("【uploadChunk】处理失败, taskId={}, chunkIndex={}", taskId, chunkIndex, e);
throw new BusinessException("分片上传失败: " + e.getMessage());
} finally { } finally {
// 释放锁（仅当当前线程持有锁时）
if (locked && lock.isHeldByCurrentThread()) { if (locked && lock.isHeldByCurrentThread()) {
lock.unlock(); lock.unlock();
log.debug("【uploadChunk】释放分布式锁, taskId={}", taskId); log.debug("【getOrCreateContext】释放锁, taskId={}", taskId);
} }
} }
} }
/** /**
* 完成分片上传（合并） * 完成分片上传（合并）
* @param taskId
* @param projectBizId
* @return
*/ */
@Override @Override
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
public Result<Map<String, Object>> finishChunks(String taskId, String projectBizId) { public Result<Map<String, Object>> finishChunks(String taskId, String projectBizId) {
// ... 原有逻辑不变（无需加锁，因为合并时上下文已存在且不再变化）
// 但为了安全，也可以加锁确保合并时上下文不会被更新，但此时上传已完成，不需要锁。
log.info("【finishChunks】开始完成分片上传, taskId={}, projectBizId={}", taskId, projectBizId); log.info("【finishChunks】开始完成分片上传, taskId={}, projectBizId={}", taskId, projectBizId);
// 1. 查询上下文 // 1. 查询上下文
...@@ -191,29 +191,40 @@ public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadConte ...@@ -191,29 +191,40 @@ public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadConte
log.info("【finishChunks】找到上下文, taskId={}, uploadId={}, status={}, objectKey={}", log.info("【finishChunks】找到上下文, taskId={}, uploadId={}, status={}, objectKey={}",
taskId, context.getUploadId(), context.getStatus(), context.getObjectKey()); taskId, context.getUploadId(), context.getStatus(), context.getObjectKey());
// 2. 解析分片列表 // 2. 解析分片列表（从Map列表转为PartETag列表）
String json = context.getPartEtagsJson(); String json = context.getPartEtagsJson();
List<PartETag> partETags; List<Map<String, Object>> partInfoList;
try { try {
partETags = objectMapper.readValue(json, new TypeReference<List<PartETag>>() {}); partInfoList = objectMapper.readValue(json, new TypeReference<List<Map<String, Object>>>() {});
} catch (Exception e) { } catch (Exception e) {
log.error("【finishChunks】解析分片 ETag 列表失败, taskId={}, json={}", taskId, json, e); log.error("【finishChunks】解析分片数据失败, taskId={}, json={}", taskId, json, e);
throw new BusinessException("数据异常，无法合并"); throw new BusinessException("数据异常，无法合并");
} }
if (partETags.isEmpty()) { if (partInfoList.isEmpty()) {
log.error("【finishChunks】没有分片数据, taskId={}", taskId); log.error("【finishChunks】没有分片数据, taskId={}", taskId);
throw new BusinessException("没有分片数据，无法合并"); throw new BusinessException("没有分片数据，无法合并");
} }
List<PartETag> partETags = new ArrayList<>();
for (Map<String, Object> map : partInfoList) {
Integer partNumber = (Integer) map.get("partNumber");
String etag = (String) map.get("etag");
if (partNumber == null || etag == null) {
log.warn("【finishChunks】分片数据不完整，跳过: {}", map);
continue;
}
partETags.add(new PartETag(partNumber, etag));
}
if (partETags.isEmpty()) {
throw new BusinessException("没有有效的分片数据，无法合并");
}
log.info("【finishChunks】解析到 {} 个分片, taskId={}", partETags.size(), taskId); log.info("【finishChunks】解析到 {} 个分片, taskId={}", partETags.size(), taskId);
// 3. 获取 OSS 服务商 // 3. 获取OSS服务商
OssProvider provider = ossProviderService.getProviderByProjectId(projectBizId); OssProvider provider = ossProviderService.getProviderByProjectId(projectBizId);
if (provider == null) { if (provider == null) {
log.error("【finishChunks】未找到OSS服务商, projectBizId={}", projectBizId); log.error("【finishChunks】未找到OSS服务商, projectBizId={}", projectBizId);
throw new BusinessException("未找到对应的OSS服务商配置"); throw new BusinessException("未找到对应的OSS服务商配置");
} }
log.info("【finishChunks】获取到OSS服务商: name={}, bucket={}", provider.getName(), provider.getBucketName());
OSS ossClient = ossClientFactory.createOssClient(provider); OSS ossClient = ossClientFactory.createOssClient(provider);
// 4. 执行合并 // 4. 执行合并
...@@ -226,7 +237,7 @@ public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadConte ...@@ -226,7 +237,7 @@ public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadConte
CompleteMultipartUploadResult completeResult = ossClient.completeMultipartUpload(completeRequest); CompleteMultipartUploadResult completeResult = ossClient.completeMultipartUpload(completeRequest);
log.info("【finishChunks】合并成功, taskId={}, location={}", taskId, completeResult.getLocation()); log.info("【finishChunks】合并成功, taskId={}, location={}", taskId, completeResult.getLocation());
// 5. 构建文件 URL 和大小 // 5. 构建文件URL和大小
String fileUrl = String.format("https://%s.%s/%s", String fileUrl = String.format("https://%s.%s/%s",
provider.getBucketName(), provider.getBucketName(),
provider.getEndpoint().replace("https://", ""), provider.getEndpoint().replace("https://", ""),
...@@ -235,8 +246,8 @@ public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadConte ...@@ -235,8 +246,8 @@ public class ApiChunkedUploadContextServiceImpl implements ApiChunkedUploadConte
ObjectMetadata metadata = ossClient.getObjectMetadata(provider.getBucketName(), context.getObjectKey()); ObjectMetadata metadata = ossClient.getObjectMetadata(provider.getBucketName(), context.getObjectKey());
long fileSize = metadata.getContentLength(); long fileSize = metadata.getContentLength();
// 6. 更新上下文状态 // 6. 更新上下文状态为已完成
context.setStatus(2); // 已完成 context.setStatus(2);
contextMapper.updateById(context); contextMapper.updateById(context);
log.info("【finishChunks】上下文状态更新为已完成, taskId={}", taskId); log.info("【finishChunks】上下文状态更新为已完成, taskId={}", taskId);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment