多线程下载资源后分片上传至oss

偏执的太偏执、 2022-09-02 05:29 221阅读 0赞

使用多线程下载资源后，分片上传至阿里oss(主流对象存储都支持分片上传)。实现边下载边上传

注意：下载的资源url必须支持部分资源返回。http状态码：206

#### 下载执行类DownloadTask ####

该类实现callable接口，负责下载资源的某一部分

使用CloseableHttpClient进行下载

package com.tinet.clink.chat.web.service.rtc;
    
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.http.HttpStatus;
    import org.apache.http.client.methods.CloseableHttpResponse;
    import org.apache.http.client.methods.HttpGet;
    import org.apache.http.impl.client.CloseableHttpClient;
    import org.apache.http.util.EntityUtils;
    
    import java.io.ByteArrayInputStream;
    import java.io.IOException;
    import java.io.InputStream;
    import java.util.concurrent.Callable;
    
    
    /** * @author dengsx * @create 2021-06-11 **/
    @Slf4j
    class DownloadTask implements Callable<InputStream> { 
        private final String url;
        private final long lowerBound; // 下载的文件区间
        private final long upperBound;
        private final CloseableHttpClient client;
    
        DownloadTask(String url, long lowerBound, long upperBound, CloseableHttpClient client) { 
            this.url = url;
            this.lowerBound = lowerBound;
            this.upperBound = upperBound;
            this.client = client;
        }
    
        @Override
        public InputStream call() throws Exception { 
            try (InputStream input = connect()) { 
                log.info("线程{}下载完成，下载区间：{}-{},下载字节：{}", Thread.currentThread().getName(),
                        this.lowerBound, this.upperBound, input.available());
                return input;
            } catch (IOException e) { 
                throw new IOException(e.getMessage());
            }
        }
    
        /** * 连接WEB服务器，并返回一个数据流 * * @return 返回输入流 * @throws IOException 网络连接错误 */
        private InputStream connect() throws IOException { 
            HttpGet get = new HttpGet(url);
            get.setHeader("Range", "bytes=" + lowerBound + "-" + upperBound);
            try (CloseableHttpResponse response = client.execute(get)) { 
                log.info("线程{}连接成功开始下载", Thread.currentThread().getName());
                int statusCode = response.getStatusLine().getStatusCode();
                if (HttpStatus.SC_PARTIAL_CONTENT != statusCode) { 
                    throw new IOException("下载连接不支持多线程下载：" + statusCode);
                }
    
                return new ByteArrayInputStream(EntityUtils.toByteArray(response.getEntity()));
            } catch (IOException e) { 
                throw new IOException("下出错:" + e.getMessage());
            }
        }
    }

#### 上传任务类UploadTask ####

该类负责将下载的资源上传至oss,使用阿里oss的分片上传

package com.tinet.clink.chat.web.service.rtc;
    
    import com.tinet.clink.common.model.PartETagAdapter;
    import com.tinet.clink.common.objectstorage.ObjectStorageClient;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.util.concurrent.Callable;
    
    
    /** * @author dengsx * @create 2021-06-11 **/
    @Slf4j
    public class UploadTask implements Callable<PartETagAdapter> { 
        private final InputStream inputStream;
        private final ObjectStorageClient objectStorageClient;
        private final int order;
        private final long partSize;
        private final String uploadId;
        private final String fileBucket;
        private final String key;
    
        public UploadTask(InputStream inputStream, ObjectStorageClient objectStorageService, String fileBucket, String key, String uploadId, int order, long partSize) { 
            this.inputStream = inputStream;
            this.objectStorageClient = objectStorageService;
            this.order = order;
            this.uploadId = uploadId;
            this.key = key;
            this.fileBucket = fileBucket;
            this.partSize = partSize;
        }
    
        @Override
        public PartETagAdapter call() throws IOException { 
            log.info("开始上传分片，分片order:{},分片大小：{}", order, partSize);
            return objectStorageClient.uploadPartObject(fileBucket, key, uploadId, inputStream, order, partSize);
        }
    }

