淺談一下Java多線程斷點復制

更新時間：2023年04月11日 10:12:20 作者：CrazyDragon_King

這篇文章主要介紹了淺談一下Java多線程斷點復制,當程序執(zhí)行中斷時（出現(xiàn)錯誤、斷電關(guān)機），仍可以從上次復制過程中重新開始（不必從頭開始復制）,需要的朋友可以參考下

上次寫了一個利用 RandomAccessFile 和多線程實現(xiàn)的多線程復制，但是沒有增加斷點復制的功能。這里的斷點復制是指：當程序執(zhí)行中斷時（出現(xiàn)錯誤、斷電關(guān)機），仍可以從上次復制過程中重新開始（不必從頭開始復制）。多線程復制博客

細節(jié)介紹

我這里是使用一個Timer類(java.util.Timer)來實現(xiàn)斷點功能的，就是使用這個類，每隔一段時間進行一次記錄，記錄的內(nèi)容是每個線程復制的進度。

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. 線程在后臺線程中調(diào)度任務以供將來執(zhí)行的工具。任務可以安排為一次性執(zhí)行，也可以安排為定期重復執(zhí)行。

根據(jù) API 中的介紹可以看出，這個 Timer 類可以只執(zhí)行一次任務，也可以周期性地執(zhí)行任務。（注意這個類是 java.util.Timer 類，不是 javax 包下面的類。）

這個類的有很多和時間相關(guān)的方法，這里就不介紹了，感興趣的可以去了解，這里只介紹我們需要使用的一個方法。

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. 為指定的任務安排在指定延遲之后開始的重復固定延遲執(zhí)行。隨后的執(zhí)行發(fā)生在按規(guī)定時間間隔的大致間隔。

使用這個方法，按照一個固定的時間間隔記錄各個線程的復制進度信息即可。

代碼部分

定時任務類

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;  //存活線程數(shù)目
		//不使用追加方式，這里只需要最新的記錄即可。
		File recordFile = new File(outputPath, filename);
		try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(recordFile))){
			//每次記錄一個線程的下載位置，但是取出來又需要進行轉(zhuǎn)換，太麻煩了。
			//我們直接使用序列化來進行操作，哈哈！
			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());
			}
			//創(chuàng)建 Record 對象，并序列化。
			oos.writeObject(new Record(curlen));
		}
		//當所有的線程都死亡時，關(guān)閉計時器，刪除記錄文件。（所有線程死亡的話，就是文件已經(jīng)復制完成了?。?
		if (aliveThreadNum == 0) {
			timer.cancel();
			recordFile.delete();
		}
		System.out.println("線程數(shù)量： "+aliveThreadNum);
	}
}

說明：

if (aliveThreadNum == 0) {
	timer.cancel();
	recordFile.delete();
}

如果線程都已經(jīng)結(jié)束了，就表示程序已經(jīng)正常執(zhí)行結(jié)束了。這個時候就刪除記錄文件。這里這個記錄文件是一個標志（flag），如果存在記錄文件就表示程序沒有正常結(jié)束，再次啟動時，會進行斷點復制。

注意：這里沒有考慮復制過程中的 IO 異常，如果線程拋出 IO 異常，那么線程的狀態(tài)也是結(jié)束了。但是考慮，本地文件復制出現(xiàn) IO 異常的情況還是比較少的，就沒有考慮，如果是網(wǎng)絡下載的話，這個程序的功能可能就需要進行改進了。

記錄信息類

每次需要依次寫入各個線程的信息，但是讀取出來還需要進行轉(zhuǎn)換，還是感覺過于麻煩了，這里直接利用Java的序列化機制了。 有時候，直接操作對象是很方便的。 注意： 數(shù)組的下標表示的就是每個線程的位置。

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);  //跳過不需要讀取的字節(jié)數(shù)，注意只能先后跳
			raf.seek(position);  //跳到需要寫入的位置，沒有這句話，會出錯，但是很難改。
			int hasRead = 0;
			byte[] b = new byte[1024];
			/**
			 * 注意，每個線程只是讀取一部分數(shù)據(jù)，不能只以 -1 作為循環(huán)結(jié)束的條件
			 * 循環(huán)退出條件應該是兩個，即寫入的字節(jié)數(shù)大于需要讀取的字節(jié)數(shù) 或者 文件讀取結(jié)束（最后一個線程讀取到文件末尾）
			 */
			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 判斷中的條件按照你要復制的文件大小進行相應的調(diào)整。如果要進行測試，可以先將這段代碼的注釋取消再執(zhí)行程序（然后程序退出，這時候文件沒有復制完成。），然后再將這段代碼注釋再次執(zhí)行程序，文件將會復制成功。

				//強制停止程序。
//				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）。
 * 
 * 進一步拓展：
 * 加強為斷點復制功能，即程序中斷以后，
 * 仍然可以繼續(xù)從上次位置恢復復制，減少不必要的重復開銷
 * */

public class FileCopyUtil {
	//設置一個常量，復制線程的數(shù)量
	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()) {     //如果輸出文件夾不存在，將會嘗試創(chuàng)建，創(chuàng)建失敗，則拋出異常。
			if(!outputFilePath.mkdir()) {
				throw new FileNotFoundException("無法創(chuàng)建輸出文件："+outputPath);
			}
		}
		
		long len = targetFile.length();
		
		File outputFile = new File(outputFilePath, "copy"+targetFile.getName());
		createOutputFile(outputFile, len);    //創(chuàng)建輸出文件，設置好大小。
		
		//創(chuàng)建計時器 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++) {
			//如果已經(jīng)有了 記錄文件，就從使用記錄數(shù)據(jù)，否則就是新的下載。
			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("開始了！");
	}
	
	//創(chuàng)建輸出文件，設置好大小。
	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;
	}
}

說明： 根據(jù)復制的目錄中，是否存在記錄文件來判斷是否啟動斷點復制。

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;
	}

啟動斷點復制原來其實很簡單，就是和復制一樣，只不過起始復制位置變成了記錄的位置了。

//如果已經(jīng)有了 記錄文件，就從使用記錄數(shù)據(jù)，否則就是新的下載。
position[i] = record == null ? i*size : record.getCurlen()[i];

總結(jié)

采用定時記錄的方法，感覺也是很不錯的，但是似乎又一個問題，當程序正在記錄序列化信息的時候，如果出現(xiàn)了錯誤（導致序列化信息沒有寫入完整），當反序列化讀取的時候，會拋出 EOFException 。不過這種情況很少發(fā)生，但是似乎在強制關(guān)閉tomcat的過程中，可能會出現(xiàn)這個問題。（Tomcat的序列化信息很多，IO 時間較長，但是我這里記錄的信息很少的，就只是一個 Java 對象而已。）

如果出現(xiàn)了這個異常，解決辦法就是刪除記錄文件，但是因為這個錯誤就無法使用斷點復制的功能了。

關(guān)于多線程下載的那部分我沒有寫，我自己想了好久，沒有想出來很好的方法（我對于線程不是很了解），我參考了網(wǎng)上的幾個實現(xiàn)（都是將每個線程的記錄寫入一個單獨文件中，但是感覺這樣不是很好，我是想寫入一個文件中，但是這樣又很麻煩。）。我想寫一個自己的方法，但是沒有想出來，就暫時放棄了。

到此這篇關(guān)于淺談一下Java多線程斷點復制的文章就介紹到這了,更多相關(guān)Java多線程斷點復制內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: