Jersey 客户端上传进度

Question

我有一个Jersey客户端，需要上传一份足够大的文件，需要使用进度条来显示进度。

问题在于，对于需要几分钟才能完成的上传，我看到传输的字节数已经达到了100%，但这是在应用程序启动后立即发生的。然后要花费几分钟才能打印出“完成”字符串。
就好像字节被发送到了缓冲区，而我正在读取传输到缓冲区的速度，而不是实际的上传速度。这使得进度条没有用处。

以下是非常简单的代码：

ClientConfig config = new DefaultClientConfig();
Client client = Client.create(config);
WebResource resource = client.resource("www.myrestserver.com/uploads");
WebResource.Builder builder = resource.type(MediaType.MULTIPART_FORM_DATA_TYPE);

FormDataMultiPart multiPart = new FormDataMultiPart();
FileDataBodyPart fdbp = new FileDataBodyPart("data.zip", new File("data.zip"));
BodyPart bp = multiPart.bodyPart(fdbp);
String response = builder.post(String.class, multiPart);

为了获取进度状态，我在调用builder.post之前添加了一个ContainerListener过滤器。

final ContainerListener containerListener = new ContainerListener() {

        @Override
        public void onSent(long delta, long bytes) {
            System.out.println(delta + " : " + long);
        }

        @Override
        public void onFinish() {
            super.onFinish();
            System.out.println("on finish");
        }

    };

    OnStartConnectionListener connectionListenerFactory = new OnStartConnectionListener() {
        @Override
        public ContainerListener onStart(ClientRequest cr) {
            return containerListener;
        }

    };

    resource.addFilter(new ConnectionListenerFilter(connectionListenerFactory));

- AgostinoX

Answer 1

在Jersey 2.X中，我使用了一个WriterInterceptor来包装输出流，该输出流是Apache Commons IO CountingOutputStream的子类，可以跟踪写入并通知我的上传进度代码（未显示）。

public class UploadMonitorInterceptor implements WriterInterceptor {

    @Override
    public void aroundWriteTo(WriterInterceptorContext context) throws IOException, WebApplicationException {

        // the original outputstream jersey writes with
        final OutputStream os = context.getOutputStream();

        // you can use Jersey's target/builder properties or 
        // special headers to set identifiers of the source of the stream
        // and other info needed for progress monitoring
        String id = (String) context.getProperty("id");
        long fileSize = (long) context.getProperty("fileSize");

        // subclass of counting stream which will notify my progress
        // indicators.
        context.setOutputStream(new MyCountingOutputStream(os, id, fileSize));

        // proceed with any other interceptors
        context.proceed();
    }

}

然后我将这个拦截器注册到客户端，或者你希望使用拦截器的特定目标上。

- David

Answer 2

你可以自己提供一个MessageBodyWriter来处理java.io.File类型的数据，这个处理过程中可以触发一些事件或通知一些监听器来展示进度变化。

@Provider()
@Produces(MediaType.APPLICATION_OCTET_STREAM)
public class MyFileProvider implements MessageBodyWriter<File> {

    public boolean isWriteable(Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) {
        return File.class.isAssignableFrom(type);
    }

    public void writeTo(File t, Class<?> type, Type genericType, Annotation annotations[], MediaType mediaType, MultivaluedMap<String, Object> httpHeaders, OutputStream entityStream) throws IOException {
        InputStream in = new FileInputStream(t);
        try {
            int read;
            final byte[] data = new byte[ReaderWriter.BUFFER_SIZE];
            while ((read = in.read(data)) != -1) {
                entityStream.write(data, 0, read);
                // fire some event as progress changes
            }
        } finally {
            in.close();
        }
    }

    @Override
    public long getSize(File t, Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) {
        return t.length();
    }
}

为了让您的客户端应用程序使用这个新提供者，只需：

ClientConfig config = new DefaultClientConfig();
config.getClasses().add(MyFileProvider.class);

或者

ClientConfig config = new DefaultClientConfig();
MyFileProvider myProvider = new MyFileProvider ();
cc.getSingletons().add(myProvider);