#### 下载器类Downloader ####

该类负责，分配任务给线程下载，上传，合并结果

package com.tinet.clink.chat.web.service.rtc;
    
    import com.tinet.clink.common.model.CompleteMultipartUploadResultAdapter;
    import com.tinet.clink.common.model.PartETagAdapter;
    import com.tinet.clink.common.objectstorage.ObjectStorageClient;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.http.HeaderElement;
    import org.apache.http.HeaderElementIterator;
    import org.apache.http.client.methods.HttpHead;
    import org.apache.http.conn.ConnectionKeepAliveStrategy;
    import org.apache.http.impl.client.CloseableHttpClient;
    import org.apache.http.impl.client.HttpClients;
    import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
    import org.apache.http.message.BasicHeaderElementIterator;
    import org.apache.http.protocol.HTTP;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.util.*;
    import java.util.concurrent.Future;
    import java.util.concurrent.TimeUnit;
    
    /** * 从腾讯云下载视频上传至oss * 多线程下载 * 边下载边上传 * * @author dengsx * @create 2021-06-11 **/
    @Component
    @Slf4j
    public class Downloader implements InitializingBean { 
    
    
        private CloseableHttpClient httpClient;
    
        private static final int DEFAULT_THREAD_COUNT;
    
        /** * 阻塞系数 大致 0.5 */
        private static final double BLOCK_FACTOR = 0.5;
    
        static { 
            // io密集型任务，线程数设置为cpu核数的两倍
            Double threadCount = Runtime.getRuntime().availableProcessors() / (1 - BLOCK_FACTOR);
            DEFAULT_THREAD_COUNT = threadCount.intValue();
        }
    
        @Autowired
        @Qualifier("downloaderTaskExecutor")
        private ThreadPoolTaskExecutor downloaderTaskExecutor;
    
        /** * 会话文件上传的桶 */
        @Value("${object.storage.bucket.chat}")
        private String fileBucket;
    
        @Autowired
        private ObjectStorageClient objectStorageClient;
    
        public CompleteMultipartUploadResultAdapter start(String url, String fileKey) throws Exception { 
            return this.start(url, DEFAULT_THREAD_COUNT, fileKey);
        }
    
        public CompleteMultipartUploadResultAdapter start(String url, Integer threadCount, String fileKey) throws Exception { 
    
            if (threadCount == null || threadCount < 0) { 
                throw new IllegalArgumentException("invalid threadCount number!");
            }
    
            threadCount = Math.min(DEFAULT_THREAD_COUNT, threadCount);
    
            long fileSize = getFileSize(url);
    
            log.info("开始下载，文件大小(kb):{},fileSize(Mb):{},下载线程数：{}", fileSize, fileSize / 1024.0 / 1024, threadCount);
    
            //记录时间
            long beginTime = System.currentTimeMillis();
            // 分配线程下载
            Map<Integer, Future<InputStream>> downloadFuture = dispatcher(url, fileSize, threadCount);
    
            //分片上传id
            String uploadId = objectStorageClient.getMultipartUploadId(fileBucket, fileKey);
    
            // 处理下载结果
            List<Future<PartETagAdapter>> uploadFuture = handleDownloadFuture(downloadFuture, uploadId, fileKey, threadCount, fileSize);
    
            //处理上传结果
            List<PartETagAdapter> uploadResult = handleUploadFuture(uploadFuture);
    
            //分片上传合并
            CompleteMultipartUploadResultAdapter result = objectStorageClient.completePartlyUpload(fileBucket, fileKey, uploadId, uploadResult);
    
            log.info("下载成功并上传oss成功：本次用时：{}秒,上传结果{},", (System.currentTimeMillis() - beginTime) / 1000, result.getKey());
            return result;
    
        }
    
        /** * 分配器，决定每个线程下载哪个区间的数据 */
        private Map<Integer, Future<InputStream>> dispatcher(String url, long fileSize, int threadCount) { 
            long blockSize = fileSize / threadCount; // 每个线程要下载的数据量
            long lowerBound;
            long upperBound;
            Map<Integer, Future<InputStream>> result = new HashMap<>();
            for (int i = 0; i < threadCount; i++) { 
    
                lowerBound = i * blockSize;
                // 最后一个分片的上限是文件大小
                upperBound = (i == threadCount - 1) ? fileSize - 1 : lowerBound + blockSize;
                // 异步提交下载任务
                Future<InputStream> future = downloaderTaskExecutor.submit(new DownloadTask(url, lowerBound, upperBound, httpClient));
                result.put(i + 1, future);
            }
            return result;
        }
    
    
        /** * @return 要下载的文件的尺寸 */
        private long getFileSize(String url) throws IOException { 
            HttpHead head = new HttpHead(url);
            String lengthValue = httpClient.execute(head).getHeaders("Content-Length")[0].getValue();
            return Long.parseLong(lengthValue);
    
        }
    
        private List<Future<PartETagAdapter>> handleDownloadFuture(Map<Integer, Future<InputStream>> downloadFuture,
                                                                   String uploadId, String key,
                                                                   int threadCount, long fileSize) throws Exception { 
            long blockSize = fileSize / threadCount;
            List<Future<PartETagAdapter>> uploadFuture = new ArrayList<>();
            while (!downloadFuture.isEmpty()) { 
                for (Iterator<Map.Entry<Integer, Future<InputStream>>> it = downloadFuture.entrySet().iterator(); it.hasNext(); ) { 
                    Map.Entry<Integer, Future<InputStream>> currentFuture = it.next();
                    if (currentFuture.getValue().isDone()) { 
                        Integer partNumber = currentFuture.getKey();
                        long partSize = blockSize;
                        if (Objects.equals(partNumber, threadCount)) { 
                            partSize = fileSize - (threadCount - 1) * blockSize;
                        }
                        InputStream inputStream = currentFuture.getValue().get();
    
                        //异步提交上传任务
                        Future<PartETagAdapter> tagFuture = downloaderTaskExecutor.submit(new UploadTask(inputStream,
                                objectStorageClient, fileBucket, key, uploadId, currentFuture.getKey(), partSize));
                        uploadFuture.add(tagFuture);
                        it.remove();
                    }
                }
                sleep(1000);
            }
            log.info("下载结果处理完成，上传任务提交完成");
            return uploadFuture;
        }
    
        private List<PartETagAdapter> handleUploadFuture(List<Future<PartETagAdapter>> partETagsFutures) throws Exception { 
            List<PartETagAdapter> uploadResult = new ArrayList<>();
            while (!partETagsFutures.isEmpty()) { 
                for (Iterator<Future<PartETagAdapter>> iterator = partETagsFutures.iterator(); iterator.hasNext(); ) { 
                    Future<PartETagAdapter> next = iterator.next();
                    if (next.isDone()) { 
                        PartETagAdapter partETag = next.get();
                        uploadResult.add(partETag);
                        iterator.remove();
                    }
                }
                sleep(1000);
            }
            log.info("上传任务结果处理完成");
            return uploadResult;
        }
    
        private void sleep(int millis) { 
            try { 
                TimeUnit.MILLISECONDS.sleep(millis);
            } catch (InterruptedException e) { 
                //ignore
            }
        }
    
        @Override
        public void afterPropertiesSet() { 
            // 设置keep alive的连接策略
            ConnectionKeepAliveStrategy strategy = (response, context) -> { 
                // Honor 'keep-alive' header
                HeaderElementIterator it = new BasicHeaderElementIterator(
                        response.headerIterator(HTTP.CONN_KEEP_ALIVE));
                while (it.hasNext()) { 
                    HeaderElement he = it.nextElement();
                    String param = he.getName();
                    String value = he.getValue();
                    if (value != null && "timeout".equalsIgnoreCase(param)) { 
                        return Long.parseLong(value) * 1000;
                    }
                }
                return 5 * 1000;
            };
    
            PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
            connectionManager.setMaxTotal(20);
            connectionManager.setDefaultMaxPerRoute(5);
    
            httpClient = HttpClients.custom()
                    .setKeepAliveStrategy(strategy).setConnectionManager(connectionManager).build();
        }
    }

