使用java+http+Range頭 實現視頻分段下載

背景：

在下載oss視頻時由於 oss后台進行優化 無法一次性下載完整個較大的視頻 所以需要分段下載。

直接下載會導致 Premature end of Content-Length delimited message body (expected 異常。

直接貼代碼：

package com.mybatis.plus.utils.rangeDownload;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.ReentrantLock;

@Component
public class MultiPartDownLoad {
 
    private static Logger logger = LoggerFactory.getLogger(MultiPartDownLoad.class);
 
    /**
     * 線程下載成功標志
     */
    private static int flag = 0;
 
    /**
     * 服務器請求路徑
     */
    private String serverPath;
    /**
     * 本地路徑
     */
    private String localPath;

    /**
     * refer頭
     */
    private String refer;
    /**
     * 線程計數同步輔助
     */
    private CountDownLatch latch;
 
    // 定長線程池
    private static ExecutorService threadPool;
 
    public MultiPartDownLoad(String serverPath, String localPath, String refer) {
        this.serverPath = serverPath;
        this.localPath = localPath;
        this.refer = refer;
    }

    public MultiPartDownLoad(String serverPath, String localPath) {
        this.serverPath = serverPath;
        this.localPath = localPath;
    }

    public MultiPartDownLoad() {
    }


 
    public boolean executeDownLoad() {
        try {
            URL url = new URL(serverPath);
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setConnectTimeout(5000);//設置超時時間
            conn.setRequestMethod("GET");//設置請求方式
            conn.setRequestProperty("Connection", "Keep-Alive");
            if (StringUtils.isNotEmpty(refer)) {
                conn.setRequestProperty("Refer", refer);
            }
            int code = conn.getResponseCode();
            if (code != 200 && code != 206) {
                logger.error(String.format("無效網絡地址：%s", serverPath));
                return false;
            }
            //服務器返回的數據的長度，實際上就是文件的長度,單位是字節
//            int length = conn.getContentLength();  //文件超過2G會有問題
            long length = getRemoteFileSize(serverPath);
 
            logger.info("文件總長度:" + length + "字節(B)");
            RandomAccessFile raf = new RandomAccessFile(localPath, "rwd");
            //指定創建的文件的長度
            raf.setLength(length);
            raf.close();
            //分割文件
            int partCount = Constans.MAX_THREAD_COUNT;
            int partSize = (int)(length / partCount);
            latch = new CountDownLatch(partCount);
            threadPool = Constans.getMyThreadPool();
            for (int threadId = 1; threadId <= partCount; threadId++) {
                // 每一個線程下載的開始位置
                long startIndex = (threadId - 1) * partSize;
                // 每一個線程下載的開始位置
                long endIndex = startIndex + partSize - 1;
                if (threadId == partCount) {
                    //最后一個線程下載的長度稍微長一點
                    endIndex = length;
                }
                logger.info("線程" + threadId + "下載:" + startIndex + "字節~" + endIndex + "字節");
                threadPool.execute(new DownLoadThread(threadId, startIndex, endIndex, latch));
            }
            latch.await();
            if(flag == 0){
                return true;
            }
        } catch (Exception e) {
            logger.error(String.format("文件下載失敗，文件地址：%s,失敗原因：%s", serverPath, e.getMessage()), e);
        }
        return false;
    }
 
 
    /**
     * 內部類用於實現下載
     */
    public class DownLoadThread implements Runnable {
        private Logger logger = LoggerFactory.getLogger(DownLoadThread.class);
 
        /**
         * 線程ID
         */
        private int threadId;
        /**
         * 下載起始位置
         */
        private long startIndex;
        /**
         * 下載結束位置
         */
        private long endIndex;
 
        private CountDownLatch latch;
 
        public DownLoadThread(int threadId, long startIndex, long endIndex, CountDownLatch latch) {
            this.threadId = threadId;
            this.startIndex = startIndex;
            this.endIndex = endIndex;
            this.latch = latch;
        }
 
        @Override
        public void run() {
            try {
                logger.info("線程" + threadId + "正在下載...");
                URL url = new URL(serverPath);
                HttpURLConnection conn = (HttpURLConnection) url.openConnection();
                conn.setRequestProperty("Connection", "Keep-Alive");
                conn.setRequestMethod("GET");
                //請求服務器下載部分的文件的指定位置
                conn.setRequestProperty("Range", "bytes=" + startIndex + "-" + endIndex);
                conn.setConnectTimeout(5000);
                int code = conn.getResponseCode();
                logger.info("線程" + threadId + "請求返回code=" + code);
                InputStream is = conn.getInputStream();//返回資源
                RandomAccessFile raf = new RandomAccessFile(localPath, "rwd");
                //隨機寫文件的時候從哪個位置開始寫
                raf.seek(startIndex);//定位文件
                int len = 0;
                byte[] buffer = new byte[1024];
                while ((len = is.read(buffer)) != -1) {
                    raf.write(buffer, 0, len);
                }
                is.close();
                raf.close();
                logger.info("線程" + threadId + "下載完畢");
            } catch (Exception e) {
                //線程下載出錯
                MultiPartDownLoad.flag = 1;
                logger.error(e.getMessage(),e);
            } finally {
                //計數值減一
                latch.countDown();
            }
 
        }
    }
 
    /**
     * 內部方法，獲取遠程文件大小
     * @param remoteFileUrl
     * @return
     * @throws IOException
     */
    private long getRemoteFileSize(String remoteFileUrl) throws IOException {
        long fileSize = 0;
        HttpURLConnection httpConnection = (HttpURLConnection) new URL(remoteFileUrl).openConnection();
        httpConnection.setRequestMethod("HEAD");
        int responseCode = 0;
        try {
            responseCode = httpConnection.getResponseCode();
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (responseCode >= 400) {
            logger.debug("Web服務器響應錯誤!");
            return 0;
        }
        String sHeader;
        for (int i = 1;; i++) {
            sHeader = httpConnection.getHeaderFieldKey(i);
            if (sHeader != null && sHeader.equals("Content-Length")) {
                fileSize = Long.parseLong(httpConnection.getHeaderField(sHeader));
                break;
            }
        }
        return fileSize;
    }

    /**
     * 下載文件執行器
     *
     * @param serverPath
     * @return
     */
    public synchronized static boolean downLoad(String serverPath, String localPath,String refer) {
        ReentrantLock lock = new ReentrantLock();
        lock.lock();
        MultiPartDownLoad m = new MultiPartDownLoad(serverPath, localPath, refer);
        long startTime = System.currentTimeMillis();
        boolean flag = false;
        try {
            flag = m.executeDownLoad();
            long endTime = System.currentTimeMillis();
            if (flag) {
                logger.info("文件下載結束,共耗時" + (endTime - startTime) + "ms");
                return true;
            }
            logger.warn("文件下載失敗");
            return false;
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
            return false;
        } finally {
            MultiPartDownLoad.flag = 0; // 重置 下載狀態
            if (!flag) {
                File file = new File(localPath);
                file.delete();
            }
            lock.unlock();
        }
    }
}

 

線程

package com.mybatis.plus.utils.rangeDownload;

import java.util.concurrent.*;
 
public class Constans {
//    public static final int MAX_THREAD_COUNT = getSystemProcessCount();
    public static final int MAX_THREAD_COUNT = 5;
    private static final int MAX_IMUMPOOLSIZE = MAX_THREAD_COUNT;
 
    /**
     * 自定義線程池
     */
    private static ExecutorService MY_THREAD_POOL;
    /**
     * 自定義線程池
     */
    public static ExecutorService getMyThreadPool(){
        if(MY_THREAD_POOL == null){
            MY_THREAD_POOL = Executors.newFixedThreadPool(MAX_IMUMPOOLSIZE);
        }
        return MY_THREAD_POOL;
    }
 
    // 線程池
    private static ThreadPoolExecutor threadPool;
 
    /**
     * 單例，單任務 線程池
     * @return
     */
    public static ThreadPoolExecutor getThreadPool(){
        if(threadPool == null){
            threadPool = new ThreadPoolExecutor(MAX_IMUMPOOLSIZE, MAX_IMUMPOOLSIZE, 3, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<Runnable>(16),
                    new ThreadPoolExecutor.CallerRunsPolicy()
            );
        }
        return threadPool;
    }
 
    /**
     * 獲取服務器cpu核數
     * @return
     */
    private static int getSystemProcessCount(){
        int count = Runtime.getRuntime().availableProcessors();
        return count;
    }
 
}

調用方法：

public static void downloadPictureWithOSS(String learnVideoSoureUrl, String filePath, String name,String referer) {
        logger.info("========正在下載==" + name);
        for (int i = 0; i < 5; i++) {
//            DefaultHttpClient httpclient = HttpClientUtil.getNewHttpClient();
//            HttpGet httpget = new HttpGet(learnVideoSoureUrl);
//            httpget.setHeader("Referer",referer);
//            if (executeDownload(filePath, name, httpclient, httpget)) break;
            if (MultiPartDownLoad.downLoad(learnVideoSoureUrl, filePath + name, referer)) {
                break;
            }
        }
    }

調用結果：