java - 使用多线程java编写自己的文件下载器相关的问题

Question

在我现在的公司，我正在做一个关于如何编写文件下载器实用程序的 PoC。我们必须使用套接字编程(TCP/IP)来下载文件。客户端的要求之一是文件(尺寸很大)应该以 block 的形式传输，例如，如果我们有一个 5Mb 大小的文件，那么我们可以有 5 个线程，每个线程传输 1Mb。我编写了一个下载文件的小应用程序。您可以下载eclipe项目

来自 http://www.fileflyer.com/view/QM1JSC0

我的类(class)的简要说明

• FileSender.java:此类提供文件的字节。它有一个方法叫做sendBytesOfFile(long start,long end, long sequenceNo) 给出字节数。

  import java.io.File;
  
  import java.io.IOException;
  
  import java.util.zip.CRC32;
  
  import org.apache.commons.io.FileUtils;
  
  
  public class FileSender {
  
      private static final String FILE_NAME = "C:\\shared\\test.pdf";
  
      public ByteArrayWrapper sendBytesOfFile(long start,long end, long sequenceNo){
          try {
              File file = new File(FILE_NAME);
              byte[] fileBytes = FileUtils.readFileToByteArray(file);
              System.out.println("Size of file is " +fileBytes.length);
              System.out.println();
              System.out.println("Start "+start +" end "+end);
              byte[] bytes = getByteArray(fileBytes, start, end);
              ByteArrayWrapper wrapper = new ByteArrayWrapper(bytes, sequenceNo);
              return wrapper;
          } catch (IOException e) {
              throw new RuntimeException(e);
          }
      }
  
      private byte[] getByteArray(byte[] bytes, long start, long end){
          long arrayLength = end-start;
          System.out.println("Start : "+start +" end : "+end + " Arraylength : "+arrayLength +" length of source array : "+bytes.length);
          byte[] arr = new byte[(int)arrayLength];
  
          for(int i = (int)start, j =0; i < end;i++,j++){
              arr[j] = bytes[i];
          }
          return arr;
      }
  
      public static long fileSize(){
          File file = new File(FILE_NAME);
          return file.length();
      }
  }
  

• FileReceiver.java - 此类接收文件。

简单解释一下这个文件的作用

1. 此类查找要从 Sender 获取的文件的大小
2. 根据文件的大小，它会找到开始和结束位置，直到需要读取字节为止。
3. 它启动 n 个线程，给出每个线程的开始、结束、序列号和所有线程共享的列表。
4. 每个线程读取字节数并创建一个 ByteArrayWrapper。
5. ByteArrayWrapper 对象已添加到列表中
6. 然后我有 while 循环，它基本上确保所有线程都已完成其工作
7. 最后它根据序列号对列表进行排序。
8. 然后将字节连接起来，形成一个完整的字节数组，并将其转换为文件。

文件接收者代码

package com.filedownloader;

import java.io.File;

import java.io.IOException;

import java.util.ArrayList;

import java.util.Collections;

import java.util.Comparator;

import java.util.List;

import java.util.zip.CRC32;

import org.apache.commons.io.FileUtils;


public class FileReceiver {

    public static void main(String[] args) {
        FileReceiver receiver = new FileReceiver();
        receiver.receiveFile();

    }
    public void receiveFile(){
        long startTime = System.currentTimeMillis();
        long numberOfThreads = 10;
        long filesize = FileSender.fileSize();

        System.out.println("File size received "+filesize);
        long start = filesize/numberOfThreads;
        List<ByteArrayWrapper> list = new ArrayList<ByteArrayWrapper>();

        for(long threadCount =0; threadCount<numberOfThreads ;threadCount++){
            FileDownloaderTask task = new FileDownloaderTask(threadCount*start,(threadCount+1)*start,threadCount,list);
            new Thread(task).start();
        }

        while(list.size() != numberOfThreads){
            // this is done so that all the threads should complete their work before processing further.
            //System.out.println("Waiting for threads to complete. List size "+list.size());
        }

        if(list.size() == numberOfThreads){
            System.out.println("All bytes received "+list);
            Collections.sort(list, new Comparator<ByteArrayWrapper>() {
                @Override
                public int compare(ByteArrayWrapper o1, ByteArrayWrapper o2) {

                    long sequence1 = o1.getSequence();
                    long sequence2 = o2.getSequence();
                    if(sequence1 < sequence2){
                        return -1;
                    }else if(sequence1 > sequence2){
                        return 1;
                    }
                    else{
                        return 0;
                    }
                }
            });


            byte[] totalBytes = list.get(0).getBytes();
            byte[] firstArr = null;
            byte[] secondArr = null;
            for(int i = 1;i<list.size();i++){
                firstArr = totalBytes;
                secondArr = list.get(i).getBytes();
                totalBytes = concat(firstArr, secondArr);

            }

            System.out.println(totalBytes.length);
            convertToFile(totalBytes,"c:\\tmp\\test.pdf");

            long endTime = System.currentTimeMillis();
            System.out.println("Total time taken with "+numberOfThreads +" threads is "+(endTime-startTime)+" ms" );

        }
    }

