电脑知识|欧美黑人一区二区三区|软件|欧美黑人一级爽快片淫片高清|系统|欧美黑人狂野猛交老妇|数据库|服务器|编程开发|网络运营|知识问答|技术教程文章 - 好吧啦网

您的位置:首頁技術文章
文章詳情頁

基于Java實現多線程下載并允許斷點續傳

瀏覽:40日期:2022-09-04 15:13:17

完整代碼:https://github.com/iyuanyb/Downloader

多線程下載及斷點續傳的實現是使用 HTTP/1.1 引入的 Range 請求參數,可以訪問Web資源的指定區間的內容。雖然實現了多線程及斷點續傳,但還有很多不完善的地方。

包含四個類:

Downloader: 主類,負責分配任務給各個子線程,及檢測進度DownloadFile: 表示要下載的哪個文件,為了能寫輸入到文件的指定位置,使用 RandomAccessFile 類操作文件,多個線程寫同一個文件需要保證線程安全,這里直接調用 getChannel 方法,獲取一個文件通道,FileChannel是線程安全的。DownloadTask: 實際執行下載的線程,獲取 [lowerBound, upperBound] 區間的數據,當下載過程中出現異常時要通知其他線程(使用 AtomicBoolean),結束下載Logger: 實時記錄下載進度,以便續傳時知道從哪開始。感覺這里做的比較差,為了能實時寫出日志及方便地使用Properties類的load/store方法格式化輸入輸出,每次都是打開后再關閉。

演示:

隨便找一個文件下載:

基于Java實現多線程下載并允許斷點續傳

強行結束程序并重新運行:

基于Java實現多線程下載并允許斷點續傳

日志文件:

斷點續傳的關鍵是記錄各個線程的下載進度,這里細節比較多,花了很久。只需要記錄每個線程請求的Range區間極客,每次成功寫數據到文件時,就更新一次下載區間。下面是下載完成后的日志內容。

基于Java實現多線程下載并允許斷點續傳

代碼:

Downloader.java

