Java 다중 스레드 중단점 복사 방법은 무엇입니까?-java지도 시간-php.cn

Details

Timer 클래스(java.util.Timer)를 사용하여 중단점 기능을 구현합니다. 기록된 내용은 각 스레드의 복사본 진행 상황을 기록하는 데 사용됩니다.

Timer 클래스 소개:

백그라운드 스레드에서 향후 실행을 위해 작업을 예약하는 기능입니다. 작업은 일회성 실행으로 예약되거나 정기적인 간격으로 반복 실행되도록 예약될 수 있습니다. 향후 실행을 위한 작업. 작업은 한 번 실행되거나 주기적으로 반복되도록 예약할 수 있습니다.

API 소개에 따르면 이 Timer 클래스는 작업을 한 번만 수행할 수도 있고, 주기적으로 작업을 수행할 수도 있음을 알 수 있습니다. (이 클래스는 javax 패키지의 클래스가 아닌 java.util.Timer 클래스입니다.)

이 클래스에는 여기에서 소개하지 않은 많은 시간 관련 메소드가 있습니다. 여기서만 소개하겠습니다. 방법을 사용해야 합니다.

public void schedule(TimerTask task, long delay, long 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();
}

로그인 후 복사

스레드가 종료되면 프로그램이 정상적으로 종료되었음을 의미합니다.

이때 녹화파일을 삭제하세요. 여기서 기록 파일은 플래그입니다. 기록 파일이 존재한다면 프로그램이 정상적으로 종료되지 않았다는 뜻입니다. 다시 시작하면 중단점 복사가 수행됩니다.

참고: 복사 프로세스 중 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 판단의 조건을 적절하게 조정할 수 있습니다. 테스트하려면 먼저 이 코드의 주석을 해제한 다음 프로그램을 실행하고(그러면 프로그램이 종료되며 이때 파일은 복사되지 않습니다.) 이 코드에 주석을 달고 프로그램을 다시 실행하면 파일이 다음과 같이 됩니다. 성공적으로 복사되었습니다.

//强制停止程序。
//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;
}

로그인 후 복사

복사 시작 위치가 기록 위치가 된다는 점만 빼면 복사와 동일합니다.

아아아아

위 내용은 Java 다중 스레드 중단점 복사 방법은 무엇입니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!