#### 阿里oss分片上传步骤 ####

##### 1.获取分片上传id #####

/** * 获取分片上传id * * @param bucketName * @param key * @return */
        public String getMultipartUploadId(String bucketName, String key) { 
            // 创建InitiateMultipartUploadRequest对象。
            InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucketName, key);
    
            // 如果需要在初始化分片时设置文件存储类型，请参考以下示例代码。
            // ObjectMetadata metadata = new ObjectMetadata();
            // metadata.setHeader(OSSHeaders.OSS_STORAGE_CLASS, StorageClass.Standard.toString());
            // request.setObjectMetadata(metadata);
    
            // 初始化分片。
            InitiateMultipartUploadResult upResult = getInternalOSSClient().initiateMultipartUpload(request);
            // 返回uploadId，它是分片上传事件的唯一标识，可以根据这个uploadId发起相关的操作，如取消分片上传、查询分片上传等。
            return upResult.getUploadId();
        }

##### 2.上传分片 #####

只需要将分片的流等参数准备好即可

/** * 分片上传 * * @param bucketName * @param key * @param uploadId * @param inputStream 要上传的流 * @param order 分片上传的序号，后面根据这序号拼接完整的文件 * @return */
        public PartETagAdapter uploadPartObject(String bucketName, String key, String uploadId, InputStream inputStream, int order, long fileSize) throws IOException { 
            UploadPartRequest uploadPartRequest = new UploadPartRequest();
            uploadPartRequest.setBucketName(bucketName);
            uploadPartRequest.setKey(key);
            uploadPartRequest.setUploadId(uploadId);
            uploadPartRequest.setInputStream(inputStream);
            // 设置分片大小。除了最后一个分片没有大小限制，其他的分片最小为100 KB。
            uploadPartRequest.setPartSize(fileSize);
            // 设置分片号。每一个上传的分片都有一个分片号，取值范围是1~10000，如果超出这个范围，OSS将返回InvalidArgument的错误码。
            uploadPartRequest.setPartNumber(order);
            // 每个分片不需要按顺序上传，甚至可以在不同客户端上传，OSS会按照分片号排序组成完整的文件。
            PartETagAdapter ossPartETagAdapter = new PartETagAdapter();
            ossPartETagAdapter.setOssPartETag(getInternalOSSClient().uploadPart(uploadPartRequest).getPartETag());
            return ossPartETagAdapter;
        }

