基于Java实现的分布式存储系统.zip

package com.ss.services.impl; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.ss.DO.CompleteChunkFileDO; import com.ss.DO.FileChunkDO; import com.ss.DO.FileChunkMeta; import com.ss.DO.MetaFile; import com.ss.DTO.FileMeta; import com.ss.VO.BucketVO; import com.ss.VO.FileChunkMetaVO; import com.ss.VO.MetaFileVO; import com.ss.config.ClientConfig; import com.ss.services.FileService; import com.ss.error.BusinessException; import com.ss.errors.EnumClientException; import com.ss.response.CommonResponse; import com.ss.utils.ChunkAddressStrategy; import com.ss.utils.ChunkDownloaderStrategy; import com.ss.utils.Md5Util; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.web.client.RestTemplate; import org.springframework.web.multipart.MultipartFile; import java.io.IOException; import java.io.InputStream; import java.text.MessageFormat; import java.util.*; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; @Service @Slf4j public class FileServiceImpl implements FileService { private final RestTemplate restTemplate; private final ClientConfig clientConfig; private final ObjectMapper objectMapper; private final ChunkAddressStrategy chunkAddressStrategy; private final ChunkDownloaderStrategy chunkDownloaderStrategy; public FileServiceImpl(RestTemplate restTemplate, ClientConfig clientConfig, ObjectMapper objectMapper, ChunkAddressStrategy chunkAddressStrategy, ChunkDownloaderStrategy chunkDownloaderStrategy) { this.restTemplate = restTemplate; this.clientConfig = clientConfig; this.objectMapper = objectMapper; this.chunkAddressStrategy = chunkAddressStrategy; this.chunkDownloaderStrategy = chunkDownloaderStrategy; } /** * 上传文件，进行切片，并发执行切片任务 * 调用chunk-server的write方法写入切片 * 分片上传完成时调用meta保存分片完成状态，并存放在MongoDB中 * * @param bucketName * @param file * @return */ @Override public String upload(String bucketName, MultipartFile file) { String originalFilename = file.getOriginalFilename(); String extension = ""; if (Objects.nonNull(originalFilename)) { int dotIndex = originalFilename.lastIndexOf("."); if (dotIndex != -1) { extension = originalFilename.substring(dotIndex); } } FileMeta fileMeta = new FileMeta() .setBucketName(bucketName) .setExtension(extension) .setFilesize(file.getSize()); Object response = restTemplate.postForObject(clientConfig.getMetaServerAddress() + "/meta/generate", fileMeta, Object.class); CommonResponse<MetaFile> commonResponse = objectMapper .convertValue(response, new TypeReference<CommonResponse<MetaFile>>() { }); if (Objects.isNull(commonResponse)) { throw new BusinessException(EnumClientException.FAILED_TO_GET_META_FILE); } MetaFile metaFile = commonResponse.getData(); if (Objects.isNull(metaFile)) { throw new BusinessException("meta file 为空", EnumClientException.FAILED_TO_GET_META_FILE); } try { uploadChunks(file, metaFile); } catch (IOException e) { throw new BusinessException(EnumClientException.FAILED_TO_UPLOAD_CHUNK_FILE); } catch (ExecutionException e) { throw new RuntimeException(e); } catch (InterruptedException e) { throw new RuntimeException(e); } return "%s/%s.%s".formatted(bucketName, metaFile.getFilename(), metaFile.getExtension()); } /** * 上传文件，进行切片，并发执行切片任务 * 调用chunk-server的write方法写入切片 * 分片上传完成时调用meta保存分片完成状态，并存放在MongoDB中 * * @param file * @param metaFile * @throws IOException * @throws ExecutionException * @throws InterruptedException */ private void uploadChunks(MultipartFile file, MetaFile metaFile) throws IOException, ExecutionException, InterruptedException { List<FileChunkMeta> chunks = metaFile.getChunks(); chunks = chunks.stream() .sorted(Comparator.comparing(FileChunkMeta::getChunkNo)) .collect(Collectors.toList()); if (chunks.size() == 0) { return; } InputStream inputStream = file.getInputStream(); // 切割分片时的操作可以并行执行，创建任务列表 CompletableFuture<?>[] tasks = new CompletableFuture[chunks.size()]; byte[] buffer = new byte[0]; int preChunkNo = -1; for (int i = 0; i < chunks.size(); i++) { FileChunkMeta chunk = chunks.get(i); FileChunkDO fileChunkDO = new FileChunkDO(); Integer chunkSize = chunk.getChunkSize(); if (chunk.getChunkNo() != preChunkNo) { preChunkNo = chunk.getChunkNo(); buffer = new byte[chunkSize]; inputStream.read(buffer); } byte[] finalBuffer = buffer; tasks[i] = CompletableFuture.runAsync(() -> { // 如果已经上传直接return if (chunk.getCompleted()) { return; } String md5 = Md5Util.getMd5(finalBuffer); fileChunkDO.setFilename(chunk.getFilename()) .setExtension(chunk.getExtension()) .setChunkNo(chunk.getChunkNo()) .setChunkSize(chunkSize) .setBucketName(chunk.getBucketName()) .setBytes(finalBuffer); String schema = chunk.getSchema(); String address = chunkAddressStrategy.get(chunk); // 调用chunk-server存放文件分片 Object response = restTemplate.postForObject(address + "/file/write", fileChunkDO, Object.class); if (Objects.isNull(response)) { throw new BusinessException("第 " + chunk.getChunkNo() + " 个分片上传失败", EnumClientException.FAILED_TO_UPLOAD_CHUNK_FILE); } CommonResponse<String> md5Response = objectMapper.convertValue(response, new TypeReference<CommonResponse<String>>() { }); if (!md5Response.getData().equals(md5)) { throw new BusinessException(EnumClientException.CHUNK_FILE_INCOMPLETE); } CompleteChunkFileDO completeChunkFileDO = new CompleteChunkFileDO(); completeChunkFileDO.setFilename(chunk.getFilename()) .setChunkNo(chunk.getChunkNo()) .setAddress(chunk.getAddress()) .setSchema(chunk.getSchema()) .setMd5(md5); // 调用meta服务设置complete为true Object resp = restTemplate.postForObject(clientConfig.getMetaServerAddress() + "/meta/chunk/complete", completeChunkFileDO, Object.class); if (Objects.isNull(resp)) { throw new BusinessException(EnumClientException.FAILED_TO_UPDATE_CHUNK_FILE_COMPLETE_STATUS); } log.info("更新分片状态: {}", resp); }).whenComplete((o, throwable) -> { if (Objects.nonNull(throwable)) { throw new RuntimeException(MessageFormat.format("第 {} 个分片上传失败", chunk.getChunkNo())); } }); }