当接收到进度事件时，您还需要包括一些算法来识别传输的文件。

编辑后：

我刚刚发现，默认情况下HTTPUrlConnection使用缓冲。要禁用缓冲，您可以做以下几件事：

1. httpUrlConnection.setChunkedStreamingMode(chunklength) - 禁用缓冲并使用分块传输编码发送请求 2. httpUrlConnection.setFixedLengthStreamingMode(contentLength) - 禁用缓冲但添加了一些流的限制：必须发送确切数量的字节

因此，我建议您的问题的最终解决方案使用第一个选项，看起来像这样：

ClientConfig config = new DefaultClientConfig();
config.getClasses().add(MyFileProvider.class);
URLConnectionClientHandler clientHandler = new URLConnectionClientHandler(new HttpURLConnectionFactory() {
     @Override
     public HttpURLConnection getHttpURLConnection(URL url) throws IOException {
           HttpURLConnection connection = (HttpURLConnection) url.openConnection();
                connection.setChunkedStreamingMode(1024);
                return connection;
            }
});
Client client = new Client(clientHandler, config);

- Tomasz Krzyżak

---

谢谢Tomasz，这个答案非常好。你提供了两种配置客户端的方式，真是令人钦佩和易于理解。不幸的是，问题仍然存在。我在entityStream.write之后加了一个System.out.println(...)，但结果是我在一瞬间写入了大文件（>10MB），然后它就卡住了，等待“真正”的上传发生。这个解决方案也出现了这种情况，这意味着问题出在其他地方。对于你的回答，我不能接受它，但我可以开始另一个具体的问题，在那里我很高兴将其标记为正确。 :-) - AgostinoX

我也尝试在entityStream.write(...)之后添加entityStream.flush()，以便强制将数据实际写入套接字而不仅仅是写入缓冲区。但结果相同 :-( - AgostinoX

好的，太棒了，它起作用了。使用监听器和自定义文件提供程序两种方式都可以。也许应该强调解决方案是第二部分，将其移至顶部可能更好。文件提供程序作为监听器的替代方案很有趣。此外，它有助于澄清Jersey架构，所以我会保留它，但不作为问题的直接答案。 - AgostinoX

Answer 3

我已经成功地使用了David的答案。然而，我想在此基础上进行扩展：

下面是我的WriterInterceptor的aroundWriteTo实现，它展示了如何将一个面板（或类似物）也传递给CountingOutputStream：

@Override
public void aroundWriteTo(WriterInterceptorContext context)
    throws IOException, WebApplicationException
{
  final OutputStream outputStream = context.getOutputStream();

  long fileSize = (long) context.getProperty(FILE_SIZE_PROPERTY_NAME);

  context.setOutputStream(new ProgressFileUploadStream(outputStream, fileSize,
      (progressPanel) context
          .getProperty(PROGRESS_PANEL_PROPERTY_NAME)));

  context.proceed();
}

< p > CountingOutputStream 的 afterWrite 方法可以设置进度:

@Override
protected void afterWrite(int n)
{
  double percent = ((double) getByteCount() / fileSize);
  progressPanel.setValue((int) (percent * 100));
}

属性可以在Invocation.Builder对象上设置：

Invocation.Builder invocationBuilder = webTarget.request();
invocationBuilder.property(
    UploadMonitorInterceptor.FILE_SIZE_PROPERTY_NAME, newFile.length());
invocationBuilder.property(
    UploadMonitorInterceptor.PROGRESS_PANEL_PROPERTY_NAME,      
    progressPanel);

也许对于David的回答来说最重要的补充，也是我决定发布自己的原因，是以下代码：

client.property(ClientProperties.CHUNKED_ENCODING_SIZE, 1024);
client.property(ClientProperties.REQUEST_ENTITY_PROCESSING, "CHUNKED");

客户端对象是 javax.ws.rs.client.Client。

使用 WriterInterceptor 方法时，禁用缓冲也是必不可少的。上述代码是使用 Jersey 2.x 实现此目的的一种简单方法。

- Thomas Kainrad