限制输入流

Question

我对 Java 还是很陌生。我编写了一个 Swing 应用程序，它通过 Internet 下载特定文件。

我想做的是限制InputStream 每秒读取n（假设为10240 字节，但这可能会改变）字节每秒。

读取整个文件后，应将文件保存到本地目录，例如C:\Downloads\

我猜我应该创建一个扩展至 InputStream 的类并覆盖它的方法，但我不确定如何以及使用哪些方法来实现我的目标。

我使用this 示例进行 HTTP 下载，this 示例用于 FTP。

HTTP 示例；

package net.codejava.networking;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.URL;

/**
 * A utility that downloads a file from a URL.
 * @author www.codejava.net
 *
 */
public class HttpDownloadUtility {
    private static final int BUFFER_SIZE = 4096;

    /**
     * Downloads a file from a URL
     * @param fileURL HTTP URL of the file to be downloaded
     * @param saveDir path of the directory to save the file
     * @throws IOException
     */
    public static void downloadFile(String fileURL, String saveDir)
            throws IOException {
        URL url = new URL(fileURL);
        HttpURLConnection httpConn = (HttpURLConnection) url.openConnection();
        int responseCode = httpConn.getResponseCode();

        // always check HTTP response code first
        if (responseCode == HttpURLConnection.HTTP_OK) {
            String fileName = "";
            String disposition = httpConn.getHeaderField("Content-Disposition");
            String contentType = httpConn.getContentType();
            int contentLength = httpConn.getContentLength();

            if (disposition != null) {
                // extracts file name from header field
                int index = disposition.indexOf("filename=");
                if (index > 0) {
                    fileName = disposition.substring(index + 10,
                            disposition.length() - 1);
                }
            } else {
                // extracts file name from URL
                fileName = fileURL.substring(fileURL.lastIndexOf("/") + 1,
                        fileURL.length());
            }

            System.out.println("Content-Type = " + contentType);
            System.out.println("Content-Disposition = " + disposition);
            System.out.println("Content-Length = " + contentLength);
            System.out.println("fileName = " + fileName);

            // opens input stream from the HTTP connection
            InputStream inputStream = httpConn.getInputStream();
            String saveFilePath = saveDir + File.separator + fileName;

            // opens an output stream to save into file
            FileOutputStream outputStream = new FileOutputStream(saveFilePath);

            int bytesRead = -1;
            byte[] buffer = new byte[BUFFER_SIZE];
            while ((bytesRead = inputStream.read(buffer)) != -1) {
                outputStream.write(buffer, 0, bytesRead);
            }

            outputStream.close();
            inputStream.close();

            System.out.println("File downloaded");
        } else {
            System.out.println("No file to download. Server replied HTTP code: " + responseCode);
        }
        httpConn.disconnect();
    }
}

FTP 示例；

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;

/**
 * A program demonstrates how to upload files from local computer to a remote
 * FTP server using Apache Commons Net API.
 * @author www.codejava.net
 */
public class FTPDownloadFileDemo {

