Java多线程断点复制的方法是什么

其他教程   发布日期:2025年01月18日   浏览次数:174

这篇文章主要介绍了Java多线程断点复制的方法是什么的相关知识,内容详细易懂,操作简单快捷,具有一定借鉴价值,相信大家阅读完这篇Java多线程断点复制的方法是什么文章都会有所收获,下面我们一起来看看吧。

细节介绍

我这里是使用一个Timer类(

  1. java.util.Timer
)来实现断点功能的,就是使用这个类,每隔一段时间进行一次记录,记录的内容是每个线程复制的进度。

Timer 类的介绍:

A facility for threads to schedule tasks for future execution in a background thread. Tasks may be scheduled for one-time execution, or for repeated execution at regular intervals. 线程在后台线程中调度任务以供将来执行的工具。任务可以安排为一次性执行,也可以安排为定期重复执行。

根据 API 中的介绍可以看出,这个 Timer 类可以只执行一次任务,也可以周期性地执行任务。(注意这个类是 java.util.Timer 类,不是 javax 包下面的类。)

这个类的有很多和时间相关的方法,这里就不介绍了,感兴趣的可以去了解,这里只介绍我们需要使用的一个方法。

  1. public void schedule(TimerTask task, long delay, long period)

Schedules the specified task for repeated fixed-delay execution beginning after the specified delay. Subsequent executions take place at approximately regular intervals separated by the specified period. 为指定的任务安排在指定延迟之后开始的重复固定延迟执行。随后的执行发生在按规定时间间隔的大致间隔。

使用这个方法,按照一个固定的时间间隔记录各个线程的复制进度信息即可。

代码部分

定时任务类

  1. package dragon.local;
  2. import java.io.File;
  3. import java.io.FileNotFoundException;
  4. import java.io.FileOutputStream;
  5. import java.io.IOException;
  6. import java.io.ObjectOutputStream;
  7. import java.util.List;
  8. import java.util.Timer;
  9. import java.util.TimerTask;
  10. public class RecordTask extends TimerTask {
  11. public static final String filename = "breakPointRecord.txt";
  12. private Timer timer;
  13. private List<FileCopyThread> copyThreads;
  14. private String outputPath;
  15. public RecordTask(Timer timer, List<FileCopyThread> copyThreads, String outputPath) {
  16. this.timer = timer;
  17. this.copyThreads = copyThreads;
  18. this.outputPath = outputPath;
  19. }
  20. @Override
  21. public void run() {
  22. try {
  23. this.breakPointRecord();
  24. } catch (IOException e) {
  25. e.printStackTrace();
  26. }
  27. }
  28. public void breakPointRecord() throws FileNotFoundException, IOException {
  29. int aliveThreadNum = 0; //存活线程数目
  30. //不使用追加方式,这里只需要最新的记录即可。
  31. File recordFile = new File(outputPath, filename);
  32. try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(recordFile))){
  33. //每次记录一个线程的下载位置,但是取出来又需要进行转换,太麻烦了。
  34. //我们直接使用序列化来进行操作,哈哈!
  35. long[] curlen = new long[4];
  36. int index = 0;
  37. for (FileCopyThread copyThread : copyThreads) {
  38. if (copyThread.isAlive()) {
  39. aliveThreadNum++;
  40. }
  41. curlen[index++] = copyThread.getCurlen();
  42. System.out.println(index+" curlen: "+copyThread.getCurlen());
  43. }
  44. //创建 Record 对象,并序列化。
  45. oos.writeObject(new Record(curlen));
  46. }
  47. //当所有的线程都死亡时,关闭计时器,删除记录文件。(所有线程死亡的话,就是文件已经复制完成了!)
  48. if (aliveThreadNum == 0) {
  49. timer.cancel();
  50. recordFile.delete();
  51. }
  52. System.out.println("线程数量: "+aliveThreadNum);
  53. }
  54. }

说明:

  1. if (aliveThreadNum == 0) {
  2. timer.cancel();
  3. recordFile.delete();
  4. }

如果线程都已经结束了,就表示程序已经正常执行结束了。这个时候就删除记录文件。这里这个记录文件是一个标志(flag),如果存在记录文件就表示程序没有正常结束,再次启动时,会进行断点复制

