浅谈一下Java多线程断点复制
上次写了一个利用 RandoMaccessFile 和 多线程实现的多线程复制，但是没有增加断点复制的功能。这里的断点复制是指：当程序执行中断时（出现错误、断电关机），仍可以从上次复制过程中重新开始（不必从头开始复制）。 多线程复制博客
细节介绍
我这里是使用一个Timer类(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 包下面的类。）
这个类的有很多和时间相关的方法，这里就不介绍了，感兴趣的可以去了解，这里只介绍我们需要使用的一个方法。
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. 为指定的任务安排在指定延迟之后开始的重复固定延迟执行。随后的执行发生在按规定时间间隔的大致间隔。
使用这个方法，按照一个固定的时间间隔记录各个线程的复制进度信息即可。
代码部分
定时任务类
package draGon.local;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
public class RecordTask extends TimerTask {
public static final String filename = "breakPointRecord.txt";
private Timer timer;
private List<FileCopyThread> copyThreads;
private String outputPath;
public RecordTask(Timer timer, List<FileCopyThread> copyThreads, String outputPath) {
this.timer = timer;
this.copyThreads = copyThreads;
this.outputPath = outputPath;
}
@Override
public void run() {
try {
this.breakPointRecord();
} catch (IOException e) {
e.printStackTrace();
}
}
public void breakPointRecord() throws FileNotFoundException, IOException {
int aliveThreadNum = 0; //存活线程数目
//不使用追加方式，这里只需要最新的记录即可。
File recordFile = new File(outputPath, filename);
try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(recordFile))){
//每次记录一个线程的下载位置，但是取出来又需要进行转换，太麻烦了。
//我们直接使用序列化来进行操作，哈哈！
long[] curlen = new long[4];
int index = 0;
for (FileCopyThread copyThread : copyThreads) {
if (copyThread.isAlive()) {
aliveThreadNum++;
}
curlen[index++] = copyThread.getCurlen();
System.out.println(index+" curlen: "+copyThread.getCurlen());
}
//创建 Record 对象，并序列化。
oos.writeObject(new Record(curlen));
}
//当所有的线程都死亡时，关闭计时器，删除记录文件。（所有线程死亡的话，就是文件已经复制完成了！）
if (aliveThreadNum == 0) {
timer.cancel();
recordFile.delete();
}
System.out.println("线程数量： "+aliveThreadNum);
}
}
说明：
if (aliveThreadNum == 0) {
timer.cancel();
recordFile.delete();
}
如果线程都已经结束了，就表示程序已经正常执行结束了。这个时候就删除记录文件。这里这个记录文件是一个标志（flag），如果存在记录文件就表示程序没有正常结束，再次启动时，会进行断点复制。
注意：这里没有考虑复制过程中的 IO 异常，如果线程抛出 IO 异常，那么线程的状态也是结束了。但是考虑，本地文件复制出现 IO 异常的情况还是比较少的，就没有考虑，如果是网络下载的话，这个程序的功能可能就需要进行改进了。
记录信息类
每次需要依次写入各个线程的信息，但是读取出来还需要进行转换，还是感觉过于麻烦了，这里直接利用Java的序列化机制了。 有时候，直接操作对象是很方便的。 注意： 数组的下标表示的就是每个线程的位置。
package dragon.local;
import java.io.Serializable;
public class Record implements Serializable{
private static final long serialVersionUID = 1L;
private long[] curlen;
public Record(long[] curlen) {
this.curlen = curlen;
}
public long[] getCurlen() {
return this.curlen;
}
}
复制线程类
package dragon.local;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
public class FileCopyThread extends Thread {
private int index;
private long position;
private long size;
private File targetFile;
private File outputFile;
private long curlen; //当前下载的长度
public FileCopyThread(int index, long position, long size, File targetFile, File outputFile) {
this.index = index;
this.position = position;
this.size = size;
this.targetFile = targetFile;
this.outputFile = outputFile;
this.curlen = 0L;
}
@Override
public void run() {
try (
BufferedInputStream bis = new BufferedInputStream(new FileInputStream(targetFile));
RandomAccessFile raf = new RandomAccessFile(outputFile, "rw")){
bis.skip(position); //跳过不需要读取的字节数，注意只能先后跳
raf.seek(position); //跳到需要写入的位置，没有这句话，会出错，但是很难改。
int hasRead = 0;
byte[] b = new byte[1024];
while(curlen < size && (hasRead = bis.read(b)) != -1) {
raf.write(b, 0, hasRead);
curlen += (long)hasRead;
//强制停止程序。
// if (curlen > 17_000_000) {
// System.exit(0);
// }
}
System.out.println(index+" "+position+" "+curlen+" "+size);
} catch (IOException e) {
e.printStackTrace();
}
}
public long getCurlen() { //获取当前的进度，用于记录，以便必要时恢复读取进度。
return position+this.curlen;
}
}
这段代码是为了测试断点复制的。如果你想要进行测试，可以将 if 判断中的条件按照你要复制的文件大小进行相应的调整。如果要进行测试，可以先将这段代码的注释取消再执行程序（然后程序退出，这时候文件没有复制完成。），然后再将这段代码注释再次执行程序，文件将会复制成功。
//强制停止程序。
// if (curlen > 17_000_000) {
// System.exit(0);
// }
复制工具类
package dragon.local;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
public class FileCopyUtil {
//设置一个常量，复制线程的数量
private static final int THREAD_NUM = 4;
private FileCopyUtil() {}
public static void transferFile(String targetPath, String outputPath) throws IOException, ClassNotFoundException {
File targetFile = new File(targetPath);
File outputFilePath = new File(outputPath);
if (!targetFile.exists() || targetFile.isDirectory()) { //目标文件不存在，或者是一个文件夹，则抛出异常
throw new FileNotFoundException("目标文件不存在："+targetPath);
}
if (!outputFilePath.exists()) { //如果输出文件夹不存在，将会尝试创建，创建失败，则抛出异常。
if(!outputFilePath.mkdir()) {
throw new FileNotFoundException("无法创建输出文件："+outputPath);
}
}
long len = targetFile.length();
File outputFile = new File(outputFilePath, "copy"+targetFile.getName());
createOutputFile(outputFile, len); //创建输出文件，设置好大小。
//创建计时器 Timer 对象
Timer timer = new Timer();
long[] position = new long[4];
//每一个线程需要复制文件的起点
long size = len / FileCopyUtil.THREAD_NUM + 1; //保存复制线程的集合
List<FileCopyThread> copyThreads = new ArrayList<>();
Record record = getRecord(outputPath);
for (int i = 0; i < FileCopyUtil.THREAD_NUM; i++) {
//如果已经有了 记录文件，就从使用记录数据，否则就是新的下载。
position[i] = record == null ? i*size : record.getCurlen()[i];
FileCopyThread copyThread = new FileCopyThread(i, position[i], size, targetFile, outputFile);
copyThread.start(); //启动复制线程
copyThreads.add(copyThread); //将复制线程添加到集合中。
}
timer.schedule(new RecordTask(timer, copyThreads, outputPath), 0L, 100L); //立即启动计时器，每隔10秒记录一次位置。
System.out.println("开始了！");
}
//创建输出文件，设置好大小。
private static void createOutputFile(File file, long length) throws IOException {
try (
RandomAccessFile raf = new RandomAccessFile(file, "rw")){
raf.setLength(length);
}
}
//获取以及下载的位置
private static Record getRecord(String outputPath) throws FileNotFoundException, IOException, ClassNotFoundException {
File recordFile = new File(outputPath, RecordTask.filename);
if (recordFile.exists()) {
try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(recordFile))){
return (Record) ois.readObject();
}
}
return null;
}
}
说明： 根据复制的目录中，是否存在记录文件来判断是否启动断点复制。
private static Record getRecord(String outputPath) throws FileNotFoundException, IOException, ClassNotFoundException {
File recordFile = new File(outputPath, RecordTask.filename);
if (recordFile.exists()) {
try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(recordFile))){
return (Record) ois.readObject();
}
}
return null;
}
启动断点复制原来其实很简单，就是和复制一样，只不过起始复制位置变成了记录的位置了。
//如果已经有了 记录文件，就从使用记录数据，否则就是新的下载。
position[i] = record == null ? i*size : record.getCurlen()[i];
总结
采用定时记录的方法，感觉也是很不错的，但是似乎又一个问题，当程序正在记录序列化信息的时候，如果出现了错误（导致序列化信息没有写入完整），当反序列化读取的时候，会抛出 EOFException 。不过这种情况很少发生，但是似乎在强制关闭Tomcat的过程中，可能会出现这个问题。（Tomcat的序列化信息很多，IO 时间较长，但是我这里记录的信息很少的，就只是一个 Java 对象而已。）
如果出现了这个异常，解决办法就是删除记录文件，但是因为这个错误就无法使用断点复制的功能了。
关于多线程下载的那部分我没有写，我自己想了好久，没有想出来很好的方法（我对于线程不是很了解），我参考了网上的几个实现（都是将每个线程的记录写入一个单独文件中，但是感觉这样不是很好，我是想写入一个文件中，但是这样又很麻烦。）。我想写一个自己的方法，但是没有想出来，就暂时放弃了。
到此这篇关于浅谈一下Java多线程断点复制的文章就介绍到这了,更多相关Java多线程断点复制内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！
相关文章