자바 기반 다 중 스 레 드 다운로드 및 정지점 전송 허용

전체 코드:https://github.com/iyuanyb/Downloader
다 중 스 레 드 다운로드 및 정지점 전송 실현 은 HTTP/1.1 에 도 입 된 Range 요청 매개 변 수 를 사용 하여 웹 자원 의 지정 구간 에 접근 할 수 있 는 내용 입 니 다.다 중 스 레 드 와 정지점 의 속전 이 이 루어 졌 지만 완선 되 지 않 은 부분 이 많다.
네 가지 종류 포함:
Downloader:주 클래스,각 하위 스 레 드 에 임 무 를 할당 하고 진 도 를 검사 합 니 다.DownloadFile:다운로드 할 파일 을 표시 합 니 다.파일 에 입력 한 위 치 를 쓸 수 있 도록 RandomAccessFile 류 작업 파일 을 사용 합 니 다.여러 스 레 드 가 같은 파일 을 쓰 려 면 스 레 드 안전 을 확보 해 야 합 니 다.여기 서 getChannel 방법 을 직접 호출 하여 파일 채널 을 가 져 옵 니 다.FileChannel 은 스 레 드 가 안전 합 니 다.DownloadTask:다운로드 한 스 레 드 를 실제 실행 하여[lowerBound,upper Bound]구간 의 데 이 터 를 가 져 옵 니 다.다운로드 과정 에서 이상 이 발생 했 을 때 다른 스 레 드(Atomic Boolean 사용)에 알 리 고 Logger 다운 로드 를 끝 냅 니 다.다운로드 진 도 를 실시 간 으로 기록 하여 전송 할 때 어디서부터 시작 해 야 하 는 지 알 수 있 습 니 다.이 곳 은 비교적 나 쁜 것 같 습 니 다.로 그 를 실시 간 으로 작성 하고 Properties 류 의 load/store 방법 으로 입 출력 을 포맷 하기 위해 매번 열 고 닫 습 니 다.
프레젠테이션:
아무 파일 이나 찾 아서 다운로드:

프로그램 을 강제로 종료 하고 다시 실행 하기:

로그 파일:
정지점 전송 의 관건 은 각 스 레 드 의 다운로드 진 도 를 기록 하 는 것 입 니 다.여기 에는 디 테 일이 많아 서 오래 걸 렸 습 니 다.각 스 레 드 가 요청 한 Range 구간 의 극 객 만 기록 하고 파일 에 데 이 터 를 성공 적 으로 쓸 때마다 다운로드 구간 을 한 번 씩 업데이트 합 니 다.다음은 다운로드 가 완 료 된 로그 내용 입 니 다.

코드:
Downloader.java

package downloader;
 
import java.io.*;
import java.net.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.concurrent.atomic.AtomicBoolean;
 
public class Downloader {
  private static final int DEFAULT_THREAD_COUNT = 4; //       
  private AtomicBoolean canceled; //     ，            ，         
  private DownloadFile file; //        
  private String storageLocation;
  private final int threadCount; //     
  private long fileSize; //     
  private final String url;
  private long beginTime; //     
  private Logger logger;
 
  public Downloader(String url) {
    this(url, DEFAULT_THREAD_COUNT);
  }
 
  public Downloader(String url, int threadCount) {
    this.url = url;
    this.threadCount = threadCount;
    this.canceled = new AtomicBoolean(false);
    this.storageLocation = url.substring(url.lastIndexOf('/')+1);
    this.logger = new Logger(storageLocation + ".log", url, threadCount);
  }
 