注意:这里没有考虑复制过程中的 IO 异常,如果线程抛出 IO 异常,那么线程的状态也是结束了。但是考虑,本地文件复制出现 IO 异常的情况还是比较少的,就没有考虑,如果是网络下载的话,这个程序的功能可能就需要进行改进了。

记录信息类

每次需要依次写入各个线程的信息,但是读取出来还需要进行转换,还是感觉过于麻烦了,这里直接利用Java的序列化机制了。 有时候,直接操作对象是很方便的。 注意: 数组的下标表示的就是每个线程的位置。

  1. package dragon.local;
  2. import java.io.Serializable;
  3. public class Record implements Serializable{
  4. /**
  5. * 序列化 id
  6. */
  7. private static final long serialVersionUID = 1L;
  8. private long[] curlen;
  9. public Record(long[] curlen) {
  10. this.curlen = curlen;
  11. }
  12. public long[] getCurlen() {
  13. return this.curlen;
  14. }
  15. }

复制线程类

  1. package dragon.local;
  2. import java.io.BufferedInputStream;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.IOException;
  6. import java.io.RandomAccessFile;
  7. public class FileCopyThread extends Thread {
  8. private int index;
  9. private long position;
  10. private long size;
  11. private File targetFile;
  12. private File outputFile;
  13. private long curlen; //当前下载的长度
  14. public FileCopyThread(int index, long position, long size, File targetFile, File outputFile) {
  15. this.index = index;
  16. this.position = position;
  17. this.size = size;
  18. this.targetFile = targetFile;
  19. this.outputFile = outputFile;
  20. this.curlen = 0L;
  21. }
  22. @Override
  23. public void run() {
  24. try (
  25. BufferedInputStream bis = new BufferedInputStream(new FileInputStream(targetFile));
  26. RandomAccessFile raf = new RandomAccessFile(outputFile, "rw")){
  27. bis.skip(position); //跳过不需要读取的字节数,注意只能先后跳
  28. raf.seek(position); //跳到需要写入的位置,没有这句话,会出错,但是很难改。
  29. int hasRead = 0;
  30. byte[] b = new byte[1024];
  31. /**
  32. * 注意,每个线程只是读取一部分数据,不能只以 -1 作为循环结束的条件
  33. * 循环退出条件应该是两个,即写入的字节数大于需要读取的字节数 或者 文件读取结束(最后一个线程读取到文件末尾)
  34. */
  35. while(curlen < size && (hasRead = bis.read(b)) != -1) {
  36. raf.write(b, 0, hasRead);
  37. curlen += (long)hasRead;
  38. //强制停止程序。
  39. // if (curlen > 17_000_000) {
  40. // System.exit(0);
  41. // }
  42. }
  43. System.out.println(index+" "+position+" "+curlen+" "+size);
  44. } catch (IOException e) {
  45. e.printStackTrace();
  46. }
  47. }
  48. public long getCurlen() { //获取当前的进度,用于记录,以便必要时恢复读取进度。
  49. return position+this.curlen;
  50. }
  51. }

这段代码是为了测试断点复制的。如果你想要进行测试,可以将 if 判断中的条件按照你要复制的文件大小进行相应的调整。如果要进行测试,可以先将这段代码的注释取消再执行程序(然后程序退出,这时候文件没有复制完成。),然后再将这段代码注释再次执行程序,文件将会复制成功。

  1. //强制停止程序。
  2. // if (curlen > 17_000_000) {
  3. // System.exit(0);
  4. // }

