基于Java实现多线程下载并允许断点续传

 更新时间:2020年03月13日 09:09:13   作者:yuanyb  
这篇文章主要介绍了基于Java实现多线程下载并允许断点续传,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下

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

多线程下载及断点续传的实现是使用 HTTP/1.1 引入的 Range 请求参数,可以访问Web资源的指定区间的内容。虽然实现了多线程及断点续传,但还有很多不完善的地方。

包含四个类:

Downloader: 主类,负责分配任务给各个子线程,及检测进度DownloadFile: 表示要下载的哪个文件,为了能写输入到文件的指定位置,使用 RandomAccessFile 类操作文件,多个线程写同一个文件需要保证线程安全,这里直接调用 getChannel 方法,获取一个文件通道,FileChannel是线程安全的。DownloadTask: 实际执行下载的线程,获取 [lowerBound, upperBound] 区间的数据,当下载过程中出现异常时要通知其他线程(使用 AtomicBoolean),结束下载Logger: 实时记录下载进度,以便续传时知道从哪开始。感觉这里做的比较差,为了能实时写出日志及方便地使用Properties类的load/store方法格式化输入输出,每次都是打开后再关闭。

演示:

随便找一个文件下载:

强行结束程序并重新运行:

日志文件:

断点续传的关键是记录各个线程的下载进度,这里细节比较多,花了很久。只需要记录每个线程请求的Range区间极客,每次成功写数据到文件时,就更新一次下载区间。下面是下载完成后的日志内容。

代码:

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]:%s\n", logger.getWroteSize() / 1014.0 / 1024, url);
    } else {
      System.out.println("* 开始下载:" + url);
    }
    if (-1 == (this.fileSize = getFileSize()))
      return;
    System.out.printf("* 文件大小:%.2fMB\n", 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/s\n",
            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中Parser的用法

    Java中Parser的用法

    这篇文章主要介绍了Java Parser使用指南,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2023-05-05
  • 详解Java中static关键字的使用和原理

    详解Java中static关键字的使用和原理

    static可以用来修饰的成员变量和成员方法,被修饰的成员是属于类的,而不是单单是属于某个对象的。本文就来详细说说他的使用和原理,需要的可以参考一下
    2022-10-10
  • Java8 Optional优雅空值判断的示例代码

    Java8 Optional优雅空值判断的示例代码

    这篇文章主要介绍了Java8 Optional优雅空值判断的相关知识,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2021-05-05
  • 让你五分钟彻底理解Spring MVC

    让你五分钟彻底理解Spring MVC

    其实MVC就是处理Web请求的一种框架模式,如果你对MVC不太熟悉的话可以看下本文,这篇文章主要给大家介绍了关于如何让你五分钟彻底理解Spring MVC的相关资料,需要的朋友可以参考下
    2021-10-10
  • Spring security中的授权

    Spring security中的授权

    本篇为大家带来Spring security的授权,首先要理解一些概念,有关于:权限、角色、安全上下文、访问控制表达式、方法级安全性、访问决策管理器,这篇文章主要介绍了Spring security中的授权,需要的朋友可以参考下
    2024-01-01
  • Java高并发系统限流算法的实现

    Java高并发系统限流算法的实现

    这篇文章主要介绍了Java高并发系统限流算法的应用,在开发高并发系统时有三把利器用来保护系统:缓存、降级和限流,限流可以认为服务降级的一种,限流是对系统的一种保护措施,需要的朋友可以参考下
    2022-05-05
  • Java日常练习题,每天进步一点点(52)

    Java日常练习题,每天进步一点点(52)

    下面小编就为大家带来一篇Java基础的几道练习题(分享)。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧,希望可以帮到你
    2021-08-08
  • 浅聊一下Spring Security的使用方法

    浅聊一下Spring Security的使用方法

    Spring Security 是一个基于 Spring 框架的安全框架,提供了一套安全性认证和授权的解决方案,用于保护 Web 应用程序和服务,接下来小编就和大家聊聊Spring Security,感兴趣的小伙伴跟着小编一起来看看吧
    2023-08-08
  • Java后台如何处理日期参数格式

    Java后台如何处理日期参数格式

    这篇文章主要介绍了Java后台如何处理日期参数格式问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2023-07-07
  • SPRING IOC注入方式过程解析

    SPRING IOC注入方式过程解析

    这篇文章主要介绍了SPRING IOC注入方式过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-01-01

最新评论