    public static void main(String[] args) {
        String server = "www.myserver.com";
        int port = 21;
        String user = "user";
        String pass = "pass";

        FTPClient ftpClient = new FTPClient();
        try {

            ftpClient.connect(server, port);
            ftpClient.login(user, pass);
            ftpClient.enterLocalPassiveMode();
            ftpClient.setFileType(FTP.BINARY_FILE_TYPE);

            // APPROACH #1: using retrieveFile(String, OutputStream)
            String remoteFile1 = "/test/video.mp4";
            File downloadFile1 = new File("D:/Downloads/video.mp4");
            OutputStream outputStream1 = new BufferedOutputStream(new FileOutputStream(downloadFile1));
            boolean success = ftpClient.retrieveFile(remoteFile1, outputStream1);
            outputStream1.close();

            if (success) {
                System.out.println("File #1 has been downloaded successfully.");
            }

            // APPROACH #2: using InputStream retrieveFileStream(String)
            String remoteFile2 = "/test/song.mp3";
            File downloadFile2 = new File("D:/Downloads/song.mp3");
            OutputStream outputStream2 = new BufferedOutputStream(new FileOutputStream(downloadFile2));
            InputStream inputStream = ftpClient.retrieveFileStream(remoteFile2);
            byte[] bytesArray = new byte[4096];
            int bytesRead = -1;
            while ((bytesRead = inputStream.read(bytesArray)) != -1) {
                outputStream2.write(bytesArray, 0, bytesRead);
            }

            success = ftpClient.completePendingCommand();
            if (success) {
                System.out.println("File #2 has been downloaded successfully.");
            }
            outputStream2.close();
            inputStream.close();

        } catch (IOException ex) {
            System.out.println("Error: " + ex.getMessage());
            ex.printStackTrace();
        } finally {
            try {
                if (ftpClient.isConnected()) {
                    ftpClient.logout();
                    ftpClient.disconnect();
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
}

我想要限制这两个输入流的下载速率，这样在从远程位置获取这些文件时下载速度会受到限制。

如果您能给我看一个带有简短解释的示例，我将不胜感激。

Answer 1

您实际上可以扩展java.util.TimerTask 而不是扩展InputStream。然后使用java.util.Timer，您可以安排一个每秒执行一次的任务，直到整个文件下载完毕。因此，您可以简单地定义一秒钟内要下载的字节数。有时下载可能低于您提供的限制（由于网络延迟等），但这不是冲突，因为在每种情况下，您的下载速率都会低于您提供的限制。此外，即使在下载过程中，您也可以随时更改下载速率。 HTTP/HTTPS 实现示例如下：

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.text.NumberFormat;
import java.util.Timer;
import java.util.TimerTask;

public class DownloaderWithLimit extends TimerTask
{
    private InputStream inputStream;
    private OutputStream outputStream;

    private int bytePerSecondLimit;
    private Timer timer;
    private float contentLength;
    private float downloadedLength;

    public DownloaderWithLimit (String urlToDownload, String destFileFullPath, int bytePerSecondLimit) throws IOException
    {
        this.bytePerSecondLimit = bytePerSecondLimit;
        //
        inputStream = createInputStreamFromUrl ( urlToDownload );
        outputStream = new FileOutputStream ( new File ( destFileFullPath ) );
    }

    public void start(){
        timer = new Timer();
        timer.scheduleAtFixedRate(this, 1000, 1000);
    }

    public void run () 
    {
        try
        {
            byte[] buffer = new byte[bytePerSecondLimit];
            int bytesRead = inputStream.read(buffer);
            if(bytesRead != -1){
                outputStream.write(buffer, 0, bytesRead);
            } else {
                outputStream.close();
                inputStream.close();
                timer.cancel();
            }
            downloadedLength += bytesRead;
            System.out.println (bytesRead + " bytes per second. "+ NumberFormat.getPercentInstance ().format ( downloadedLength/contentLength )+" completed...");
        }
        catch ( IOException e )
        {
            e.printStackTrace();
            throw new RuntimeException ( "Error During Download..." , e.getCause () );
        }
    }

    private InputStream createInputStreamFromUrl ( String fileUrl ) throws IOException
    {
        URL url = new URL ( fileUrl );
        HttpURLConnection httpConn = (HttpURLConnection) url.openConnection ();
        int responseCode = httpConn.getResponseCode ();

        // always check HTTP response code first
        if ( responseCode == HttpURLConnection.HTTP_OK )
        {
            String fileName = "";
            String disposition = httpConn
                    .getHeaderField ( "Content-Disposition" );
            String contentType = httpConn.getContentType ();
            contentLength = httpConn.getContentLength ();

            if ( disposition != null )
            {
                // extracts file name from header field
                int index = disposition.indexOf ( "filename=" );
                if ( index > 0 )
                {
                    fileName = disposition.substring ( index + 10 , disposition.length () - 1 );
                }
            }
            else
            {
                // extracts file name from URL
                fileName = fileUrl.substring ( fileUrl.lastIndexOf ( "/" ) + 1 , fileUrl.length () );
            }

            System.out.println ( "Content-Type = " + contentType );
            System.out.println ( "Content-Disposition = " + disposition );
            System.out.println ( "Content-Length = " + contentLength );
            System.out.println ( "fileName = " + fileName );

            // opens input stream from the HTTP connection
            InputStream inputStream = httpConn.getInputStream ();

            return inputStream;
        }

        return null;
    }

    /**
     * @return the bytePerSecondLimit
     */
    public int getBytePerSecondLimit ()
    {
        return bytePerSecondLimit;
    }

    /**
     * @param bytePerSecondLimit the bytePerSecondLimit to set
     */
    public void setBytePerSecondLimit ( int bytePerSecondLimit )
    {
        this.bytePerSecondLimit = bytePerSecondLimit;
    }

    public static void main ( String[] args ) throws IOException
    {
        DownloaderWithLimit d = new DownloaderWithLimit ( "https://download.mozilla.org/?product=firefox-46.0.1-SSL&os=win64&lang=en-US" , "c:/firefox-46.0.1_x64.exe" , 10240);//10Kb/s
        d.start ();
    }
}

而 FTP 的速率限制示例如下：

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.text.NumberFormat;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;


public class FtpDownloaderWithLimit extends TimerTask
{
    private InputStream inputStream;
    private OutputStream outputStream;

    private int bytePerSecondLimit;
    private Timer timer;
    private float contentLength;
    private float downloadedLength;
    private FTPClient ftpClient;

    public FtpDownloaderWithLimit (String ftpServer, int port, String username, String password, String srcFileRelativePath, String destFileFullPath, int bytePerSecondLimit) throws IOException
    {
        this.bytePerSecondLimit = bytePerSecondLimit;
        //
        inputStream = createInputStreamFromUrl ( ftpServer, port, username, password, srcFileRelativePath );
        outputStream = new FileOutputStream ( new File ( destFileFullPath ) );
    }

    public void start(){
        timer = new Timer();
        timer.scheduleAtFixedRate(this, 1000, 1000);
    }

    public void run () 
    {
        try
        {
            byte[] buffer = new byte[bytePerSecondLimit];
            int bytesRead = inputStream.read(buffer);
            if(bytesRead != -1){
                outputStream.write(buffer, 0, bytesRead);
            } else {
                boolean success = ftpClient.completePendingCommand();
                if (success) {
                    System.out.println("File #2 has been downloaded successfully.");
                }
                outputStream.close();
                inputStream.close();
                timer.cancel();
            }
            downloadedLength += bytesRead;
            System.out.println (bytesRead + " bytes per second. "+ NumberFormat.getPercentInstance ().format ( downloadedLength/contentLength )+" completed...");
        }
        catch ( IOException e )
        {
            e.printStackTrace();
            throw new RuntimeException ( "Error During Download..." , e.getCause () );
        }
    }

    private InputStream createInputStreamFromUrl(String ftpServer, int port,
            String username, String password, String srcFileRelativePath) throws IOException{

        ftpClient = new FTPClient();
        ftpClient.connect(ftpServer, port);
        ftpClient.login(username, password);
        ftpClient.enterLocalPassiveMode();
        ftpClient.setFileType(FTP.BINARY_FILE_TYPE);

        InputStream inputStream = ftpClient.retrieveFileStream(srcFileRelativePath);

        return inputStream;
    }

    /**
     * @return the bytePerSecondLimit
     */
    public int getBytePerSecondLimit ()
    {
        return bytePerSecondLimit;
    }

    /**
     * @param bytePerSecondLimit the bytePerSecondLimit to set
     */
    public void setBytePerSecondLimit ( int bytePerSecondLimit )
    {
        this.bytePerSecondLimit = bytePerSecondLimit;
    }

    public static void main ( String[] args ) throws IOException
    {
        FtpDownloaderWithLimit d = new FtpDownloaderWithLimit ( "www.myserver.com" , 9111 /*sample port*/, "USERNAME", "PASSWORD", "/My/File/To/Downlaod.xxx", "c:/your-path-to-dest-file" , 10240);//10Kb/s
        d.start ();
    }

}

希望这会有所帮助。

Answer 2

您可以尝试任何 InputStream.len 的 read(byte[] b, int off, int len) 是读取的最大字节数。 off 是流开始插入的位置，b 是流写入的字节 [] 因此，如果您只想读取 1000 个字节，可以尝试使用

int whereYouWantToStartInserting = 0;
int maximumOfBytes;
read(buffer,whereYouWantToStartInserting,maximumOfBytes);

希望这段代码对你有所帮助。

Answer 3

这有点原始，但它应该可以满足您的要求（虽然没有测试）。

int bytesReadSinceSleep = 0;
long lastSleepTime = System.currentTimeMillis();

while ((bytesRead = inputStream.read(buffer)) != -1) {
    outputStream.write(buffer, 0, bytesRead);

    // increment bytes read this second
    bytesReadSinceSleep += bytesRead;

    // if we've passed the throttling point, sleep for the remainder of the second
    if (bytesReadSinceSleep >= MAX_BYTES_PER_SECOND) {

        // calculate time elapsed since last sleep
        long timeElapsed = System.currentTimeMillis() - lastSleepTime;

        // sleep for the remainder of 1 second (if there is a remainder)
        Thread.sleep(Math.max(1000 - timeElapsed, 0));

        // reset byte count
        bytesReadSinceSleep = 0;

        // reset sleep time
        lastSleepTime = System.currentTimeMillis();
    }
}

如果BUFFER_SIZE 不是MAX_BYTES_PER_SECOND 的一个因子，并且您对节流速率很特别，您可能希望使用接受偏移和限制参数的read() 重载来准确读取正确的量。

---

为了更清洁的解决方案，您可以使用 Guava 的RateLimiter：

RateLimiter limiter = RateLimiter.create(MAX_BYTES_PER_SECOND);
while ((bytesRead = inputStream.read(buffer)) != -1) {
    outputStream.write(buffer, 0, bytesRead);
    limiter.acquire(bytesRead);
}