复制工具类

  1. package dragon.local;
  2. import java.io.File;
  3. import java.io.FileInputStream;
  4. import java.io.FileNotFoundException;
  5. import java.io.IOException;
  6. import java.io.ObjectInputStream;
  7. import java.io.RandomAccessFile;
  8. import java.util.ArrayList;
  9. import java.util.List;
  10. import java.util.Timer;
  11. /**
  12. * 设计思路:
  13. * 获取目标文件的大小,然后设置复制文件的大小(这样做是有好处的),
  14. * 然后使用将文件分为 n 分,使用 n 个线程同时进行复制(这里我将 n 取为 4)。
  15. *
  16. * 进一步拓展:
  17. * 加强为断点复制功能,即程序中断以后,
  18. * 仍然可以继续从上次位置恢复复制,减少不必要的重复开销
  19. * */
  20. public class FileCopyUtil {
  21. //设置一个常量,复制线程的数量
  22. private static final int THREAD_NUM = 4;
  23. private FileCopyUtil() {}
  24. /**
  25. * @param targetPath 目标文件的路径
  26. * @param outputPath 复制输出文件的路径
  27. * @throws IOException
  28. * @throws ClassNotFoundException
  29. * */
  30. public static void transferFile(String targetPath, String outputPath) throws IOException, ClassNotFoundException {
  31. File targetFile = new File(targetPath);
  32. File outputFilePath = new File(outputPath);
  33. if (!targetFile.exists() || targetFile.isDirectory()) { //目标文件不存在,或者是一个文件夹,则抛出异常
  34. throw new FileNotFoundException("目标文件不存在:"+targetPath);
  35. }
  36. if (!outputFilePath.exists()) { //如果输出文件夹不存在,将会尝试创建,创建失败,则抛出异常。
  37. if(!outputFilePath.mkdir()) {
  38. throw new FileNotFoundException("无法创建输出文件:"+outputPath);
  39. }
  40. }
  41. long len = targetFile.length();
  42. File outputFile = new File(outputFilePath, "copy"+targetFile.getName());
  43. createOutputFile(outputFile, len); //创建输出文件,设置好大小。
  44. //创建计时器 Timer 对象
  45. Timer timer = new Timer();
  46. long[] position = new long[4];
  47. //每一个线程需要复制文件的起点
  48. long size = len / FileCopyUtil.THREAD_NUM + 1; //保存复制线程的集合
  49. List<FileCopyThread> copyThreads = new ArrayList<>();
  50. Record record = getRecord(outputPath);
  51. for (int i = 0; i < FileCopyUtil.THREAD_NUM; i++) {
  52. //如果已经有了 记录文件,就从使用记录数据,否则就是新的下载。
  53. position[i] = record == null ? i*size : record.getCurlen()[i];
  54. FileCopyThread copyThread = new FileCopyThread(i, position[i], size, targetFile, outputFile);
  55. copyThread.start(); //启动复制线程
  56. copyThreads.add(copyThread); //将复制线程添加到集合中。
  57. }
  58. timer.schedule(new RecordTask(timer, copyThreads, outputPath), 0L, 100L); //立即启动计时器,每隔10秒记录一次位置。
  59. System.out.println("开始了!");
  60. }
  61. //创建输出文件,设置好大小。
  62. private static void createOutputFile(File file, long length) throws IOException {
  63. try (
  64. RandomAccessFile raf = new RandomAccessFile(file, "rw")){
  65. raf.setLength(length);
  66. }
  67. }
  68. //获取以及下载的位置
  69. private static Record getRecord(String outputPath) throws FileNotFoundException, IOException, ClassNotFoundException {
  70. File recordFile = new File(outputPath, RecordTask.filename);
  71. if (recordFile.exists()) {
  72. try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(recordFile))){
  73. return (Record) ois.readObject();
  74. }
  75. }
  76. return null;
  77. }
  78. }

说明: 根据复制的目录中,是否存在记录文件来判断是否启动断点复制。

  1. private static Record getRecord(String outputPath) throws FileNotFoundException, IOException, ClassNotFoundException {
  2. File recordFile = new File(outputPath, RecordTask.filename);
  3. if (recordFile.exists()) {
  4. try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(recordFile))){
  5. return (Record) ois.readObject();
  6. }
  7. }
  8. return null;
  9. }

启动断点复制原来其实很简单,就是和复制一样,只不过起始复制位置变成了记录的位置了。

  1. //如果已经有了 记录文件,就从使用记录数据,否则就是新的下载。
  2. position[i] = record == null ? i*size : record.getCurlen()[i];

以上就是Java多线程断点复制的方法是什么的详细内容,更多关于Java多线程断点复制的方法是什么的资料请关注九品源码其它相关文章!