我這裡是使用一個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.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{ /** * 序列化 id */ 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]; /** * 注意,每个线程只是读取一部分数据,不能只以 -1 作为循环结束的条件 * 循环退出条件应该是两个,即写入的字节数大于需要读取的字节数 或者 文件读取结束(最后一个线程读取到文件末尾) */ 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 (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; /** * 设计思路: * 获取目标文件的大小,然后设置复制文件的大小(这样做是有好处的), * 然后使用将文件分为 n 分,使用 n 个线程同时进行复制(这里我将 n 取为 4)。 * * 进一步拓展: * 加强为断点复制功能,即程序中断以后, * 仍然可以继续从上次位置恢复复制,减少不必要的重复开销 * */ public class FileCopyUtil { //设置一个常量,复制线程的数量 private static final int THREAD_NUM = 4; private FileCopyUtil() {} /** * @param targetPath 目标文件的路径 * @param outputPath 复制输出文件的路径 * @throws IOException * @throws ClassNotFoundException * */ 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];
以上是Java多執行緒斷點複製的方法是什麼的詳細內容。更多資訊請關注PHP中文網其他相關文章!