  public void start() {
    boolean reStart = Files.exists(Path.of(storageLocation + ".log"));
    if (reStart) {
      logger = new Logger(storageLocation + ".log");
      System.out.printf("*         [   ：%.2fMB]：%s
", logger.getWroteSize() / 1014.0 / 1024, url);
    } else {
      System.out.println("*     ：" + url);
    }
    if (-1 == (this.fileSize = getFileSize()))
      return;
    System.out.printf("*     ：%.2fMB
", fileSize / 1024.0 / 1024);
 
    this.beginTime = System.currentTimeMillis();
    try {
      this.file = new DownloadFile(storageLocation, fileSize, logger);
      if (reStart) {
        file.setWroteSize(logger.getWroteSize());
      }
      //       
      dispatcher(reStart);
      //       
      printDownloadProgress();
    } catch (IOException e) {
      System.err.println("x       [" + e.getMessage() + "]");
    }
  }
 
  /**
   *    ，               
   */
  private void dispatcher(boolean reStart) {
    long blockSize = fileSize / threadCount; //            
    long lowerBound = 0, upperBound = 0;
    long[][] bounds = null;
    int threadID = 0;
    if (reStart) {
      bounds = logger.getBounds();
    }
    for (int i = 0; i < threadCount; i++) {
      if (reStart) {
        threadID = (int)(bounds[i][0]);
        lowerBound = bounds[i][1];
        upperBound = bounds[i][2];
      } else {
        threadID = i;
        lowerBound = i * blockSize;
        // fileSize-1 !!!!! fu.ck，       
        upperBound = (i == threadCount - 1) ? fileSize-1 : lowerBound + blockSize;
      }
      new DownloadTask(url, lowerBound, upperBound, file, canceled, threadID).start();
    }
  }
 
  /**
   *       ，      ，      
   */
  private void printDownloadProgress() {
    long downloadedSize = file.getWroteSize();
    int i = 0;
    long lastSize = 0; //        
    while (!canceled.get() && downloadedSize < fileSize) {
      if (i++ % 4 == 3) { //  3     
        System.out.printf("    ：%.2f%%,    ：%.2fMB，    ：%.2fMB/s
",
            downloadedSize / (double)fileSize * 100 ,
            downloadedSize / 1024.0 / 1024,
            (downloadedSize - lastSize) / 1024.0 / 1024 / 3);
        lastSize = downloadedSize;
        i = 0;
      }
      try {
        Thread.sleep(1000);
      } catch (InterruptedException ignore) {}
      downloadedSize = file.getWroteSize();
    }
    file.close();
    if (canceled.get()) {
      try {
        Files.delete(Path.of(storageLocation));
      } catch (IOException ignore) {
      }
      System.err.println("x     ，     ");
    } else {
      System.out.println("*     ，    "+ (System.currentTimeMillis() - beginTime) / 1000 +" ");
    }
  }
 
  /**
   * @return          
   */
  private long getFileSize() {
    if (fileSize != 0) {
      return fileSize;
    }
    HttpURLConnection conn = null;
    try {
      conn = (HttpURLConnection)new URL(url).openConnection();
      conn.setConnectTimeout(3000);
      conn.setRequestMethod("HEAD");
      conn.connect();
      System.out.println("*        ");
    } catch (MalformedURLException e) {
      throw new RuntimeException("URL  ");
    } catch (IOException e) {
      System.err.println("x        ["+ e.getMessage() +"]");
      return -1;
    }
    return conn.getContentLengthLong();
  }
 
  public static void main(String[] args) throws IOException {
    new Downloader("http://js.xiazaicc.com//down2/ucliulanqi_downcc.zip").start();
  }
}

DownloadTask.java

package downloader;
 
import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.util.concurrent.atomic.AtomicBoolean;
 
class DownloadTask extends Thread {
  private final String url;
  private long lowerBound; //        
  private long upperBound;
  private AtomicBoolean canceled;
  private DownloadFile downloadFile;
  private int threadId;
 
  DownloadTask(String url, long lowerBound, long upperBound, DownloadFile downloadFile,
            AtomicBoolean canceled, int threadID) {
    this.url = url;
    this.lowerBound = lowerBound;
    this.upperBound = upperBound;
    this.canceled = canceled;
    this.downloadFile = downloadFile;
    this.threadId = threadID;
  }
 
  @Override
  public void run() {
    ReadableByteChannel input = null;
    try {
      ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024 * 2); // 2MB
      input = connect();
      System.out.println("* [  " + threadId + "]    ，    ...");
 
      int len;
      while (!canceled.get() && lowerBound <= upperBound) {
        buffer.clear();
        len = input.read(buffer);
        downloadFile.write(lowerBound, buffer, threadId, upperBound);
        lowerBound += len;
      }
      if (!canceled.get()) {
        System.out.println("* [  " + threadId + "]    " + ": " + lowerBound + "-" + upperBound);
      }
    } catch (IOException e) {
      canceled.set(true);
      System.err.println("x [  " + threadId + "]    [" + e.getMessage() + "]，    ");
    } finally {
      if (input != null) {
        try {
          input.close();
        } catch (IOException e) {
          e.printStackTrace();
        }
      }
    }
  }
 
  /**
   *   WEB   ，         
   * @return     
   * @throws IOException       
   */
  private ReadableByteChannel connect() throws IOException {
    HttpURLConnection conn = (HttpURLConnection)new URL(url).openConnection();
    conn.setConnectTimeout(3000);
    conn.setRequestMethod("GET");
    conn.setRequestProperty("Range", "bytes=" + lowerBound + "-" + upperBound);
//    System.out.println("thread_"+ threadId +": " + lowerBound + "-" + upperBound);
    conn.connect();
 
    int statusCode = conn.getResponseCode();
    if (HttpURLConnection.HTTP_PARTIAL != statusCode) {
      conn.disconnect();
      throw new IOException("     ：" + statusCode);
    }
 
    return Channels.newChannel(conn.getInputStream());
  }
}

DownloadFile.java

package downloader;
 
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.atomic.AtomicLong;
 
class DownloadFile {
  private final RandomAccessFile file;
  private final FileChannel channel; //      
  private AtomicLong wroteSize; //       
  private Logger logger;
 