package downloader; import java.io.*;import java.net.*;import java.nio.file.Files;import java.nio.file.Path;import java.util.concurrent.atomic.AtomicBoolean; public class Downloader { private static final int DEFAULT_THREAD_COUNT = 4; // 默認線程數量 private AtomicBoolean canceled; // 取消狀態,如果有一個子線程出現異常,則取消整個下載任務 private DownloadFile file; // 下載的文件對象 private String storageLocation; private final int threadCount; // 線程數量 private long fileSize; // 文件大小 private final String url; private long beginTime; // 開始時間 private Logger logger; public Downloader(String url) { this(url, DEFAULT_THREAD_COUNT); } public Downloader(String url, int threadCount) { this.url = url; this.threadCount = threadCount; this.canceled = new AtomicBoolean(false); this.storageLocation = url.substring(url.lastIndexOf(’/’)+1); this.logger = new Logger(storageLocation + '.log', url, threadCount); } public void start() { boolean reStart = Files.exists(Path.of(storageLocation + '.log')); if (reStart) { logger = new Logger(storageLocation + '.log'); System.out.printf('* 繼續上次下載進度[已下載:%.2fMB]:%sn', logger.getWroteSize() / 1014.0 / 1024, url); } else { System.out.println('* 開始下載:' + url); } if (-1 == (this.fileSize = getFileSize())) return; System.out.printf('* 文件大小:%.2fMBn', fileSize / 1024.0 / 1024); this.beginTime = System.currentTimeMillis(); try { this.file = new DownloadFile(storageLocation, fileSize, logger); if (reStart) {file.setWroteSize(logger.getWroteSize()); } // 分配線程下載 dispatcher(reStart); // 循環打印進度 printDownloadProgress(); } catch (IOException e) { System.err.println('x 創建文件失敗[' + e.getMessage() + ']'); } } /** * 分配器,決定每個線程下載哪個區間的數據 */ private void dispatcher(boolean reStart) { long blockSize = fileSize / threadCount; // 每個線程要下載的數據量 long lowerBound = 0, upperBound = 0; long[][] bounds = null; int threadID = 0; if (reStart) { bounds = logger.getBounds(); } for (int i = 0; i < threadCount; i++) { if (reStart) {threadID = (int)(bounds[i][0]);lowerBound = bounds[i][1];upperBound = bounds[i][2]; } else {threadID = i;lowerBound = i * blockSize;// fileSize-1 !!!!! fu.ck,找了一下午的錯upperBound = (i == threadCount - 1) ? fileSize-1 : lowerBound + blockSize; } new DownloadTask(url, lowerBound, upperBound, file, canceled, threadID).start(); } } /** * 循環打印進度,直到下載完畢,或任務被取消 */ private void printDownloadProgress() { long downloadedSize = file.getWroteSize(); int i = 0; long lastSize = 0; // 三秒前的下載量 while (!canceled.get() && downloadedSize < fileSize) { if (i++ % 4 == 3) { // 每3秒打印一次System.out.printf('下載進度:%.2f%%, 已下載:%.2fMB,當前速度:%.2fMB/sn', downloadedSize / (double)fileSize * 100 , downloadedSize / 1024.0 / 1024, (downloadedSize - lastSize) / 1024.0 / 1024 / 3);lastSize = downloadedSize;i = 0; } try {Thread.sleep(1000); } catch (InterruptedException ignore) {} downloadedSize = file.getWroteSize(); } file.close(); if (canceled.get()) { try {Files.delete(Path.of(storageLocation)); } catch (IOException ignore) { } System.err.println('x 下載失敗,任務已取消'); } else { System.out.println('* 下載成功,本次用時'+ (System.currentTimeMillis() - beginTime) / 1000 +'秒'); } } /** * @return 要下載的文件的尺寸 */ private long getFileSize() { if (fileSize != 0) { return fileSize; } HttpURLConnection conn = null; try { conn = (HttpURLConnection)new URL(url).openConnection(); conn.setConnectTimeout(3000); conn.setRequestMethod('HEAD'); conn.connect(); System.out.println('* 連接服務器成功'); } catch (MalformedURLException e) { throw new RuntimeException('URL錯誤'); } catch (IOException e) { System.err.println('x 連接服務器失敗['+ e.getMessage() +']'); return -1; } return conn.getContentLengthLong(); } public static void main(String[] args) throws IOException { new Downloader('http://js.xiazaicc.com//down2/ucliulanqi_downcc.zip').start(); }}

DownloadTask.java

package downloader; import java.io.*;import java.net.HttpURLConnection;import java.net.URL;import java.nio.ByteBuffer;import java.nio.channels.Channels;import java.nio.channels.ReadableByteChannel;import java.util.concurrent.atomic.AtomicBoolean; class DownloadTask extends Thread { private final String url; private long lowerBound; // 下載的文件區間 private long upperBound; private AtomicBoolean canceled; private DownloadFile downloadFile; private int threadId; DownloadTask(String url, long lowerBound, long upperBound, DownloadFile downloadFile, AtomicBoolean canceled, int threadID) { this.url = url; this.lowerBound = lowerBound; this.upperBound = upperBound; this.canceled = canceled; this.downloadFile = downloadFile; this.threadId = threadID; } @Override public void run() { ReadableByteChannel input = null; try { ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024 * 2); // 2MB input = connect(); System.out.println('* [線程' + threadId + ']連接成功,開始下載...'); int len; while (!canceled.get() && lowerBound <= upperBound) {buffer.clear();len = input.read(buffer);downloadFile.write(lowerBound, buffer, threadId, upperBound);lowerBound += len; } if (!canceled.get()) {System.out.println('* [線程' + threadId + ']下載完成' + ': ' + lowerBound + '-' + upperBound); } } catch (IOException e) { canceled.set(true); System.err.println('x [線程' + threadId + ']遇到錯誤[' + e.getMessage() + '],結束下載'); } finally { if (input != null) {try { input.close();} catch (IOException e) { e.printStackTrace();} } } } /** * 連接WEB服務器,并返回一個數據通道 * @return 返回通道 * @throws IOException 網絡連接錯誤 */ private ReadableByteChannel connect() throws IOException { HttpURLConnection conn = (HttpURLConnection)new URL(url).openConnection(); conn.setConnectTimeout(3000); conn.setRequestMethod('GET'); conn.setRequestProperty('Range', 'bytes=' + lowerBound + '-' + upperBound);// System.out.println('thread_'+ threadId +': ' + lowerBound + '-' + upperBound); conn.connect(); int statusCode = conn.getResponseCode(); if (HttpURLConnection.HTTP_PARTIAL != statusCode) { conn.disconnect(); throw new IOException('狀態碼錯誤:' + statusCode); } return Channels.newChannel(conn.getInputStream()); }}

DownloadFile.java

package downloader; import java.io.IOException;import java.io.RandomAccessFile;import java.nio.ByteBuffer;import java.nio.channels.FileChannel;import java.util.concurrent.atomic.AtomicLong; class DownloadFile { private final RandomAccessFile file; private final FileChannel channel; // 線程安全類 private AtomicLong wroteSize; // 已寫入的長度 private Logger logger; DownloadFile(String fileName, long fileSize, Logger logger) throws IOException { this.wroteSize = new AtomicLong(0); this.logger = logger; this.file = new RandomAccessFile(fileName, 'rw'); file.setLength(fileSize); channel = file.getChannel(); } /** * 寫數據 * @param offset 寫偏移量 * @param buffer 數據 * @throws IOException 寫數據出現異常 */ void write(long offset, ByteBuffer buffer, int threadID, long upperBound) throws IOException { buffer.flip(); int length = buffer.limit(); while (buffer.hasRemaining()) { channel.write(buffer, offset); } wroteSize.addAndGet(length); logger.updateLog(threadID, length, offset + length, upperBound); // 更新日志 } /** * @return 已經下載的數據量,為了知道何時結束整個任務,以及統計信息 */ long getWroteSize() { return wroteSize.get(); } // 繼續下載時調用 void setWroteSize(long wroteSize) { this.wroteSize.set(wroteSize); } void close() { try { file.close(); } catch (IOException e) { e.printStackTrace(); } }}

Logger.java

package downloader; import java.io.*;import java.util.Properties; class Logger { private String logFileName; // 下載的文件的名字 private Properties log; /** * 重新開始下載時,使用該構造函數 * @param logFileName */ Logger(String logFileName) { this.logFileName = logFileName; log = new Properties(); FileInputStream fin = null; try { log.load(new FileInputStream(logFileName)); } catch (IOException ignore) { } finally { try {fin.close(); } catch (Exception ignore) {} } } Logger(String logFileName, String url, int threadCount) { this.logFileName = logFileName; this.log = new Properties(); log.put('url', url); log.put('wroteSize', '0'); log.put('threadCount', String.valueOf(threadCount)); for (int i = 0; i < threadCount; i++) { log.put('thread_' + i, '0-0'); } } synchronized void updateLog(int threadID, long length, long lowerBound, long upperBound) { log.put('thread_'+threadID, lowerBound + '-' + upperBound); log.put('wroteSize', String.valueOf(length + Long.parseLong(log.getProperty('wroteSize')))); FileOutputStream file = null; try { file = new FileOutputStream(logFileName); // 每次寫時都清空文件 log.store(file, null); } catch (IOException e) { e.printStackTrace(); } finally { if (file != null) {try { file.close();} catch (IOException e) { e.printStackTrace();} } } } /** * 獲取區間信息 * ret[i][0] = threadID, ret[i][1] = lowerBoundID, ret[i][2] = upperBoundID * @return */ long[][] getBounds() { long[][] bounds = new long[Integer.parseInt(log.get('threadCount').toString())][3]; int[] index = {0}; log.forEach((k, v) -> { String key = k.toString(); if (key.startsWith('thread_')) {String[] interval = v.toString().split('-');bounds[index[0]][0] = Long.parseLong(key.substring(key.indexOf('_') + 1));bounds[index[0]][1] = Long.parseLong(interval[0]);bounds[index[0]++][2] = Long.parseLong(interval[1]); } }); return bounds; } long getWroteSize() { return Long.parseLong(log.getProperty('wroteSize')); }}

以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持好吧啦網。

標簽: Java
相關文章:
主站蜘蛛池模板: 烟气在线监测系统_烟气在线监测仪_扬尘检测仪_空气质量监测站「山东风途物联网」 | 睿婕轻钢别墅_钢结构别墅_厂家设计施工报价 | 钢丝绳探伤仪-钢丝绳检测仪-钢丝绳探伤设备-洛阳泰斯特探伤技术有限公司 | 2025世界机器人大会_IC China_半导体展_集成电路博览会_智能制造展览网 | 高精度-恒温冷水机-螺杆式冰水机-蒸发冷冷水机-北京蓝海神骏科技有限公司 | 金属清洗剂,防锈油,切削液,磨削液-青岛朗力防锈材料有限公司 | 「钾冰晶石」氟铝酸钾_冰晶石_氟铝酸钠「价格用途」-亚铝氟化物厂家 | 紧急切断阀_气动切断阀_不锈钢阀门_截止阀_球阀_蝶阀_闸阀-上海上兆阀门制造有限公司 | 合金耐磨锤头_破碎机锤头_郑州市德勤建材有限公司 | 北京中航时代-耐电压击穿试验仪厂家-电压击穿试验机 | 代写标书-专业代做标书-商业计划书代写「深圳卓越创兴公司」 | 明渠式紫外线杀菌器-紫外线消毒器厂家-定州市优威环保 | MOOG伺服阀维修,ATOS比例流量阀维修,伺服阀维修-上海纽顿液压设备有限公司 | 实验室pH计|电导率仪|溶解氧测定仪|离子浓度计|多参数水质分析仪|pH电极-上海般特仪器有限公司 | 混合生育酚_醋酸生育酚粉_琥珀酸生育酚-山东新元素生物科技 | 广州/东莞小字符喷码机-热转印打码机-喷码机厂家-广州瑞润科技 | 超声骨密度仪,双能X射线骨密度仪【起草单位】,骨密度检测仪厂家 - 品源医疗(江苏)有限公司 | 蒸压釜-陶粒板隔墙板蒸压釜-山东鑫泰鑫智能装备有限公司 | 螺杆式冷水机-低温冷水机厂家-冷冻机-风冷式-水冷式冷水机-上海祝松机械有限公司 | 江苏密集柜_电动_手动_移动_盛隆柜业江苏档案密集柜厂家 | 上海瑶恒实业有限公司|消防泵泵|离心泵|官网 | 新密高铝耐火砖,轻质保温砖价格,浇注料厂家直销-郑州荣盛窑炉耐火材料有限公司 | 棕刚玉_白刚玉_铝酸钙-锐石新材料| 南京展台搭建-南京展会设计-南京展览设计公司-南京展厅展示设计-南京汇雅展览工程有限公司 | 涂层测厚仪_漆膜仪_光学透过率仪_十大创新厂家-果欧电子科技公司 | 谈股票-今日股票行情走势分析-牛股推荐排行榜 | 闭端端子|弹簧螺式接线头|防水接线头|插线式接线头|端子台|电源线扣+护线套|印刷电路板型端子台|金笔电子代理商-上海拓胜电气有限公司 | 抖音短视频运营_企业网站建设_网络推广_全网自媒体营销-东莞市凌天信息科技有限公司 | 管理会计网-PCMA初级管理会计,中级管理会计考试网站 | 创客匠人-让IP变现不走弯路 | 拉力机-万能试验机-材料拉伸试验机-电子拉力机-拉力试验机厂家-冲击试验机-苏州皖仪实验仪器有限公司 | 天津仓库出租网-天津电商仓库-天津云仓一件代发-【博程云仓】 | 飞扬动力官网-广告公司管理软件,广告公司管理系统,喷绘写真条幅制作管理软件,广告公司ERP系统 | 今日娱乐圈——影视剧集_八卦娱乐_明星八卦_最新娱乐八卦新闻 | 地源热泵一体机,地源热泵厂家-淄博汇能环保设备有限公司 | 手持式3d激光扫描仪-便携式三维立体扫描仪-北京福禄克斯 | 水性绝缘漆_凡立水_绝缘漆树脂_环保绝缘漆-深圳维特利环保材料有限公司 | 柔软云母板-硬质-水位计云母片组件-首页-武汉长丰云母绝缘材料有限公司 | 网优资讯-为循环资源、大宗商品、工业服务提供资讯与行情分析的数据服务平台 | 热处理温控箱,热处理控制箱厂家-吴江市兴达电热设备厂 | 免费分销系统 — 分销商城系统_分销小程序开发 -【微商来】 |