##### 3.合并上传结果 #####

/** * 完成分片上传 * * @param bucketName * @param key * @param uploadId * @param partETags * @return */
        public CompleteMultipartUploadResultAdapter completePartlyUpload(String bucketName, String key, String uploadId, List<PartETagAdapter> partETags) { 
            // 创建CompleteMultipartUploadRequest对象。
            // 在执行完成分片上传操作时，需要提供所有有效的partETags。OSS收到提交的partETags后，会逐一验证每个分片的有效性。当所有的数据分片验证通过后，OSS将把这些分片组合成一个完整的文件。
            CompleteMultipartUploadRequest completeMultipartUploadRequest =
                    new CompleteMultipartUploadRequest(bucketName, key, uploadId,
                            partETags.stream().map(PartETagAdapter::getOssPartETag).collect(Collectors.toList()));
    
            // 如果需要在完成文件上传的同时设置文件访问权限，请参考以下示例代码。
            // completeMultipartUploadRequest.setObjectACL(CannedAccessControlList.PublicRead);
    
            // 完成上传。
            CompleteMultipartUploadResultAdapter resultAdapter = new CompleteMultipartUploadResultAdapter();
            resultAdapter.setOssMultiPartUploadResult(getInternalOSSClient().completeMultipartUpload(completeMultipartUploadRequest));
            return resultAdapter;
        }

end~