如何跟踪databufferutils.write（）的进度和速度？

Question

Flux<DataBuffer> dataBufferFlux = webClientEmbed.get()
  .uri(someUri)
  // ...
  .retrieve()
  .bodyToFlux(DataBuffer.class);


DataBufferUtils
  .write(dataBufferFlux, somePath, CREATE)
  .block();

使用DataBuffer类时，跟踪WebClient下载进度的正确方法是什么？

Answer 1

下面是一个代码片段，说明如何跟踪WebClient和DataBufferUtils的进度

    public DownloadProgress download(String url, String targetFile) {

        WebClient client = WebClient.builder()
                .clientConnector(
                    new ReactorClientHttpConnector(HttpClient.create().followRedirect(true)
                 ))
                .baseUrl(url)
                .build();
        
        final DownloadProgress progress = new DownloadProgress();

        Flux<DataBuffer> dataBufferFlux = client.get()
            .exchangeToFlux( response -> {
                long contentLength = response.headers().contentLength().orElse(-1);
                progress.setLength(contentLength);
                return response.bodyToFlux(DataBuffer.class);
            });
        
        Path path = Paths.get(targetFile);
        FileOutputStream fout = null;
        try {
            fout = new FileOutputStream(path.toFile());
        } catch (FileNotFoundException e) {
            throw new RuntimeException("Unable to create targetFile", e);
        }
        
        DataBufferUtils.write(dataBufferFlux, progress.getOutputStream(fout) )
            .subscribe( DataBufferUtils.releaseConsumer() );
        
        return progress;
    }

此方法返回一个在写入outpout流时更新的“DownloadProgress”对象。下面是这个DownloadProgress类：

public class DownloadProgress {

    private long contentLength = -1;
    private long downloaded;

    public void setLength(long contentLength) {
        this.contentLength = contentLength;
    }
    
    public OutputStream getOutputStream(FileOutputStream fileOutputStream) {
        return new FilterOutputStream(fileOutputStream) {
            @Override
            public void write(byte[] b, int off, int len) throws IOException {
                out.write(b, off, len);
                downloaded += len;
            }
            @Override
            public void write(int b) throws IOException {
                out.write(b);
                downloaded++;
            }
            @Override
            public void close() throws IOException {
                super.close();
                done();
            }
        };
    }
    
    public void done() {
        if ( contentLength == -1 ) {
            contentLength = downloaded;
        } else {
            downloaded = contentLength;
        }
    }
    
    public double getProgress() {
        if ( contentLength == -1 ) return 0;
        return downloaded / (double) contentLength;
    }
    
    public long getDownloaded() {
        return downloaded;
    }

    public boolean finished() {
        return downloaded > 0 && downloaded == contentLength;
    }
    
}

然后，为了监视下载，您可以在finished返回false时循环并显示进度。

            do {
                String progress = String.format("%.2f", dl.getProgress() * 100.0);
                log.info( "Downloaded: {}%", progress );
                try {
                    Thread.sleep(250);
                } catch (InterruptedException e) {
                    return;
                }
            } while(!dl.finished());