    private byte[] concat(byte[] A, byte[] B) {
       byte[] C= new byte[A.length+B.length];
       System.arraycopy(A, 0, C, 0, A.length);
       System.arraycopy(B, 0, C, A.length, B.length);
       return C;
    }

    private void convertToFile(byte[] totalBytes,String name) {
        try {
            FileUtils.writeByteArrayToFile(new File(name), totalBytes);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

ByteArrayWrapper代码

package com.filedownloader;

import java.io.Serializable;

public class ByteArrayWrapper  implements Serializable{

    private static final long serialVersionUID = 3499562855188457886L;

    private byte[] bytes;
    private long sequence;

    public ByteArrayWrapper(byte[] bytes, long sequenceNo) {
        this.bytes = bytes;
        this.sequence = sequenceNo;
    }

    public byte[] getBytes() {
        return bytes;
    }

    public long getSequence() {
        return sequence;
    }
}

FileDownloaderTask的代码

import java.util.List;


public class FileDownloaderTask implements Runnable {

    private List<ByteArrayWrapper> list;
    private long start;
    private long end;
    private long sequenceNo;

    public FileDownloaderTask(long start,long end,long sequenceNo,List<ByteArrayWrapper> list) {
        this.list = list;
        this.start = start;
        this.end = end;
        this.sequenceNo = sequenceNo;
    }

    @Override
    public void run() {
        ByteArrayWrapper wrapper = new FileSender().sendBytesOfFile(start, end, sequenceNo);
        list.add(wrapper);
    }
}

与此代码相关的问题

1. 使用多线程时文件下载是否会变快？在这段代码中我看不到好处。

2. 我应该如何决定应该创建多少个线程？

3. 他们有这样做的开源库吗

4. 文件接收器接收到的文件有效且未损坏，但校验和(我使用 common-io 的 FileUtils)不匹配。有什么问题吗？

5. 当与大文件(超过 100 Mb)一起使用时，此代码会出现内存不足的情况，即因为创建了字节数组。我该如何避免？

我知道这是一个非常糟糕的代码，但我必须在一天内编写它 -:)。请建议任何其他好方法来做到这一点？

Answer 1

这里有很多问题需要回答。我不会详细介绍所有代码，但我可以给您一些提示。

首先，一些下载加速器所做的确实是使用 HTTP Range header 并行下载文件的各个部分。为什么这有效？ TCP 尝试为每个连接公平分配带宽。因此，如果您从带宽已满的服务器下载文件，那么您可以通过添加更多连接来获得更大份额的带宽。同样的原则也适用于限制传出带宽的服务器，这通常也适用于每个连接(有时考虑到 IP)。

显然，如果每个人都这样做，我们就会留下大量 TCP 连接及其开销，而没有太多带宽来进行实际下载，这就是为什么即使这些下载加速器也只使用 2- 4 个连接。此外，如果您是编写服务器的人，您真的不需要担心这一点，因为您只会减慢自己的速度(通过增加更多开销)。

内存不足:不要使用字节数组，使用(缓冲的)InputStream(或者如果您有时间，请学习如何使用java.nio和字节缓冲区)并在发送文件时读取 block 。 The java tutorials cover all the basics .