  DownloadFile(String fileName, long fileSize, Logger logger) throws IOException {
    this.wroteSize = new AtomicLong(0);
    this.logger = logger;
    this.file = new RandomAccessFile(fileName, "rw");
    file.setLength(fileSize);
    channel = file.getChannel();
  }
 
  /**
   *    
   * @param offset     
   * @param buffer   
   * @throws IOException        
   */
  void write(long offset, ByteBuffer buffer, int threadID, long upperBound) throws IOException {
    buffer.flip();
    int length = buffer.limit();
    while (buffer.hasRemaining()) {
      channel.write(buffer, offset);
    }
    wroteSize.addAndGet(length);
    logger.updateLog(threadID, length, offset + length, upperBound); //     
  }
 
  /**
   * @return         ，            ，      
   */
  long getWroteSize() {
    return wroteSize.get();
  }
 
  //        
  void setWroteSize(long wroteSize) {
    this.wroteSize.set(wroteSize);
  }
 
  void close() {
    try {
      file.close();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
}

Logger.java

package downloader;
 
import java.io.*;
import java.util.Properties;
 
class Logger {
  private String logFileName; //         
  private Properties log;
 
   /**
   *        ，       
   * @param logFileName
   */
  Logger(String logFileName) {
    this.logFileName = logFileName;
    log = new Properties();
    FileInputStream fin = null;
    try {
      log.load(new FileInputStream(logFileName));
    } catch (IOException ignore) {
    } finally {
      try {
        fin.close();
      } catch (Exception ignore) {}
    }
  }
 
  Logger(String logFileName, String url, int threadCount) {
    this.logFileName = logFileName;
    this.log = new Properties();
    log.put("url", url);
    log.put("wroteSize", "0");
    log.put("threadCount", String.valueOf(threadCount));
    for (int i = 0; i < threadCount; i++) {
      log.put("thread_" + i, "0-0");
    }
  }
 
 
  synchronized void updateLog(int threadID, long length, long lowerBound, long upperBound) {
    log.put("thread_"+threadID, lowerBound + "-" + upperBound);
    log.put("wroteSize", String.valueOf(length + Long.parseLong(log.getProperty("wroteSize"))));
 
    FileOutputStream file = null;
    try {
      file = new FileOutputStream(logFileName); //          
      log.store(file, null);
    } catch (IOException e) {
      e.printStackTrace();
    } finally {
      if (file != null) {
        try {
          file.close();
        } catch (IOException e) {
          e.printStackTrace();
        }
      }
    }
  }
 
  /**
   *       
   *   ret[i][0] = threadID, ret[i][1] = lowerBoundID, ret[i][2] = upperBoundID
   * @return
   */
  long[][] getBounds() {
    long[][] bounds = new long[Integer.parseInt(log.get("threadCount").toString())][3];
    int[] index = {0};
    log.forEach((k, v) -> {
      String key = k.toString();
      if (key.startsWith("thread_")) {
        String[] interval = v.toString().split("-");
        bounds[index[0]][0] = Long.parseLong(key.substring(key.indexOf("_") + 1));
        bounds[index[0]][1] = Long.parseLong(interval[0]);
        bounds[index[0]++][2] = Long.parseLong(interval[1]);
      }
    });
    return bounds;
  }
  long getWroteSize() {
    return Long.parseLong(log.getProperty("wroteSize"));
  }
}

이상 이 바로 본 고의 모든 내용 입 니 다.여러분 의 학습 에 도움 이 되 고 저 희 를 많이 응원 해 주 셨 으 면 좋 겠 습 니 다.